"""周期性自我检查 — period math, opener idempotency, control eval, HTTP RBAC.

The test schema is built via ``create_all`` (no migrations), so the六个
seed categories from migration 322 are absent — every test seeds its own
categories, keeping them hermetic.
"""
from __future__ import annotations

import uuid
from datetime import date, datetime

import pytest
from httpx import AsyncClient

from app.domains.auditor import self_check_service as svc
from app.domains.auditor.models import (
    AuditorSelfCheckAttachment,
    AuditorSelfCheckCategory,
    AuditorSelfCheckRecord,
)

# ── period math ──────────────────────────────────────────────────────


def test_compute_current_period_all_cadences():
    assert svc.compute_current_period("quarterly", date(2026, 8, 15)) == (
        2026, "Q3", date(2026, 7, 1), date(2026, 9, 30),
    )
    assert svc.compute_current_period("semiannual", date(2026, 3, 1)) == (
        2026, "H1", date(2026, 1, 1), date(2026, 6, 30),
    )
    assert svc.compute_current_period("semiannual", date(2026, 11, 1)) == (
        2026, "H2", date(2026, 7, 1), date(2026, 12, 31),
    )
    assert svc.compute_current_period("annual", date(2026, 6, 22)) == (
        2026, "FY", date(2026, 1, 1), date(2026, 12, 31),
    )
    assert svc.compute_current_period("monthly", date(2026, 2, 9)) == (
        2026, "M02", date(2026, 2, 1), date(2026, 2, 28),
    )


def test_parse_period_bounds_roundtrip_and_validation():
    assert svc.parse_period_bounds("quarterly", 2026, "Q2") == (
        date(2026, 4, 1), date(2026, 6, 30),
    )
    with pytest.raises(Exception):
        svc.parse_period_bounds("quarterly", 2026, "Q9")
    with pytest.raises(Exception):
        svc.parse_period_bounds("semiannual", 2026, "Q1")


# ── helpers ──────────────────────────────────────────────────────────


async def _seed_category(db, *, key="cat_a", periodicity="quarterly",
                         requires_attachment=True):
    cat = AuditorSelfCheckCategory(
        key=key, name=f"类目 {key}", periodicity=periodicity,
        check_item="检查项", requires_attachment=requires_attachment,
        enabled=True, sort_order=0,
    )
    db.add(cat)
    await db.flush()
    return cat


# ── opener idempotency ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_open_due_records_idempotent(db_session):
    await _seed_category(db_session, key="open_test")
    await db_session.commit()

    r1 = await svc.open_due_records(db_session)
    await db_session.commit()
    assert r1["opened_count"] == 1

    r2 = await svc.open_due_records(db_session)
    await db_session.commit()
    assert r2["opened_count"] == 0
    assert r2["skipped"] == 1


# ── control evaluation ───────────────────────────────────────────────


@pytest.mark.asyncio
async def test_control_not_configured_when_no_categories(db_session):
    ev = await svc.evaluate_self_check_control(db_session)
    assert ev["status"] == "not_configured"


@pytest.mark.asyncio
async def test_control_fail_when_overdue(db_session):
    cat = await _seed_category(db_session, key="overdue_test")
    # A record whose deadline is in the past and still pending → overdue.
    rec = AuditorSelfCheckRecord(
        category_id=cat.id,
        period_year=2025,
        period_label="Q1",
        period_start=date(2025, 1, 1),
        period_end=date(2025, 3, 31),
        due_at=datetime(2025, 3, 31, 23, 59, 59),
        status="pending",
        check_item_snapshot="检查项",
    )
    db_session.add(rec)
    await db_session.commit()

    ev = await svc.evaluate_self_check_control(db_session)
    assert ev["status"] == "fail"
    assert ev["detail"]["overdue"] == 1


# ── HTTP RBAC ────────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_categories_rbac(client: AsyncClient, auditor_headers, employee_headers):
    # auditor can read
    ok = await client.get(
        "/api/v1/auditor/self-checks/categories", headers=auditor_headers,
    )
    assert ok.status_code == 200, ok.text

    # employee is forbidden
    denied = await client.get(
        "/api/v1/auditor/self-checks/categories", headers=employee_headers,
    )
    assert denied.status_code == 403

    # auditor can create
    created = await client.post(
        "/api/v1/auditor/self-checks/categories",
        headers=auditor_headers,
        json={
            "key": "http_cat",
            "name": "HTTP 类目",
            "periodicity": "quarterly",
            "check_item": "检查项",
            "requires_attachment": False,
        },
    )
    assert created.status_code == 200, created.text

    # employee cannot create (valid body so the RBAC gate, not pydantic,
    # is what rejects it)
    denied_create = await client.post(
        "/api/v1/auditor/self-checks/categories",
        headers=employee_headers,
        json={
            "key": "emp_denied",
            "name": "禁止",
            "periodicity": "quarterly",
            "check_item": "检查项",
        },
    )
    assert denied_create.status_code == 403


@pytest.mark.asyncio
async def test_record_lifecycle_and_attachment_gate(
    client: AsyncClient, auditor_headers, db_session, monkeypatch,
):
    # Registration HEADs the object (#2152), so stub the S3 probe — the
    # point of this test is the completion gate, not storage.
    async def _fake_stat(_db, *, object_key, **_kw):
        return {"size_bytes": 1000, "content_type": "image/png"}

    monkeypatch.setattr(
        "app.domains.auditor.self_check_service.stat_object", _fake_stat,
    )
    claimed_resource_ids = []

    async def _fake_claim(_db, *, object_keys, **_kw):
        from types import SimpleNamespace

        claimed_resource_ids.append(_kw["resource_id"])
        return {
            key: SimpleNamespace(
                declared_size_bytes=1000, content_type="image/jpeg",
            )
            for key in object_keys
        }
    monkeypatch.setattr("app.core.storage_uploads.mark_uploads_referenced", _fake_claim)
    # category that REQUIRES attachment
    cat = await _seed_category(db_session, key="gate_cat", requires_attachment=True)
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()

    listed = await client.get(
        f"/api/v1/auditor/self-checks/records?category_id={cat.id}",
        headers=auditor_headers,
    )
    assert listed.status_code == 200, listed.text
    items = listed.json()["data"]["items"]
    assert len(items) == 1
    rec_id = items[0]["id"]

    # completing without evidence is blocked
    blocked = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/complete",
        headers=auditor_headers,
        json={"outcome": "normal", "remark": "done"},
    )
    assert blocked.status_code == 422 or blocked.status_code == 400

    # register evidence under the right prefix, then complete succeeds
    obj_key = f"dev/self_check_evidence/2026/06/22/{rec_id}/abc-shot.png"
    reg = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": obj_key, "filename": "shot.png",
            "content_type": "image/png", "size_bytes": 1000,
        },
    )
    assert reg.status_code == 200, reg.text
    attachment_id = reg.json()["data"]["id"]
    attachment = await db_session.get(AuditorSelfCheckAttachment, uuid.UUID(attachment_id))
    assert attachment is not None
    assert attachment.content_type == "image/jpeg"
    assert claimed_resource_ids == [attachment_id]

    done = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/complete",
        headers=auditor_headers,
        json={"outcome": "normal", "remark": "checked, no anomaly"},
    )
    assert done.status_code == 200, done.text
    assert done.json()["data"]["status"] == "completed"

    # Re-completing an already-signed-off record must be rejected — a
    # second auditor cannot silently overwrite the prior sign-off without
    # an explicit reopen first.
    twice = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/complete",
        headers=auditor_headers,
        json={"outcome": "issue_found", "remark": "tamper"},
    )
    assert twice.status_code in (400, 422), twice.text

    # After reopening it returns to pending and can be completed again.
    reopened = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/reopen",
        headers=auditor_headers,
    )
    assert reopened.status_code == 200, reopened.text
    again = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/complete",
        headers=auditor_headers,
        json={"outcome": "normal", "remark": "re-checked"},
    )
    assert again.status_code == 200, again.text


@pytest.mark.asyncio
async def test_attachment_key_rejects_foreign_or_traversal(
    client: AsyncClient, auditor_headers, db_session,
):
    """object_key validation is segment-exact: a key outside the evidence
    namespace, or one smuggling a ``..`` traversal, must be rejected."""
    cat = await _seed_category(db_session, key="key_guard_cat")
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()
    listed = await client.get(
        f"/api/v1/auditor/self-checks/records?category_id={cat.id}",
        headers=auditor_headers,
    )
    rec_id = listed.json()["data"]["items"][0]["id"]

    # Foreign namespace (no self_check_evidence segment) → rejected.
    bad_ns = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": f"dev/approval_supplements/{rec_id}/secret.pdf",
            "filename": "x.pdf", "content_type": "application/pdf",
            "size_bytes": 10,
        },
    )
    assert bad_ns.status_code in (400, 422), bad_ns.text

    # Traversal segment → rejected even though both required substrings
    # are present.
    traversal = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": (
                f"dev/self_check_evidence/{rec_id}/../../other/secret.pdf"
            ),
            "filename": "x.pdf", "content_type": "application/pdf",
            "size_bytes": 10,
        },
    )
    assert traversal.status_code in (400, 422), traversal.text

    # Wrong record id (belongs to a different record) → rejected.
    other_rec = uuid.uuid4()
    wrong_rec = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": (
                f"dev/self_check_evidence/2026/06/22/{other_rec}/x.png"
            ),
            "filename": "x.png", "content_type": "image/png",
            "size_bytes": 10,
        },
    )
    assert wrong_rec.status_code in (400, 422), wrong_rec.text


@pytest.mark.asyncio
async def test_overview_and_export(
    client: AsyncClient, auditor_headers, db_session,
):
    await _seed_category(db_session, key="ov_cat")
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()

    ov = await client.get(
        "/api/v1/auditor/self-checks/overview", headers=auditor_headers,
    )
    assert ov.status_code == 200, ov.text
    assert ov.json()["data"]["summary"]["category_count"] >= 1

    exp = await client.get(
        "/api/v1/auditor/self-checks/records/export", headers=auditor_headers,
    )
    assert exp.status_code == 200, exp.text
    assert exp.headers["content-type"].startswith(
        "application/vnd.openxmlformats",
    )


@pytest.mark.asyncio
async def test_attachment_bad_key_rejected(
    client: AsyncClient, auditor_headers, db_session,
):
    cat = await _seed_category(db_session, key="badkey_cat", requires_attachment=False)
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()
    rec = (await db_session.execute(
        AuditorSelfCheckRecord.__table__.select().where(
            AuditorSelfCheckRecord.category_id == cat.id,
        )
    )).first()
    rec_id = rec.id

    bad = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": "dev/other_prefix/x.png", "filename": "x.png",
            "content_type": "image/png", "size_bytes": 10,
        },
    )
    assert bad.status_code in (400, 422)


@pytest.mark.asyncio
async def test_attachment_download_writes_an_audit_row(
    client: AsyncClient, auditor_headers, db_session, monkeypatch,
):
    """Minting a presigned GET is data egress and must be traceable.

    The URL bypasses Helix auth for its whole lifetime and points at raw
    evidence, so "who pulled this file" has to be answerable — the xlsx
    export of the same record has always left a ``*.export`` row, while
    this endpoint left nothing at all (#2161).
    """
    from sqlalchemy import select

    from app.audit.models import AuditLog
    from app.core import storage as storage_mod

    cat = await _seed_category(
        db_session, key="dl_audit_cat", requires_attachment=False,
    )
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()

    listed = await client.get(
        f"/api/v1/auditor/self-checks/records?category_id={cat.id}",
        headers=auditor_headers,
    )
    rec_id = listed.json()["data"]["items"][0]["id"]

    async def _fake_stat(_db, *, object_key, **_kw):
        return {"size_bytes": 2048, "content_type": "application/pdf"}

    monkeypatch.setattr(
        "app.domains.auditor.self_check_service.stat_object", _fake_stat,
    )
    async def _fake_claim(_db, *, object_keys, **_kw):
        from types import SimpleNamespace
        return {
            key: SimpleNamespace(
                declared_size_bytes=2048, content_type="application/pdf",
            )
            for key in object_keys
        }
    monkeypatch.setattr("app.core.storage_uploads.mark_uploads_referenced", _fake_claim)

    obj_key = f"dev/self_check_evidence/2026/07/31/{rec_id}/pentest.pdf"
    reg = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": obj_key, "filename": "pentest.pdf",
            "content_type": "application/pdf", "size_bytes": 2048,
        },
    )
    assert reg.status_code == 200, reg.text
    att_id = reg.json()["data"]["id"]

    async def _fake_presign(_db, *, object_key, **_kw):
        return f"https://s3.example/{object_key}?sig=x"

    monkeypatch.setattr(
        "app.domains.auditor.self_check_service.presign_get", _fake_presign,
    )

    resp = await client.get(
        f"/api/v1/auditor/self-checks/attachments/{att_id}/download",
        headers=auditor_headers,
    )
    assert resp.status_code == 200, resp.text
    assert resp.json()["data"]["url"].startswith("https://s3.example/")

    rows = (await db_session.execute(
        select(AuditLog).where(
            AuditLog.action == "auditor.self_check.attachment.download",
        )
    )).scalars().all()
    assert len(rows) == 1, (
        "no audit row for a presigned evidence download — forensics "
        "cannot answer 'who pulled this file'"
    )
    assert rows[0].detail["attachment_id"] == att_id
    assert rows[0].actor_user_id is not None
    assert storage_mod is not None


@pytest.mark.asyncio
async def test_ghost_attachment_cannot_satisfy_the_completion_gate(
    client: AsyncClient, auditor_headers, db_session, monkeypatch,
):
    """#2152 — registering a key that was never PUT must fail.

    Without the HEAD probe a caller could presign, skip the upload, and
    register anyway: ``attachment_count`` goes to 1, the
    ``requires_attachment`` gate opens, and the compliance ledger keeps
    a signed self-check whose evidence 404s when an auditor clicks it.
    The vendor-assessment twin grew this guard in the #1609 round; this
    half was left behind.
    """
    cat = await _seed_category(
        db_session, key="ghost_cat", requires_attachment=True,
    )
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()

    listed = await client.get(
        f"/api/v1/auditor/self-checks/records?category_id={cat.id}",
        headers=auditor_headers,
    )
    rec_id = listed.json()["data"]["items"][0]["id"]

    async def _absent(_db, *, object_key, **_kw):
        return None  # HEAD says the object is not there

    monkeypatch.setattr(
        "app.domains.auditor.self_check_service.stat_object", _absent,
    )

    reg = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": (
                f"dev/self_check_evidence/2026/07/31/{rec_id}/ghost.png"
            ),
            "filename": "ghost.png", "content_type": "image/png",
            "size_bytes": 1,
        },
    )
    assert reg.status_code in (400, 422), reg.text

    done = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/complete",
        headers=auditor_headers,
        json={"outcome": "normal", "remark": "no real evidence"},
    )
    assert done.status_code in (400, 422), (
        "a ghost attachment opened the requires_attachment gate"
    )


@pytest.mark.asyncio
async def test_size_cap_uses_the_authoritative_s3_size(
    client: AsyncClient, auditor_headers, db_session, monkeypatch,
):
    """#2152 — a client-declared ``size_bytes=None`` used to skip the cap."""
    cat = await _seed_category(
        db_session, key="bigfile_cat", requires_attachment=False,
    )
    await db_session.commit()
    await svc.open_due_records(db_session)
    await db_session.commit()
    listed = await client.get(
        f"/api/v1/auditor/self-checks/records?category_id={cat.id}",
        headers=auditor_headers,
    )
    rec_id = listed.json()["data"]["items"][0]["id"]

    async def _huge(_db, *, object_key, **_kw):
        return {"size_bytes": 40 * 1024 * 1024, "content_type": "image/png"}

    monkeypatch.setattr(
        "app.domains.auditor.self_check_service.stat_object", _huge,
    )

    reg = await client.post(
        f"/api/v1/auditor/self-checks/records/{rec_id}/attachments",
        headers=auditor_headers,
        json={
            "object_key": (
                f"dev/self_check_evidence/2026/07/31/{rec_id}/big.png"
            ),
            "filename": "big.png", "content_type": "image/png",
            "size_bytes": None,
        },
    )
    assert reg.status_code in (400, 422), (
        "40 MB object registered because the client claimed no size"
    )