"""Security-focused tests for staging_server.py (PR2 + QB6 hardening).

Covers:
  S2 — _check_host: allow/deny by Host header
  S3 — _cors_headers: reflects allowlisted origin / omits for others / never *
  S7 — do_HEAD responds 405
  S8 — _resolve_course_dir rejects symlinked course dirs outside courses_root
  S4 — unified_staging_review output sandboxes untrusted HTML
  QB6 — per-process synchronizer token (ASUCB_STAGING_TOKEN / X-ASUCB-Staging-Token)
"""

import io
import json
import pathlib
import sys
import types

import pytest

SCRIPTS = pathlib.Path(__file__).resolve().parents[1] / "scripts"
sys.path.insert(0, str(SCRIPTS))

import staging_server


# ---------------------------------------------------------------------------
# Shared harness
# ---------------------------------------------------------------------------

class MockServer:
    """Minimal stand-in for HTTPServer so handlers can read server_address."""

    def __init__(self, port=8111):
        self.server_address = ("127.0.0.1", port)


class SecurityHarness(staging_server.StagingHandler):
    """Test harness that supplies a mock server object and records headers."""

    def __init__(self, method, path, body=None, headers=None, port=8111):
        payload = b""
        req_headers = dict(headers or {})
        if body is not None:
            payload = json.dumps(body).encode("utf-8")
            req_headers.setdefault("Content-Type", "application/json")
            req_headers.setdefault("Content-Length", str(len(payload)))
        self.command = method
        self.path = path
        self.headers = req_headers
        self.rfile = io.BytesIO(payload)
        self.wfile = io.BytesIO()
        self.server = MockServer(port=port)
        self.status = None
        self.response_headers = []

    def send_response(self, code, message=None):
        self.status = code

    def send_header(self, keyword, value):
        self.response_headers.append((keyword, value))

    def end_headers(self):
        # Override: call _cors_headers so we can inspect them, but DON'T call
        # super().end_headers() (which would try to write to a real socket).
        self._cors_headers()

    def get_header(self, name):
        """Return all values for the given response header (case-insensitive)."""
        name_lower = name.lower()
        return [v for k, v in self.response_headers if k.lower() == name_lower]


def make_request(method, path, headers=None, body=None, port=8111):
    h = SecurityHarness(method, path, body=body, headers=headers, port=port)
    getattr(h, f"do_{method}")()
    return h


# ---------------------------------------------------------------------------
# S2 — _check_host
# ---------------------------------------------------------------------------

class TestCheckHost:
    def test_missing_host_returns_false_and_403(self):
        h = SecurityHarness("GET", "/", headers={})
        result = h._check_host()
        assert result is False
        assert h.status == 403

    def test_evil_host_returns_false_and_403(self):
        h = SecurityHarness("GET", "/", headers={"Host": "evil.com"})
        result = h._check_host()
        assert result is False
        assert h.status == 403

    def test_localhost_host_allowed(self):
        h = SecurityHarness("GET", "/", headers={"Host": "localhost:8111"}, port=8111)
        result = h._check_host()
        assert result is True
        assert h.status is None  # no 403 sent

    def test_127_host_allowed(self):
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:8111"}, port=8111)
        result = h._check_host()
        assert result is True
        assert h.status is None

    def test_non_default_port_respected(self):
        """With --port 9000, Host: 127.0.0.1:9000 should be accepted."""
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:9000"}, port=9000)
        assert h._check_host() is True

    def test_non_default_port_rejects_wrong_port(self):
        """With --port 9000, Host: 127.0.0.1:8111 should be rejected."""
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:8111"}, port=9000)
        result = h._check_host()
        assert result is False
        assert h.status == 403

    def test_evil_host_on_get_returns_403(self, tmp_path, monkeypatch):
        """do_GET with evil Host → 403 before any routing."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("GET", "/", headers={"Host": "evil.com"})
        assert h.status == 403

    def test_evil_host_on_put_returns_403(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers={"Host": "evil.com", "Origin": "http://localhost:8111"},
            body={"html": "<p>x</p>"},
        )
        assert h.status == 403

    def test_evil_host_on_delete_returns_403(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request(
            "DELETE",
            "/api/staging/course1/pages/lesson",
            headers={"Host": "evil.com", "Origin": "http://localhost:8111"},
        )
        assert h.status == 403

    def test_evil_host_on_options_returns_403(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("OPTIONS", "/", headers={"Host": "evil.com"})
        assert h.status == 403

    def test_evil_host_on_head_returns_403(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("HEAD", "/.env", headers={"Host": "evil.com"})
        assert h.status == 403


# ---------------------------------------------------------------------------
# S3 — _cors_headers: no wildcard, reflects allowlisted origin only
# ---------------------------------------------------------------------------

class TestCorsHeaders:
    def test_allowed_origin_reflected(self):
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:8111", "Origin": "http://127.0.0.1:8111"}, port=8111)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == ["http://127.0.0.1:8111"]

    def test_unknown_origin_not_reflected(self):
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:8111", "Origin": "https://evil.example"}, port=8111)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == []

    def test_no_origin_header_no_acao(self):
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:8111"}, port=8111)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == []

    def test_wildcard_never_emitted(self):
        for origin in ["https://evil.example", None]:
            headers = {"Host": "127.0.0.1:8111"}
            if origin:
                headers["Origin"] = origin
            h = SecurityHarness("GET", "/", headers=headers, port=8111)
            h._cors_headers()
            acao = h.get_header("Access-Control-Allow-Origin")
            assert "*" not in acao, f"Wildcard emitted for origin={origin!r}"

    def test_localhost_origin_reflected(self):
        h = SecurityHarness("GET", "/", headers={"Host": "localhost:8111", "Origin": "http://localhost:8111"}, port=8111)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == ["http://localhost:8111"]

    def test_non_default_port_cors(self):
        """With --port 9000, http://127.0.0.1:9000 is the allowed origin."""
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:9000", "Origin": "http://127.0.0.1:9000"}, port=9000)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == ["http://127.0.0.1:9000"]

    def test_non_default_port_wrong_origin_rejected(self):
        """With --port 9000, http://127.0.0.1:8111 is NOT allowed."""
        h = SecurityHarness("GET", "/", headers={"Host": "127.0.0.1:9000", "Origin": "http://127.0.0.1:8111"}, port=9000)
        h._cors_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        assert acao == []

    def test_cors_emitted_exactly_once_per_response(self, tmp_path, monkeypatch):
        """end_headers() must be the single emission point; no duplicates."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = SecurityHarness(
            "GET", "/",
            headers={"Host": "127.0.0.1:8111", "Origin": "http://127.0.0.1:8111"},
            port=8111,
        )
        # Call end_headers (our override) which triggers _cors_headers once.
        h.end_headers()
        acao = h.get_header("Access-Control-Allow-Origin")
        # Should appear exactly once (not 0, not 2+).
        assert len(acao) == 1
        assert acao[0] == "http://127.0.0.1:8111"


# ---------------------------------------------------------------------------
# S7 — do_HEAD responds 405
# ---------------------------------------------------------------------------

class TestDoHead:
    def test_head_env_returns_405(self, tmp_path, monkeypatch):
        """HEAD /.env must not return 200 with Content-Length."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("HEAD", "/.env", headers={"Host": "127.0.0.1:8111"})
        assert h.status == 405

    def test_head_any_path_returns_405(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("HEAD", "/courses/course1/review", headers={"Host": "127.0.0.1:8111"})
        assert h.status == 405

    def test_head_root_returns_405(self, tmp_path, monkeypatch):
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        h = make_request("HEAD", "/", headers={"Host": "127.0.0.1:8111"})
        assert h.status == 405

    def test_head_does_not_call_super(self, tmp_path, monkeypatch):
        """Verify super().do_HEAD (file serving) is NOT invoked."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        called = []

        class PatchedHandler(staging_server.StagingHandler):
            def send_response(self, code, message=None):
                self.status = code

            def send_header(self, keyword, value):
                if not hasattr(self, "response_headers"):
                    self.response_headers = []
                self.response_headers.append((keyword, value))

            def end_headers(self):
                self._cors_headers()

        import http.server as _hs
        orig_send_head = _hs.SimpleHTTPRequestHandler.send_head

        def spy_send_head(self):
            called.append(True)
            return orig_send_head(self)

        monkeypatch.setattr(_hs.SimpleHTTPRequestHandler, "send_head", spy_send_head)

        h = SecurityHarness("HEAD", "/.env", headers={"Host": "127.0.0.1:8111"})
        h.do_HEAD()
        assert not called, "super().send_head() was called — file probing not blocked"


# ---------------------------------------------------------------------------
# S8 — _resolve_course_dir rejects external symlinks
# ---------------------------------------------------------------------------

class TestResolveCourseDir:
    def test_symlink_outside_courses_root_raises(self, tmp_path, monkeypatch):
        """A course directory that is a symlink pointing outside courses/ must raise."""
        courses_root = tmp_path / "courses"
        courses_root.mkdir()
        external_target = tmp_path / "external"
        external_target.mkdir()
        # Create a symlink inside courses/ pointing to the external dir.
        (courses_root / "course1").symlink_to(external_target)
        monkeypatch.setattr(staging_server, "COURSES_DIR", courses_root)

        with pytest.raises(ValueError):
            staging_server._resolve_course_dir("course1")

    def test_real_course_dir_inside_courses_root_allowed(self, tmp_path, monkeypatch):
        """A normal course directory inside courses/ must resolve fine."""
        courses_root = tmp_path / "courses"
        course_dir = courses_root / "course42"
        course_dir.mkdir(parents=True)
        monkeypatch.setattr(staging_server, "COURSES_DIR", courses_root)

        result = staging_server._resolve_course_dir("course42")
        assert result == course_dir.resolve()

    def test_traversal_raises(self, tmp_path, monkeypatch):
        """Path traversal via course_id must raise (existing behaviour)."""
        courses_root = tmp_path / "courses"
        courses_root.mkdir()
        monkeypatch.setattr(staging_server, "COURSES_DIR", courses_root)

        # COURSE_ID_RE won't match ".." directly, but _resolve_course_dir
        # alone (called without the regex check) should also be safe.
        # We test the full _resolve_safe which includes the regex check.
        with pytest.raises(ValueError):
            staging_server._resolve_safe("..", "pages", "lesson")


# ---------------------------------------------------------------------------
# S4 — unified_staging_review sandboxes untrusted HTML
# ---------------------------------------------------------------------------

class TestUnifiedStagingReviewSandbox:
    """Tests for the sandbox wrapper in unified_staging_review._sandbox_preview."""

    def setup_method(self):
        try:
            import unified_staging_review as _usr
            self.usr = _usr
        except ImportError:
            pytest.skip("unified_staging_review not importable")

    def test_sandbox_preview_wraps_in_iframe(self):
        out = self.usr._sandbox_preview("<p>hello</p>")
        assert "<iframe" in out
        assert 'sandbox' in out

    def test_sandbox_preview_has_no_allow_scripts(self):
        """sandbox attribute must be empty (no allow-scripts, no allow-same-origin)."""
        out = self.usr._sandbox_preview("<p>test</p>")
        # Ensure 'sandbox' is not followed by 'allow-scripts' or 'allow-same-origin'
        assert "allow-scripts" not in out
        assert "allow-same-origin" not in out

    def test_sandbox_preview_strips_script_tag(self):
        xss = "<script>document.title='XSS'</script><p>content</p>"
        out = self.usr._sandbox_preview(xss)
        # The script must be sanitized before going into srcdoc.
        # The srcdoc is attr-escaped so we check the decoded content.
        import html
        srcdoc_match = out
        assert "document.title" not in html.unescape(srcdoc_match)

    def test_sandbox_preview_strips_onerror(self):
        xss = '<img src=x onerror="fetch(\'/api/staging/1/pages/x\',{method:\'DELETE\'})">'
        out = self.usr._sandbox_preview(xss)
        import html
        decoded = html.unescape(out)
        assert "onerror" not in decoded
        assert "DELETE" not in decoded

    def test_sandbox_preview_escapes_srcdoc_attribute(self):
        """Verify that quote characters in content are escaped so srcdoc can't break out."""
        payload = '<p class="test">Don\'t "break" out</p>'
        out = self.usr._sandbox_preview(payload)
        # The iframe srcdoc=... attribute must be properly escaped.
        # If the attribute is broken, the HTML would have unescaped " in it.
        # We verify the iframe tag is well-formed by checking structure.
        assert 'srcdoc="' in out

    def test_sandbox_preview_content_visible(self):
        """Benign content must survive inside the sandboxed iframe srcdoc."""
        import html
        out = self.usr._sandbox_preview("<p>Hello world</p>")
        # Decode the srcdoc attribute value
        decoded = html.unescape(out)
        assert "Hello world" in decoded

    def test_generate_review_output_contains_sandbox_for_xss_payload(self, tmp_path, monkeypatch):
        """generate_review for a staged page with an XSS payload must NOT contain raw script."""
        import unified_staging_review as usr

        # Arrange: write a staged page with a script injection.
        xss_payload = '<script>document.title="XSS"</script><img src=x onerror=alert(1)><p>visible</p>'
        course_id = "course99"
        staging_dir = tmp_path / course_id / "staging" / "pages"
        staging_dir.mkdir(parents=True)
        (staging_dir / "lesson.html").write_text(xss_payload, encoding="utf-8")

        monkeypatch.setattr(usr, "COURSES_DIR", tmp_path)

        # Patch the Canvas module fetcher so it doesn't make network calls.
        monkeypatch.setattr(usr, "fetch_canvas_modules", lambda *a, **kw: [])
        # Patch course name loader so it doesn't read real configs.
        monkeypatch.setattr(usr, "load_course_name", lambda *a, **kw: "Test Course")

        try:
            output = usr.generate_review(course_id)
        except Exception:
            # If generate_review can't fully run (e.g. Canvas API missing), skip.
            pytest.skip("generate_review requires additional Canvas API context")

        # The raw script tag must NOT appear verbatim in the rendered HTML.
        assert "<script>document.title" not in output
        # The onerror attribute must not appear raw.
        assert "onerror=alert" not in output
        # But a sandbox attribute must be present.
        assert "sandbox" in output


# ---------------------------------------------------------------------------
# Editable-field integrity — regression for _sandbox_quiz_html removal
# ---------------------------------------------------------------------------

class TestQuizEditableFieldIntegrity:
    """Regression: editable quiz fields must be sanitized in place, NOT wrapped
    in a sandboxed iframe.

    Before PR2 fix: `_sandbox_quiz_html` wrapped question_text / description /
    feedback in a null-origin `<iframe srcdoc=...>`, which (a) made them
    uneditable and (b) caused the editor JS to serialize the literal iframe
    wrapper into the staged JSON on save — data corruption.

    After fix: `_sanitize_html` is used in place. XSS vectors are stripped;
    editable fields remain in the outer DOM so contenteditable + data-field
    serialization work correctly.
    """

    def _make_quiz(self):
        return {
            "questions": [
                {
                    "question_name": "Q1",
                    "question_text": '<p>Stem</p><img src=x onerror="alert(1)">',
                    "question_type": "multiple_choice_question",
                    "points_possible": 1,
                    "answers": [
                        {
                            "text": "A",
                            "weight": 100,
                            "answer_comments_html": '<em>note</em><img src=x onerror="alert(3)">',
                        },
                        {"text": "B", "weight": 0},
                    ],
                    "correct_comments_html": "<strong>Right</strong>",
                    "incorrect_comments_html": "<em>Wrong</em>",
                }
            ],
            "quiz": {
                "description": '<p>Desc</p><script>alert(2)</script>',
            },
        }

    def test_question_text_sanitized_not_iframed(self):
        """question_text: XSS stripped, benign content present, no iframe wrapper."""
        out = staging_server._render_quiz(
            self._make_quiz(), title="T", course_id="c1", obj_id="q1"
        )
        import re
        # Find the data-field="question_text" div and its content.
        m = re.search(
            r'data-field="question_text"[^>]*>(.*?)</div>',
            out, re.DOTALL
        )
        assert m, "question_text div not found in output"
        field_content = m.group(1)

        # (a) Sanitized text is present directly in the div.
        assert "<p>Stem</p>" in field_content
        # (b) No iframe wrapper inside the field.
        assert "<iframe" not in field_content, (
            "question_text must NOT be wrapped in an iframe — breaks editing"
        )
        # (c) XSS vector stripped.
        assert "onerror=" not in field_content

    def test_description_sanitized_not_iframed(self):
        """description: XSS stripped, benign content present, no iframe wrapper."""
        out = staging_server._render_quiz(
            self._make_quiz(), title="T", course_id="c1", obj_id="q1"
        )
        import re
        m = re.search(
            r'data-quiz-field-html="description"[^>]*>(.*?)</div>',
            out, re.DOTALL
        )
        assert m, "description field div not found in output"
        field_content = m.group(1)

        # (a) Sanitized text present.
        assert "Desc" in field_content
        # (b) No iframe wrapper.
        assert "<iframe" not in field_content, (
            "description field must NOT be wrapped in an iframe"
        )
        # (c) <script> stripped.
        assert "<script" not in field_content

    def test_answer_comments_html_sanitized_not_raw(self):
        """answer_comments_html: XSS stripped, benign tag survives, no raw onerror."""
        out = staging_server._render_quiz(
            self._make_quiz(), title="T", course_id="c1", obj_id="q1"
        )
        import re
        # Find all answer_comments_html divs.
        matches = re.findall(
            r'data-field="answer_comments_html"[^>]*>(.*?)</div>',
            out, re.DOTALL
        )
        assert matches, "answer_comments_html div(s) not found in output"

        # Check the one that had HTML content.
        html_comment = next(
            (m for m in matches if "note" in m or "<em>" in m), None
        )
        assert html_comment is not None, (
            "Expected a comment field containing 'note' or <em>"
        )
        # (a) Benign tag survives.
        assert "<em>note</em>" in html_comment
        # (b) No iframe wrapper.
        assert "<iframe" not in html_comment, (
            "answer_comments_html must NOT be wrapped in an iframe"
        )
        # (c) onerror stripped.
        assert "onerror=" not in html_comment

    def test_no_srcdoc_in_editable_quiz_fields(self):
        """Confirm srcdoc attribute does not appear anywhere in quiz render output.

        srcdoc is the fingerprint of the old sandboxed-iframe approach which
        broke the editor serialization loop. Its presence in _render_quiz output
        means the iframe-wrapper regression has returned.
        """
        out = staging_server._render_quiz(
            self._make_quiz(), title="T", course_id="c1", obj_id="q1"
        )
        assert "srcdoc" not in out, (
            "srcdoc found in _render_quiz output — sandboxed-iframe approach "
            "has returned and will corrupt editable-field serialization"
        )


# ---------------------------------------------------------------------------
# QB6 — Per-process synchronizer token (ASUCB_STAGING_TOKEN / X-ASUCB-Staging-Token)
# ---------------------------------------------------------------------------
# The token is now stored in os.environ["ASUCB_STAGING_TOKEN"] so that both the
# __main__ module identity (script run) and the lazy `staging_server` import
# identity (unified_staging_review) see the same value.  Tests use
# monkeypatch.setenv / monkeypatch.delenv for auto-cleanup.

_GOOD_HEADERS = {
    "Host": "127.0.0.1:8111",
    "Origin": "http://127.0.0.1:8111",
}

_GOOD_BODY = {"html": "<p>test</p>"}


def _set_token(monkeypatch, value):
    """Set ASUCB_STAGING_TOKEN in the environment and restore afterward.

    Pass None to simulate the 'no token set' (pre-main) state.
    """
    if value is None:
        monkeypatch.delenv("ASUCB_STAGING_TOKEN", raising=False)
    else:
        monkeypatch.setenv("ASUCB_STAGING_TOKEN", value)


class TestSessionTokenEnforcement:
    """PUT/DELETE with ASUCB_STAGING_TOKEN set must require the correct header."""

    def test_put_without_token_header_is_403(self, tmp_path, monkeypatch):
        """PUT with token set but missing X-ASUCB-Staging-Token → 403."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, "test-token-abc123")
        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers=_GOOD_HEADERS,
            body=_GOOD_BODY,
        )
        assert h.status == 403

    def test_put_with_wrong_token_is_403(self, tmp_path, monkeypatch):
        """PUT with wrong token value → 403."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, "correct-token")
        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers={**_GOOD_HEADERS, "X-ASUCB-Staging-Token": "wrong-token"},
            body=_GOOD_BODY,
        )
        assert h.status == 403

    def test_delete_without_token_header_is_403(self, tmp_path, monkeypatch):
        """DELETE with token set but missing X-ASUCB-Staging-Token → 403."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, "test-token-xyz")
        h = make_request(
            "DELETE",
            "/api/staging/course1/pages/lesson",
            headers=_GOOD_HEADERS,
        )
        assert h.status == 403

    def test_delete_with_wrong_token_is_403(self, tmp_path, monkeypatch):
        """DELETE with wrong token → 403."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, "the-real-token")
        h = make_request(
            "DELETE",
            "/api/staging/course1/pages/lesson",
            headers={**_GOOD_HEADERS, "X-ASUCB-Staging-Token": "not-the-real-token"},
        )
        assert h.status == 403

    def test_token_check_403_body_ok_false(self, tmp_path, monkeypatch):
        """403 from token check must include {ok: false} JSON body."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, "tok123")
        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers=_GOOD_HEADERS,
            body=_GOOD_BODY,
        )
        assert h.status == 403
        body = json.loads(h.wfile.getvalue())
        assert body.get("ok") is False
        assert "token" in body.get("error", "").lower()

    def test_put_with_correct_token_passes_token_check(self, tmp_path, monkeypatch):
        """PUT with correct token + valid Host + Origin reaches the handler (not 403 for token)."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        tok = "abc-correct-token"
        _set_token(monkeypatch, tok)

        # Ensure the staged file exists so the handler can write it.
        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "lesson.html").write_text("<p>orig</p>", encoding="utf-8")

        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers={**_GOOD_HEADERS, "X-ASUCB-Staging-Token": tok},
            body=_GOOD_BODY,
        )
        # Should NOT be 403 for token — it reaches the write handler (200 or other).
        assert h.status != 403

    def test_delete_with_correct_token_passes_token_check(self, tmp_path, monkeypatch):
        """DELETE with correct token + valid Host + Origin reaches the handler."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        tok = "xyz-correct-token"
        _set_token(monkeypatch, tok)

        # Ensure the staged file exists so the handler can delete it.
        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "lesson.html").write_text("<p>data</p>", encoding="utf-8")

        h = make_request(
            "DELETE",
            "/api/staging/course1/pages/lesson",
            headers={**_GOOD_HEADERS, "X-ASUCB-Staging-Token": tok},
        )
        assert h.status != 403


class TestSessionTokenNoneSkipsCheck:
    """When ASUCB_STAGING_TOKEN is unset (default / test mode), token check is skipped."""

    def test_put_without_token_passes_when_session_token_is_none(self, tmp_path, monkeypatch):
        """Existing handler tests without a token still work when ASUCB_STAGING_TOKEN is unset."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        # Ensure ASUCB_STAGING_TOKEN is absent (default state — main() not called).
        _set_token(monkeypatch, None)

        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "lesson.html").write_text("<p>orig</p>", encoding="utf-8")

        h = make_request(
            "PUT",
            "/api/staging/course1/pages/lesson",
            headers=_GOOD_HEADERS,
            body=_GOOD_BODY,
        )
        # Without a session token set, the check is a no-op → should reach the handler.
        assert h.status != 403

    def test_delete_without_token_passes_when_session_token_is_none(self, tmp_path, monkeypatch):
        """DELETE without token header passes when ASUCB_STAGING_TOKEN is unset."""
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        _set_token(monkeypatch, None)

        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "lesson.html").write_text("<p>data</p>", encoding="utf-8")

        h = make_request(
            "DELETE",
            "/api/staging/course1/pages/lesson",
            headers=_GOOD_HEADERS,
        )
        assert h.status != 403


class TestSessionTokenMetaTag:
    """The <meta name='asucb-staging-token'> tag is embedded in served pages."""

    @staticmethod
    def _seed_staged_page(courses_dir, course_id="course1"):
        """Write a minimal staged page so _build_index reaches the full HTML path."""
        staged_dir = courses_dir / course_id / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "m1-overview.html").write_text("<p>hello</p>", encoding="utf-8")

    def test_build_index_embeds_meta_when_token_set(self, tmp_path, monkeypatch):
        """_build_index output contains the <meta> tag when ASUCB_STAGING_TOKEN is set."""
        self._seed_staged_page(tmp_path)
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        monkeypatch.setenv("ASUCB_STAGING_TOKEN", "index-test-token")
        out = staging_server._build_index()
        assert '<meta name="asucb-staging-token"' in out
        assert 'content="index-test-token"' in out

    def test_build_index_omits_meta_when_token_none(self, tmp_path, monkeypatch):
        """_build_index output does NOT contain the <meta> tag when ASUCB_STAGING_TOKEN is unset."""
        self._seed_staged_page(tmp_path)
        monkeypatch.setattr(staging_server, "COURSES_DIR", tmp_path)
        monkeypatch.delenv("ASUCB_STAGING_TOKEN", raising=False)
        out = staging_server._build_index()
        # The <meta> HTML tag must be absent (a JS querySelector selector string is OK).
        assert '<meta name="asucb-staging-token"' not in out

    def test_render_quiz_embeds_meta_when_token_set(self, monkeypatch):
        """_render_quiz output contains the <meta> tag when ASUCB_STAGING_TOKEN is set."""
        monkeypatch.setenv("ASUCB_STAGING_TOKEN", "quiz-test-token")
        quiz_data = {"quiz": {"title": "Test"}, "questions": []}
        out = staging_server._render_quiz(quiz_data, title="Test", course_id="c1", obj_id="q1")
        assert '<meta name="asucb-staging-token"' in out
        assert 'content="quiz-test-token"' in out

    def test_render_quiz_omits_meta_when_token_none(self, monkeypatch):
        """_render_quiz output does NOT have the <meta> tag when ASUCB_STAGING_TOKEN is unset."""
        monkeypatch.delenv("ASUCB_STAGING_TOKEN", raising=False)
        quiz_data = {"quiz": {"title": "Test"}, "questions": []}
        out = staging_server._render_quiz(quiz_data, title="Test", course_id="c1", obj_id="q1")
        # The <meta> HTML tag must be absent (JS querySelector selectors are OK).
        assert '<meta name="asucb-staging-token"' not in out

    def test_generate_review_embeds_meta_when_token_set(self, tmp_path, monkeypatch):
        """generate_review output contains the token meta tag when ASUCB_STAGING_TOKEN is set."""
        import unified_staging_review as usr

        monkeypatch.setenv("ASUCB_STAGING_TOKEN", "review-test-token")
        monkeypatch.setattr(usr, "COURSES_DIR", tmp_path)
        monkeypatch.setattr(usr, "fetch_canvas_modules", lambda *a, **kw: [])
        monkeypatch.setattr(usr, "load_course_name", lambda *a, **kw: "Test Course")

        # Create minimal staged content so generate_review produces a full page.
        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "m1-overview.html").write_text("<p>hello</p>", encoding="utf-8")

        out = usr.generate_review("course1")
        assert '<meta name="asucb-staging-token"' in out
        assert 'content="review-test-token"' in out

    def test_generate_review_omits_meta_when_token_none(self, tmp_path, monkeypatch):
        """generate_review output does NOT have the <meta> tag when ASUCB_STAGING_TOKEN is unset."""
        import unified_staging_review as usr

        monkeypatch.delenv("ASUCB_STAGING_TOKEN", raising=False)
        monkeypatch.setattr(usr, "COURSES_DIR", tmp_path)
        monkeypatch.setattr(usr, "fetch_canvas_modules", lambda *a, **kw: [])
        monkeypatch.setattr(usr, "load_course_name", lambda *a, **kw: "Test Course")

        staged_dir = tmp_path / "course1" / "staging" / "pages"
        staged_dir.mkdir(parents=True)
        (staged_dir / "m1-overview.html").write_text("<p>hello</p>", encoding="utf-8")

        out = usr.generate_review("course1")
        # The <meta> HTML tag must be absent (JS querySelector selectors are OK).
        assert '<meta name="asucb-staging-token"' not in out


class TestGetSessionTokenEnvVar:
    """_get_session_token() reads from os.environ, not a module global.

    This is the regression test for the __main__ / 'staging_server' module-identity
    split that caused the review page to embed an empty token (QB6 bug).
    """

    def test_returns_none_when_env_var_absent(self, monkeypatch):
        """_get_session_token() returns None when ASUCB_STAGING_TOKEN is not set."""
        monkeypatch.delenv("ASUCB_STAGING_TOKEN", raising=False)
        assert staging_server._get_session_token() is None

    def test_returns_value_from_env_var(self, monkeypatch):
        """_get_session_token() returns the env-var value when ASUCB_STAGING_TOKEN is set."""
        monkeypatch.setenv("ASUCB_STAGING_TOKEN", "env-based-token-xyz")
        assert staging_server._get_session_token() == "env-based-token-xyz"

    def test_value_visible_via_lazy_import(self, monkeypatch):
        """The env-var value is visible to a lazy import of staging_server.

        This is the specific failure mode the fix addresses: when staging_server is run
        as __main__, unified_staging_review lazy-imports it as 'staging_server' — a
        different module object.  The env-var approach means both identities see the
        same token.
        """
        import importlib
        import sys

        monkeypatch.setenv("ASUCB_STAGING_TOKEN", "shared-across-identities")

        # Force a re-import under the 'staging_server' name so we simulate the lazy
        # import identity that unified_staging_review uses.
        saved = sys.modules.pop("staging_server", None)
        try:
            import staging_server as _ss2
            assert _ss2._get_session_token() == "shared-across-identities"
        finally:
            # Restore the original module object so other tests are unaffected.
            if saved is not None:
                sys.modules["staging_server"] = saved
            else:
                sys.modules.pop("staging_server", None)
