"""E16 S1 — debug sessions: Start → reproduce → Stop, the tag on every line, the screenshots.

Four parts:
  * the session's life: one open per user, a second Start joins it, a Stop ends it, the cap ends it
    on its own, the demo account cannot record, nobody reaches another user's session (DB lane);
  * the tag: a request naming the caller's own open session is logged at its level with every line
    tagged, while a header naming anything else is ignored (a real sign-in through `_session_user`);
  * the browser's batches: a session counts as consent to its level, and only its own events keep
    its tag (DB lane);
  * screenshots: judged by their bytes, capped in size and count, served back only to their owner.
"""

import logging
from datetime import datetime, timedelta, timezone

import pytest
from fastapi import BackgroundTasks
from sqlalchemy import select

from app import debugsession, logs
from app.models import DebugScreenshot, DebugSession, LogEvent, User
from app.routes import diagnostics as diagnostics_routes

_PW = "correct horse battery"
_PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 32
_JPEG = b"\xff\xd8\xff\xe0" + b"\x00" * 32
_WEBP = b"RIFF\x00\x00\x00\x00WEBPVP8 " + b"\x00" * 16


def _signed_in(client, make_user, email="rec@example.com", **kw) -> User:
    """A real sign-in, so the auth dependency itself reads `X-Debug-Session` (`as_user` bypasses
    it)."""
    user = make_user(email=email, password=_PW, **kw)
    r = client.post("/auth/password-login", json={"email": email, "password": _PW})
    assert r.status_code == 200, r.text
    return user


def _start(client, level="DEBUG") -> dict:
    r = client.post("/api/debug-sessions", json={"level": level})
    assert r.status_code == 200, r.text
    return r.json()["session"]


# --- the session's life ---------------------------------------------------------------------


def test_start_opens_a_capped_session_that_me_reports(db, client, make_user):
    _signed_in(client, make_user)
    session = _start(client, "TRACE")
    assert session["level"] == "TRACE" and session["ended_at"] is None
    started = datetime.fromisoformat(session["started_at"])
    until = datetime.fromisoformat(session["until"])
    assert until - started == timedelta(minutes=debugsession.MAX_MINUTES)
    assert client.get("/api/me").json()["debug_session"]["id"] == session["id"]


def test_a_second_start_joins_the_running_session(db, make_user, as_user):
    client = as_user(make_user())
    first = _start(client, "DEBUG")
    assert _start(client, "TRACE")["id"] == first["id"]
    assert db.scalar(select(DebugSession).where(DebugSession.id == first["id"])).level == "DEBUG"


def test_only_the_two_raised_levels_can_be_recorded(db, make_user, as_user):
    client = as_user(make_user())
    for bogus in ("INFO", "WARNING", "trace"):
        assert client.post("/api/debug-sessions", json={"level": bogus}).status_code == 422


def test_the_demo_account_cannot_record(db, make_user, as_user):
    r = as_user(make_user(is_demo=True)).post("/api/debug-sessions", json={"level": "DEBUG"})
    assert r.status_code == 403 and r.json()["detail"]["code"] == "demo"


def test_stop_ends_it_and_me_stops_reporting_it(db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    r = client.post(f"/api/debug-sessions/{sid}/stop")
    assert r.status_code == 200 and r.json()["session"]["ended_reason"] == "stopped"
    assert client.get("/api/me").json()["debug_session"] is None
    # A second Stop changes nothing.
    again = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
    assert again["ended_at"] == r.json()["session"]["ended_at"]


def _expire(db, sid: str) -> None:
    session = db.get(DebugSession, sid)
    session.started_at -= timedelta(hours=2)
    session.until = datetime.now(timezone.utc) - timedelta(minutes=1)
    db.commit()


def test_a_session_past_its_cap_is_over_before_anything_closes_it(db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client)["id"]
    _expire(db, sid)
    assert client.get("/api/me").json()["debug_session"] is None
    # A late Stop records that it ended on its own, at its cap.
    stopped = client.post(f"/api/debug-sessions/{sid}/stop").json()["session"]
    assert stopped["ended_reason"] == "expired"
    assert stopped["ended_at"] == stopped["until"]


def test_a_start_after_an_expired_session_closes_it_and_opens_a_new_one(db, make_user, as_user):
    client = as_user(make_user())
    old = _start(client)["id"]
    _expire(db, old)
    new = _start(client)["id"]
    assert new != old
    db.expire_all()
    assert db.get(DebugSession, old).ended_reason == "expired"


def test_nobody_reaches_another_users_session(db, make_user, as_user):
    owner, stranger = as_user(make_user()), as_user(make_user())
    sid = _start(owner)["id"]
    shot = owner.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=_PNG,
        headers={"content-type": "image/png"},
    ).json()["screenshot"]["id"]
    for method, path in (
        ("get", f"/api/debug-sessions/{sid}"),
        ("post", f"/api/debug-sessions/{sid}/stop"),
        ("get", f"/api/debug-sessions/{sid}/events"),
        ("get", f"/api/debug-sessions/{sid}/screenshots"),
        ("get", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
        ("delete", f"/api/debug-sessions/{sid}/screenshots/{shot}"),
    ):
        r = getattr(stranger, method)(path)
        assert r.status_code == 404, (method, path, r.status_code)
        assert r.json()["detail"] == {"code": "not_found", "reason": "debug_session"}
    r = stranger.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=_PNG,
        headers={"content-type": "image/png"},
    )
    assert r.status_code == 404
    assert stranger.get("/api/debug-sessions").json()["items"] == []
    db.expire_all()
    assert db.get(DebugSession, sid).ended_at is None


def test_the_list_is_the_users_own_newest_first(db, make_user, as_user):
    client = as_user(make_user())
    first = _start(client)["id"]
    client.post(f"/api/debug-sessions/{first}/stop")
    second = _start(client)["id"]
    assert [s["id"] for s in client.get("/api/debug-sessions").json()["items"]] == [second, first]


def test_the_hourly_job_closes_expired_sessions_and_drops_old_ones(db, make_user):
    user = make_user()
    now = datetime.now(timezone.utc)
    lapsed = DebugSession(
        id="b" * 16,
        user_id=user.id,
        level="DEBUG",
        started_at=now - timedelta(hours=2),
        until=now - timedelta(hours=1),
    )
    db.add(lapsed)
    ancient = DebugSession(
        id="c" * 16,
        user_id=user.id,
        level="DEBUG",
        started_at=now - timedelta(days=30),
        until=now - timedelta(days=30) + timedelta(hours=1),
        ended_at=now - timedelta(days=30),
        ended_reason="stopped",
    )
    db.add(ancient)
    db.flush()  # no ORM relationship orders the inserts: the session row first
    db.add(
        DebugScreenshot(session_id=ancient.id, created_at=now, mime="image/png", size=4, data=b"x")
    )
    db.commit()

    assert debugsession.sweep(db) == 1
    assert debugsession.gc(db, 7) == 1
    db.commit()
    db.expire_all()
    assert db.get(DebugSession, lapsed.id).ended_reason == "expired"
    assert db.get(DebugSession, lapsed.id).ended_at == lapsed.until
    assert db.get(DebugSession, ancient.id) is None
    assert db.scalars(select(DebugScreenshot)).all() == []


def test_an_erased_account_takes_its_sessions_with_it(db, make_user, as_user):
    from app.auth import purge_user

    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    client.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=_PNG,
        headers={"content-type": "image/png"},
    )
    purge_user(db, user, BackgroundTasks())
    assert db.get(DebugSession, sid) is None
    assert db.scalars(select(DebugScreenshot)).all() == []


# --- the tag --------------------------------------------------------------------------------


class _Capture(logging.Handler):
    def __init__(self):
        super().__init__(level=logging.NOTSET)
        self.records: list[logging.LogRecord] = []
        self.addFilter(logs.ContextFilter())
        self.addFilter(logs.LevelGate())

    def emit(self, record):
        self.records.append(record)


@pytest.fixture
def gated():
    """The process at INFO with a gated handler on the root, as test_diagnostics' fixture."""
    root = logging.getLogger()
    saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level)
    handler = _Capture()
    root.addHandler(handler)
    logs._elevations.clear()
    logs.set_level("INFO")
    try:
        yield handler
    finally:
        root.removeHandler(handler)
        logs._base_level, logs._root_applied = saved[0], saved[2]
        logs._elevations.clear()
        logs._elevations.update(saved[1])
        root.setLevel(saved[3])


def _request_lines(handler: _Capture) -> list[logging.LogRecord]:
    return [r for r in handler.records if getattr(r, "event", None) == "http.request"]


def test_a_request_naming_the_own_open_session_is_tagged_and_raised(gated, db, client, make_user):
    _signed_in(client, make_user)
    sid = _start(client, "DEBUG")["id"]
    gated.records.clear()
    client.get("/api/debug-sessions", headers={"x-debug-session": sid})
    (line,) = _request_lines(gated)
    assert getattr(line, "debug_session_id", None) == sid
    # Raised to the session's level: the line is kept for storage though the user's own is Alap.
    assert getattr(line, "elevated", False) is True


def test_a_header_naming_anything_else_is_ignored_not_refused(gated, db, client, make_user):
    other = make_user(email="other@example.com")
    foreign = debugsession.start(db, other, "TRACE").id
    db.commit()
    _signed_in(client, make_user)
    own = _start(client)["id"]
    client.post(f"/api/debug-sessions/{own}/stop")
    for value in (foreign, own, "not-a-session", "f" * 16, "A" * 16):
        gated.records.clear()
        # Not `/api/me`: a polled route logs at DEBUG, and an un-raised request would drop the very
        # line this checks.
        r = client.get("/api/debug-sessions", headers={"x-debug-session": value})
        assert r.status_code == 200, value
        (line,) = _request_lines(gated)
        assert getattr(line, "debug_session_id", None) is None, value
        assert getattr(line, "elevated", False) is False, value


def test_a_session_never_lowers_the_users_own_level():
    now = datetime.now(timezone.utc)
    user = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(hours=3))
    session = DebugSession(level="DEBUG", until=now + timedelta(minutes=30))
    assert debugsession.level_for(user, session) == ("TRACE", user.diagnostics_until)
    assert debugsession.level_for(User(), session) == ("DEBUG", session.until)
    assert debugsession.level_for(User(), None) == ("INFO", None)


# --- the browser's batches -------------------------------------------------------------------


def _event(**over) -> dict:
    return {
        "ts": datetime.now(timezone.utc).isoformat(),
        "level": "DEBUG",
        "event": "ui.click",
        "msg": "click",
        "tab_id": "a1b2c3",
        **over,
    }


@pytest.fixture
def batches():
    diagnostics_routes._batches.reset()
    yield
    diagnostics_routes._batches.reset()


def test_a_session_lets_its_own_events_in_at_its_level_and_tags_them(
    batches, db, make_user, as_user
):
    user = make_user()
    client = as_user(user)
    sid = _start(client, "DEBUG")["id"]
    other = debugsession.start(db, make_user(), "DEBUG").id
    db.commit()
    events = [
        _event(debug_session_id=sid),
        _event(debug_session_id=other),  # not this session's: kept, untagged
        _event(),  # captured before Start: kept, untagged
        _event(level="TRACE", debug_session_id=sid),  # under the session's level: dropped
    ]
    r = client.post(
        "/api/diagnostics/events", json={"events": events}, headers={"x-debug-session": sid}
    )
    assert r.json() == {"ok": True, "stored": 3}
    rows = db.scalars(select(LogEvent).order_by(LogEvent.id)).all()
    assert [row.debug_session_id for row in rows] == [sid, None, None]
    assert {row.user_id for row in rows} == {user.id}


def test_without_the_header_a_base_level_user_is_still_refused(batches, db, make_user, as_user):
    client = as_user(make_user())
    sid = _start(client)["id"]
    r = client.post("/api/diagnostics/events", json={"events": [_event(debug_session_id=sid)]})
    assert r.status_code == 409 and r.json()["detail"]["code"] == "diagnostics_off"


def test_the_evidence_is_the_sessions_rows_in_time_order(db, make_user, as_user):
    user = make_user()
    client = as_user(user)
    sid = _start(client)["id"]
    now = datetime.now(timezone.utc)

    def row(ts, source, session, event="x.y", attrs=None):
        return LogEvent(
            ts=ts,
            level="INFO",
            sev=9,
            source=source,
            logger=source,
            event=event,
            msg=source,
            attrs=attrs,
            user_id=user.id,
            debug_session_id=session,
        )

    db.add_all(
        [
            row(now + timedelta(seconds=2), "api", sid),
            row(now + timedelta(seconds=1), "web", sid),
            row(now, "web", None),
            # The recording's own log upload is plumbing, not evidence…
            row(now, "api", sid, "http.request", {"path": debugsession.INGEST_PATH}),
            # …but a server line with no event code and no attrs is evidence all the same.
            row(now + timedelta(seconds=3), "api", sid, None),
        ]
    )
    db.commit()
    body = client.get(f"/api/debug-sessions/{sid}/events").json()
    assert [(i["source"], i["event"]) for i in body["items"]] == [
        ("web", "x.y"),
        ("api", "x.y"),
        ("api", None),
    ]
    assert body["total"] == 3


# --- screenshots ----------------------------------------------------------------------------


@pytest.fixture
def recording(db, make_user, as_user):
    client = as_user(make_user())
    return client, _start(client)["id"]


def _upload(client, sid, data, content_type="image/png"):
    return client.post(
        f"/api/debug-sessions/{sid}/screenshots",
        content=data,
        headers={"content-type": content_type},
    )


@pytest.mark.parametrize(
    ("data", "mime"), [(_PNG, "image/png"), (_JPEG, "image/jpeg"), (_WEBP, "image/webp")]
)
def test_a_screenshot_is_kept_as_what_its_bytes_are(recording, data, mime):
    client, sid = recording
    # The declared type is not trusted: every one of these claims to be a PNG.
    r = _upload(client, sid, data)
    assert r.status_code == 200 and r.json()["screenshot"]["mime"] == mime
    shot = r.json()["screenshot"]["id"]
    got = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert got.content == data
    assert got.headers["content-type"] == mime
    assert got.headers["x-content-type-options"] == "nosniff"


@pytest.mark.parametrize(
    ("data", "reason"), [(b"<svg onload=alert(1)>", "not_an_image"), (b"", "empty")]
)
def test_what_is_not_an_image_is_refused(recording, data, reason):
    client, sid = recording
    r = _upload(client, sid, data)
    assert r.status_code == 422
    assert r.json()["detail"] == {"code": "screenshot_refused", "reason": reason}


def test_screenshots_are_capped_in_size_and_count(recording, monkeypatch):
    client, sid = recording
    monkeypatch.setattr(debugsession, "SCREENSHOT_MAX_BYTES", len(_PNG) + 1)
    assert _upload(client, sid, _PNG + b"\x00\x00").status_code == 413
    for _ in range(debugsession.SCREENSHOTS_MAX):
        assert _upload(client, sid, _PNG).status_code == 200
    r = _upload(client, sid, _PNG)
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "too_many"
    assert (
        client.get(f"/api/debug-sessions/{sid}").json()["screenshots"]
        == debugsession.SCREENSHOTS_MAX
    )


def test_a_sent_session_takes_no_more_screenshots(db, recording):
    client, sid = recording
    db.get(DebugSession, sid).issue_url = "https://git.example/o/r/issues/1"
    db.commit()
    r = _upload(client, sid, _PNG)
    assert r.status_code == 422 and r.json()["detail"]["reason"] == "already_sent"


def test_a_screenshot_can_be_removed_and_is_then_gone(recording):
    client, sid = recording
    shot = _upload(client, sid, _PNG).json()["screenshot"]["id"]
    assert client.get(f"/api/debug-sessions/{sid}/screenshots").json()["items"][0]["id"] == shot
    r = client.delete(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.json() == {"ok": True, "deleted": shot}
    r = client.get(f"/api/debug-sessions/{sid}/screenshots/{shot}")
    assert r.status_code == 404 and r.json()["detail"]["reason"] == "screenshot"


def test_the_upload_path_left_out_of_the_evidence_is_the_real_one():
    assert debugsession.INGEST_PATH in {r.path for r in diagnostics_routes.router.routes}