"""Adapter output contract tests — gatekeeper for §7.5.7.

Two layers of tests:

  A. **Contract self-tests** — the Pydantic models in
     `govcrawler.adapters.contract` reject malformed input (proves the
     guard is real, not ceremonial).

  B. **Real-data projection** — load the gkmlpt API JSON we captured in
     `data/probe/verify_httpx.json` and mechanically project each
     article into `CrawlItem`. If the contract is well-designed, this
     should never raise; if it does, the contract is too strict or the
     adapter will need a cleanup step before constructing `CrawlItem`.

New adapters add their own layer-B tests here (or in a sibling file)
pointing at their saved probe sample. Code review blocks the merge when
contract tests fail — enforces §7.5.7 "契约不通过不合并".
"""
from __future__ import annotations

import hashlib
import json
from datetime import date, datetime, timezone
from pathlib import Path

import pytest
from pydantic import ValidationError

from govcrawler.adapters.contract import (
    AttachmentItem,
    ContractViolation,
    CrawlItem,
    CrawlLogItem,
    FetchStrategy,
    Status,
)
from govcrawler.utils.url_norm import url_hash as compute_url_hash

PROBE = Path(__file__).resolve().parents[1] / "data" / "probe"

# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
SHA_EMPTY = hashlib.sha256(b"").hexdigest()  # valid 64-char hex for fixtures


def _make_minimal_item(**overrides) -> dict:
    """Smallest dict that should pass CrawlItem validation."""
    base = {
        "site_id": "qingcheng_fgw",
        "url": "http://www.qingcheng.gov.cn/post/1.html",
        "url_hash": compute_url_hash("http://www.qingcheng.gov.cn/post/1.html"),
        "title": "测试标题",
    }
    base.update(overrides)
    return base


# ===========================================================================
# LAYER A — Contract self-tests
# ===========================================================================
class TestCrawlItemBasics:
    def test_minimal_item_parses(self):
        item = CrawlItem(**_make_minimal_item())
        assert item.status == Status.RAW
        assert item.attachments == []
        assert item.native_post_id is None  # allowed to be None

    def test_empty_title_rejected(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(title=""))

    def test_bad_url_hash_rejected(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(url_hash="not-hex"))

    def test_extra_fields_forbidden(self):
        # extra=forbid guards against adapter typos silently dropping fields
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(sneaky="oops"))

    def test_content_simhash_forbidden(self):
        # §5.4 explicitly drops content_simhash; contract must refuse it
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(content_simhash="deadbeefcafe1234"))


class TestPublishTime:
    def test_naive_datetime_rejected(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(publish_time=datetime(2026, 4, 1)))

    def test_utc_datetime_accepted(self):
        dt = datetime(2026, 4, 1, tzinfo=timezone.utc)
        item = CrawlItem(**_make_minimal_item(publish_time=dt))
        assert item.publish_time == dt

    def test_absurd_year_rejected(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(
                publish_time=datetime(1900, 1, 1, tzinfo=timezone.utc)
            ))


class TestAttachmentConsistency:
    def _att(self, **k):
        base = {
            "file_path": "storage/a.pdf",
            "file_hash": SHA_EMPTY,
            "source_url": "http://x/a.pdf",
        }
        base.update(k)
        return AttachmentItem(**base)

    def test_has_attachment_requires_list(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(has_attachment=True))

    def test_list_requires_has_attachment(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(attachments=[self._att()]))

    def test_consistent_pair_ok(self):
        item = CrawlItem(**_make_minimal_item(
            has_attachment=True, attachments=[self._att()]
        ))
        assert item.has_attachment and len(item.attachments) == 1

    def test_bad_attachment_hash_rejected(self):
        with pytest.raises(ValidationError):
            AttachmentItem(file_path="x", file_hash="nope", source_url="http://x")


class TestReadyStatus:
    def test_ready_without_paths_rejected(self):
        with pytest.raises(ValidationError):
            CrawlItem(**_make_minimal_item(status=Status.READY))

    def test_ready_with_paths_ok(self):
        item = CrawlItem(**_make_minimal_item(
            status=Status.READY,
            content_text="正文",
            raw_html_path="storage/1.html",
            text_path="storage/1.txt",
        ))
        assert item.status == Status.READY


class TestCrawlLogItem:
    def test_success_with_error_msg_rejected(self):
        with pytest.raises(ValidationError):
            CrawlLogItem(
                site_id="x", strategy=FetchStrategy.HTTPX,
                success=True, error_msg="oops",
            )

    def test_fail_without_diagnostic_rejected(self):
        with pytest.raises(ValidationError):
            CrawlLogItem(site_id="x", strategy=FetchStrategy.HTTPX, success=False)

    def test_happy_path(self):
        log = CrawlLogItem(
            site_id="x", strategy=FetchStrategy.HTTPX, success=True,
            http_status=200, duration_ms=123,
        )
        assert log.strategy == FetchStrategy.HTTPX
        assert log.occurred_at.tzinfo is timezone.utc


# ===========================================================================
# LAYER B — Project real gkmlpt probe data through the contract
# ===========================================================================
def _load_gkmlpt_samples() -> list[dict]:
    """Grab the sample_article dicts from the httpx verify report."""
    report = PROBE / "verify_httpx.json"
    if not report.exists():
        pytest.skip(f"probe report not found; run scripts/verify_gkmlpt_httpx.py first: {report}")
    data = json.loads(report.read_text(encoding="utf-8"))
    return [r for r in data if r.get("sample_article")]


def _project_gkmlpt_article(sample: dict, site_id: str) -> CrawlItem:
    """Mechanical projection of gkmlpt list-API article → CrawlItem.

    This is what the real gkmlpt adapter's list-parse step must produce.
    Test-level version kept minimal & explicit so contract violations
    surface instead of being silently coerced.
    """
    a = sample["sample_article"]
    # `first_publish_time` is unix seconds; some rows have only `date`.
    ts = a.get("first_publish_time") or a.get("date")
    publish_time = (
        datetime.fromtimestamp(ts, tz=timezone.utc) if ts is not None else None
    )
    return CrawlItem(
        site_id=site_id,
        native_post_id=str(a["id"]),
        url=a["url"],
        url_hash=compute_url_hash(a["url"]),
        title=a["title"],
        publish_time=publish_time,
        source_raw=a.get("publisher"),
        publisher=a.get("publisher"),
        doc_no=a.get("document_number") or None,
        index_no=a.get("identifier") or None,
        # list stage: no body yet → status stays raw
        status=Status.RAW,
    )


@pytest.mark.parametrize(
    "report",
    _load_gkmlpt_samples() if (PROBE / "verify_httpx.json").exists() else [],
    ids=lambda r: r["label"],
)
def test_gkmlpt_sample_projects_cleanly(report):
    site_id = report["label"]
    item = _project_gkmlpt_article(report, site_id=site_id)
    # basic invariants that matter to RAG
    assert item.site_id == site_id
    assert item.native_post_id and item.native_post_id.isdigit()
    assert item.url_hash and len(item.url_hash) == 64
    assert item.title
    assert item.publish_time is not None
    assert item.publish_time.tzinfo is timezone.utc


def test_gkmlpt_bucket_invariant_holds():
    """url bucket = floor(native_post_id / 1000). Not part of the contract
    itself, but a gkmlpt-specific sanity check useful to keep near the
    contract tests (catches CMS format drift early)."""
    import re

    for report in _load_gkmlpt_samples():
        a = report["sample_article"]
        pid = a["id"]
        m = re.search(r"/content/\d+/(\d+)/post_(\d+)\.html", a["url"])
        assert m, f"{report['label']}: url shape drifted: {a['url']}"
        assert m.group(1) == str(pid // 1000)
        assert m.group(2) == str(pid)
