"""安全问卷 · 乙方侧 — answer library, evidence docs, inbound lifecycle, AI apply.

Schema via ``create_all`` (no migrations), so no seed rows — each test
builds its own. The Bailian call is monkeypatched (no network / key).
"""
from __future__ import annotations

import io
from datetime import date, timedelta
from uuid import uuid4

import pytest
from httpx import AsyncClient
from openpyxl import Workbook
from sqlalchemy import select

from app.domains.auditor import (
    answer_library_service as lib,
)
from app.domains.auditor import (
    evidence_doc_service as docs,
)
from app.domains.auditor import (
    inbound_service as inb,
)
from app.domains.auditor.models import (
    AuditorInboundQuestion,
    AuditorInboundVersion,
)

# ── answer library ────────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_answer_library_crud_and_search(db_session):
    e = await lib.create_entry(
        db_session,
        payload={
            "category": "encryption",
            "question_zh": "静态数据是否加密?", "question_en": "Is data encrypted at rest?",
            "answer_zh": "是,AES-256。", "answer_en": "Yes, AES-256.",
        },
        actor_id=None,
    )
    await db_session.commit()
    res = await lib.list_entries(db_session, search="rest")
    assert res["meta"]["total"] == 1
    assert res["items"][0]["category"] == "encryption"
    # usage increment
    await lib.increment_usage(db_session, e.id)
    await db_session.commit()
    fresh = await lib.get_entry(db_session, e.id)
    assert fresh.usage_count == 1


@pytest.mark.asyncio
async def test_answer_library_backfill_creates_and_updates(db_session):
    e = await lib.backfill_from_answer(
        db_session, question_text="Do you have MFA?", answer_text="Yes, enforced.",
        category="access_control", actor_id=None,
    )
    await db_session.commit()
    assert e.answer_en == "Yes, enforced." and e.answer_zh == "Yes, enforced."
    e2 = await lib.backfill_from_answer(
        db_session, question_text="ignored", answer_text="Yes, enforced for all.",
        category="access_control", actor_id=None, existing_entry_id=e.id,
    )
    await db_session.commit()
    assert e2.id == e.id and e2.answer_en == "Yes, enforced for all."


# ── evidence documents ────────────────────────────────────────────────


@pytest.mark.asyncio
async def test_evidence_doc_key_validation_and_expiry(db_session):
    with pytest.raises(Exception):
        await docs.create_doc(
            db_session,
            payload={"title": "X", "object_key": "wrong/place/x.pdf",
                     "filename": "x.pdf", "content_type": "application/pdf"},
            actor_id=None,
        )
    d = await docs.create_doc(
        db_session,
        payload={
            "title": "ISO 27701 Cert", "category": "certificate",
            "object_key": "evidence_documents/2026/07/02/sys/abc-cert.pdf",
            "filename": "cert.pdf", "content_type": "application/pdf",
            "valid_until": (date.today() - timedelta(days=1)).isoformat(),
        },
        actor_id=None,
    )
    await db_session.commit()
    view = await docs.doc_view(db_session, d)
    assert view["is_expired"] is True
    listed = await docs.list_docs(db_session, category="certificate")
    assert len(listed) == 1


# ── inbound lifecycle ─────────────────────────────────────────────────


async def _make_questionnaire(db):
    q = await inb.create_questionnaire(
        db, payload={"customer_name": "BigCo", "title": "Vendor Security Review",
                     "due_date": date.today().isoformat()},
        actor_id=None,
    )
    await db.commit()
    return q


@pytest.mark.asyncio
async def test_inbound_add_answer_review_export(db_session, monkeypatch):
    # S3 isn't configured in the test schema; stub the archive upload so the
    # version-row / increment logic is what's under test (not boto3).
    async def _fake_put(db, *, relative_key, content, content_type):
        return f"dev/{relative_key}"

    monkeypatch.setattr(inb, "put_object_bytes", _fake_put)
    q = await _make_questionnaire(db_session)
    added = await inb.add_questions(db_session, q.id, items=[
        {"text": "Do you encrypt data at rest?", "section": "Tech"},
        {"text": "Do you have a DR plan?", "section": "BCP"},
    ])
    await db_session.commit()
    assert added == 2
    q = await inb.get_questionnaire(db_session, q.id)
    assert q.status == "answering"  # auto-advanced from received

    questions = (await db_session.execute(
        select(AuditorInboundQuestion).where(
            AuditorInboundQuestion.questionnaire_id == q.id,
        ).order_by(AuditorInboundQuestion.seq)
    )).scalars().all()
    # answer + review
    await inb.set_answer(db_session, questions[0].id, answer_text="Yes, AES-256.")
    await db_session.commit()
    assert (await inb.get_question(db_session, questions[0].id)).status == "drafted"
    with pytest.raises(Exception):
        # cannot review without an answer
        await inb.review_question(db_session, questions[1].id, reviewer_id=None, reviewed=True)
    await inb.review_question(db_session, questions[0].id, reviewer_id=None, reviewed=True)
    await db_session.commit()
    assert (await inb.get_question(db_session, questions[0].id)).status == "reviewed"

    # export archives a version
    buf, filename, version_no = await inb.export_and_archive(db_session, q.id, actor_id=None)
    await db_session.commit()
    assert version_no == 1 and filename.endswith(".xlsx") and buf.getvalue()[:2] == b"PK"
    versions = (await db_session.execute(
        select(AuditorInboundVersion).where(
            AuditorInboundVersion.questionnaire_id == q.id,
        )
    )).scalars().all()
    assert len(versions) == 1
    # second export → version 2
    _b, _f, v2 = await inb.export_and_archive(db_session, q.id, actor_id=None)
    await db_session.commit()
    assert v2 == 2


@pytest.mark.asyncio
async def test_inbound_import_questions_xlsx(db_session):
    q = await _make_questionnaire(db_session)
    wb = Workbook()
    ws = wb.active
    ws.append(["Section", "Question"])
    ws.append(["Tech", "Is MFA enforced?"])
    ws.append(["BCP", "Do you test backups?"])
    ws.append([None, None])  # blank row ignored
    buf = io.BytesIO()
    wb.save(buf)
    result = await inb.import_questions_xlsx(db_session, q.id, payload=buf.getvalue())
    await db_session.commit()
    assert result["added"] == 2
    qs = (await db_session.execute(
        select(AuditorInboundQuestion).where(
            AuditorInboundQuestion.questionnaire_id == q.id,
        )
    )).scalars().all()
    assert {x.section for x in qs} == {"Tech", "BCP"}


@pytest.mark.asyncio
async def test_ai_draft_applies_matches(db_session, monkeypatch):
    # seed a library entry + a questionnaire with a pending question
    entry = await lib.create_entry(
        db_session,
        payload={"category": "encryption", "question_zh": "加密?", "question_en": "Encryption?",
                 "answer_zh": "是", "answer_en": "Yes, AES-256 at rest and TLS in transit."},
        actor_id=None,
    )
    q = await _make_questionnaire(db_session)
    await inb.add_questions(db_session, q.id, items=[{"text": "Do you encrypt data?"}])
    await db_session.commit()
    question = (await db_session.execute(
        select(AuditorInboundQuestion).where(
            AuditorInboundQuestion.questionnaire_id == q.id,
        )
    )).scalar_one()

    async def _fake_match(db, *, questions, library):
        return {
            questions[0]["id"]: {
                "matched_id": str(entry.id), "confidence": 92,
                "draft": "Yes — AES-256 at rest, TLS 1.2+ in transit.",
            }
        }

    from app.domains.auditor import inbound_ai
    monkeypatch.setattr(inbound_ai, "match_and_draft", _fake_match)
    result = await inb.ai_draft_pending(db_session, q.id)
    await db_session.commit()
    assert result["drafted"] == 1 and result["matched"] == 1
    fresh = await inb.get_question(db_session, question.id)
    assert fresh.status == "drafted"
    assert fresh.match_confidence == 92
    assert str(fresh.matched_library_entry_id) == str(entry.id)
    # usage incremented on the matched entry
    assert (await lib.get_entry(db_session, entry.id)).usage_count == 1


@pytest.mark.asyncio
async def test_match_and_draft_calls_client_with_correct_kwargs(db_session, monkeypatch):
    """Regression (#1640): ``match_and_draft`` must call
    ``chat_json(system=..., user=...)`` — the old ``system_prompt=/``
    ``user_prompt=`` kwargs raised TypeError on every batch, which the
    swallow turned into a silent empty ``{}`` (drafted:0) result."""
    from app.domains.auditor import inbound_ai
    from app.integrations.models import PlatformSetting

    db_session.add(PlatformSetting(key="ai_spending.llm.api_key", value="sk-test"))
    await db_session.flush()

    entry = await lib.create_entry(
        db_session,
        payload={"category": "encryption", "question_zh": "加密?",
                 "question_en": "Encryption?", "answer_zh": "是",
                 "answer_en": "Yes, AES-256."},
        actor_id=None,
    )
    library = await lib.list_active_for_matching(db_session)

    seen_kwargs: dict = {}

    class _FakeClient:
        def __init__(self, *a, **kw):
            pass

        async def chat_json(self, **kwargs):
            seen_kwargs.update(kwargs)
            return {"answers": [{"id": "Q1", "matched_id": str(entry.id),
                                 "confidence": 88, "draft": "Yes — AES-256."}]}

    monkeypatch.setattr(
        "app.domains.bailian.llm_client.BailianLLMClient", _FakeClient,
    )
    results = await inbound_ai.match_and_draft(
        db_session,
        questions=[{"id": "Q1", "text": "Do you encrypt?"}],
        library=library,
    )
    # Correct kwarg names were used (would have raised TypeError otherwise).
    assert set(seen_kwargs) == {"system", "user"}
    assert results["Q1"]["draft"] == "Yes — AES-256."
    assert results["Q1"]["matched_id"] == str(entry.id)


@pytest.mark.asyncio
async def test_match_and_draft_raises_when_all_batches_fail(db_session, monkeypatch):
    """Regression (#1640): if EVERY batch fails, surface an error instead
    of returning ``{}`` as a silent false success."""
    from app.core.exceptions import AppException
    from app.domains.auditor import inbound_ai
    from app.integrations.models import PlatformSetting

    db_session.add(PlatformSetting(key="ai_spending.llm.api_key", value="sk-test"))
    await db_session.flush()
    library = await lib.list_active_for_matching(db_session)

    class _BoomClient:
        def __init__(self, *a, **kw):
            pass

        async def chat_json(self, **kwargs):
            raise RuntimeError("bailian down")

    monkeypatch.setattr(
        "app.domains.bailian.llm_client.BailianLLMClient", _BoomClient,
    )
    with pytest.raises(AppException) as ei:
        await inbound_ai.match_and_draft(
            db_session,
            questions=[{"id": "Q1", "text": "Do you encrypt?"}],
            library=library,
        )
    assert ei.value.code == "INBOUND_AI_DRAFT_FAILED"


@pytest.mark.asyncio
async def test_due_reminder_stages(db_session):
    await inb.create_questionnaire(
        db_session, payload={"customer_name": "C", "title": "T",
                             "due_date": date(2026, 8, 1).isoformat()},
        actor_id=None,
    )
    await db_session.commit()
    # 40 days out → nothing
    assert await inb.find_due_reminders(db_session, ref=date(2026, 6, 22)) == []
    # 5 days out → stage "7"
    due = await inb.find_due_reminders(db_session, ref=date(2026, 7, 27))
    assert len(due) == 1 and due[0][1] == "7"
    inb.mark_reminder_sent(due[0][0], "7")
    await db_session.commit()
    # same window → already reminded
    assert await inb.find_due_reminders(db_session, ref=date(2026, 7, 28)) == []
    # overdue → new stage
    due = await inb.find_due_reminders(db_session, ref=date(2026, 8, 5))
    assert len(due) == 1 and due[0][1] == "overdue"


def test_source_key_validation():
    # traversal + wrong-namespace rejected; a storage-prefixed in-namespace
    # key is accepted (position-independent segment membership).
    with pytest.raises(Exception):
        inb._validate_source_key("inbound_questionnaire/../secret.xlsx")
    with pytest.raises(Exception):
        inb._validate_source_key("payroll_exports/2026/salaries.xlsx")
    inb._validate_source_key("dev/inbound_questionnaire/2026/07/03/abc/x.xlsx")


@pytest.mark.asyncio
async def test_source_upload_rejects_bad_content_type(db_session):
    q = await _make_questionnaire(db_session)
    with pytest.raises(Exception):
        await inb.presign_source_upload(
            db_session, qid=q.id, person_id=uuid4(), filename="x.exe",
            content_type="application/x-msdownload", size_bytes=1,
            checksum_sha256="47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=",
        )


def test_stage_for_brackets_are_reachable():
    # Regression: ascending thresholds → the tightest window is returned, so
    # "3" and "due" are reachable. A descending tuple collapses every
    # days_left in [0,7] to "7" and these asserts fail.
    assert inb._stage_for(8) is None
    assert inb._stage_for(7) == "7"
    assert inb._stage_for(5) == "7"
    assert inb._stage_for(3) == "3"
    assert inb._stage_for(1) == "3"
    assert inb._stage_for(0) == "due"
    assert inb._stage_for(-1) == "overdue"


@pytest.mark.asyncio
async def test_due_reminder_fires_each_tighter_stage(db_session):
    # End-to-end: the 3-day and due-day nudges must fire (not just 7 + overdue).
    await inb.create_questionnaire(
        db_session, payload={"customer_name": "C2", "title": "T2",
                             "due_date": date(2026, 8, 1).isoformat()},
        actor_id=None,
    )
    await db_session.commit()
    seen = []
    for ref in (date(2026, 7, 27), date(2026, 7, 29), date(2026, 8, 1), date(2026, 8, 3)):
        due = await inb.find_due_reminders(db_session, ref=ref)
        rows = [d for d in due if d[0].customer_name == "C2"]
        if rows:
            seen.append(rows[0][1])
            inb.mark_reminder_sent(rows[0][0], rows[0][1])
            await db_session.commit()
    assert seen == ["7", "3", "due", "overdue"]


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


@pytest.mark.asyncio
async def test_inbound_rbac(client: AsyncClient, auditor_headers, employee_headers):
    r = await client.get(
        "/api/v1/auditor/inbound-questionnaires/overview", headers=auditor_headers,
    )
    assert r.status_code == 200
    r = await client.get(
        "/api/v1/auditor/answer-library", headers=employee_headers,
    )
    assert r.status_code == 403
    r = await client.post(
        "/api/v1/auditor/inbound-questionnaires", headers=employee_headers,
        json={"customer_name": "X", "title": "Y"},
    )
    assert r.status_code == 403