"""Pattern-library coverage (#2692) — the extractor that mirrors the hook,
the shape/record matcher, and the end-to-end measurement on real Postgres.

The extractor here and the hook's awk program (scribe_prior_art.sh ARM 1)
must agree on what counts as "a definition" — the metric and the write-path
backstop are two views of the same doctrine. The EXTRACTION_VECTORS below
deliberately reuse the definitions test_write_path_trigger stages for the
hook; extending one detector means extending both, and this comment is the
tripwire.
"""
import io
import json
import tarfile

import pytest
import pytest_asyncio

from scribe.services.coverage import (
    coverage_line,
    extract_shapes,
    largest_gaps,
    match_shapes,
    scannable,
    shapes_from_archive,
)

# --- unit: the definition extractor (shared vectors with the hook) -----------

EXTRACTION_VECTORS = [
    # (id, source text, expected (kind, name) list)
    ("python", "def make_app():\n    pass\nclass Config:\n    pass\n",
     [("sym", "make_app"), ("sym", "Config")]),
    ("python-dunder-skip", "class C:\n    def __init__(self):\n        pass\n",
     [("sym", "C")]),
    ("go-func", "func Resolve(x int) error {\n\treturn nil\n}\n",
     [("sym", "Resolve")]),
    ("go-method", "func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n",
     [("sym", "Resolve")]),
    ("kotlin-fun", "suspend fun refreshQueue(id: Long) {\n}\n",
     [("sym", "refreshQueue")]),
    ("rust-fn", "pub async fn fetch_all() -> u32 {\n    0\n}\n",
     [("sym", "fetch_all")]),
    ("go-type", "type ForgeAdapter struct {\n\tname string\n}\n",
     [("sym", "ForgeAdapter")]),
    ("rust-pub-crate", "pub(crate) struct Widget {}\n",
     [("sym", "Widget")]),
    ("js-export-default", "export default function App() {}\n",
     [("sym", "App")]),
    ("js-arrow", "const useThing = (id) => id;\nlet fetcher = async () => 0;\n",
     [("sym", "useThing"), ("sym", "fetcher")]),
    # Every selector line in a group counts — .btn-ghost, and .btn-text { }
    # both announce a class, exactly as the hook's awk sees them.
    ("css", ".btn-primary {\n  color: red;\n}\n.btn-ghost,\n.btn-text { }\n",
     [("css", "btn-primary"), ("css", "btn-ghost"), ("css", "btn-text")]),
    # Call sites, imports, and impl blocks are NOT definitions — matching
    # them would drown the metric exactly as it would drown the hook.
    ("non-definitions",
     "make_app()\nimpl Widget {\nreturn fetch_all\nimport os\nx = 1\n",
     []),
    ("dedup-within-file", "def f():\n    pass\ndef f():\n    pass\n",
     [("sym", "f")]),
]


@pytest.mark.parametrize(
    ("text", "expected"),
    [(t, e) for _i, t, e in EXTRACTION_VECTORS],
    ids=[i for i, _t, _e in EXTRACTION_VECTORS],
)
def test_extractor_agrees_with_the_hook_on_what_defines(text, expected):
    assert extract_shapes(text) == expected


def test_scannable_gates_prose_vendored_and_sourcemaps():
    assert scannable("src/app.py")
    assert scannable("web/button.css")
    assert scannable(".gitea/workflows/ci.yml")  # config IS worth recording
    assert not scannable("README.md")
    assert not scannable("dist/bundle.js.map")
    assert not scannable("node_modules/x/index.js")
    assert not scannable("web/node_modules/y/util.ts")
    # A FILE named like a skip-dir is not a directory hit.
    assert scannable("src/vendor.py")


# --- unit: reading shapes out of a forge tarball -----------------------------


def _tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
    buf = io.BytesIO()
    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
        for path, data in files.items():
            info = tarfile.TarInfo(f"{top}/{path}")
            info.size = len(data)
            tar.addfile(info, io.BytesIO(data))
    return buf.getvalue()


TREE = {
    "src/app.py": b"def make_app():\n    pass\nclass Config:\n    def __init__(self):\n        pass\n",
    "src/util.py": b"def helper():\n    pass\n",
    "web/button.css": b".btn {\n  color: red;\n}\n",
    "README.md": b"def not_code(): pass\n",
    "node_modules/x/index.js": b"function vendored() {}\n",
    "data.bin": b"\xff\xfe\x00\x01",
}
# What TREE holds once the gates run: 4 shapes, none from the skipped files.
TREE_SHAPES = [
    ("src/app.py", "sym", "make_app"),
    ("src/app.py", "sym", "Config"),
    ("src/util.py", "sym", "helper"),
    ("web/button.css", "css", "btn"),
]


def test_shapes_from_archive_strips_the_wrapper_and_gates_files():
    assert shapes_from_archive(_tarball(TREE)) == TREE_SHAPES


# --- unit: matching shapes against recorded locations ------------------------


def test_match_covers_by_exact_path_dir_prefix_and_css_dot():
    recorded = [
        ("src/app.py", "make_app"),   # exact file
        ("web", ".btn"),              # dir prefix + css dot normalization
    ]
    matched = match_shapes(TREE_SHAPES, recorded)
    covered = {name for _p, _k, name, ok in matched if ok}
    assert covered == {"make_app", "btn"}


def test_a_symbol_less_record_covers_nothing():
    """A whole-file snippet makes no claim about any particular definition
    inside it — crediting all of them would inflate the number for free."""
    matched = match_shapes(TREE_SHAPES, [("src/app.py", "")])
    assert not any(ok for *_x, ok in matched)


def test_no_prefix_bleed_between_sibling_directories():
    matched = match_shapes(
        [("src/library/x.py", "sym", "helper")], [("src/lib", "helper")]
    )
    assert not matched[0][3]


def test_largest_gaps_ranks_by_uncovered_and_drops_clean_dirs():
    matched = match_shapes(TREE_SHAPES, [("src/app.py", "make_app"), ("web", ".btn")])
    gaps = largest_gaps(matched)
    assert gaps == [{"dir": "src", "uncovered": 2, "total": 3}]


def test_coverage_line_is_evidence_carrying_and_labeled_estimate():
    line = coverage_line({
        "total": 210, "recorded": 34, "estimate": True,
        "computed_at": "2026-08-16T12:00:00+00:00",
        "largest_gaps": [
            {"dir": "internal/api", "uncovered": 40, "total": 60},
            {"dir": "web/src/components", "uncovered": 25, "total": 30},
        ],
    })
    assert "34/210 shapes recorded" in line
    assert "estimate" in line
    assert "2026-08-16" in line
    assert "internal/api, web/src/components" in line


def test_coverage_routes_are_registered():
    from scribe.app import create_app

    rules = {r.rule for r in create_app().url_map.iter_rules()}
    assert "/api/projects/<int:project_id>/coverage" in rules
    assert "/api/projects/<int:project_id>/coverage/refresh" in rules


# --- integration: the measurement end to end on real Postgres ----------------


def _forge(tar_bytes: bytes):
    import httpx

    from scribe.services.forge import GiteaForge

    def handler(request: httpx.Request) -> httpx.Response:
        path = request.url.path
        if path == "/api/v1/repos/alice/widget":
            return httpx.Response(200, json={"default_branch": "main"})
        if path == "/api/v1/repos/alice/widget/archive/main.tar.gz":
            return httpx.Response(200, content=tar_bytes)
        return httpx.Response(404, json={"message": "not found"})

    return GiteaForge(
        "https://git.example.com", "tok", transport=httpx.MockTransport(handler)
    )


def _selector(tar_bytes: bytes):
    """The keyring shape compute_coverage consumes since #2778 — one owner
    keyring holding the mocked Gitea adapter."""
    from scribe.services.forge import ForgeSelector

    return ForgeSelector((_forge(tar_bytes),))


@pytest_asyncio.fixture
async def _dispose_engine():
    from scribe.models import engine
    yield
    await engine.dispose()


@pytest_asyncio.fixture
async def seeded(_dispose_engine):
    """User + project + binding + two snippets that cover 2 of TREE's 4 shapes."""
    from sqlalchemy import select

    from scribe.models import async_session
    from scribe.models.project import Project
    from scribe.models.user import User
    from scribe.services import snippets as svc
    from scribe.services.repo_bindings import set_binding

    async with async_session() as s:
        user = (
            await s.execute(select(User).where(User.username == "coverage_itest"))
        ).scalar_one_or_none()
        if user is None:
            user = User(username="coverage_itest")
            s.add(user)
            await s.flush()
        project = Project(user_id=user.id, title="Widget")
        s.add(project)
        await s.flush()
        uid, pid = user.id, project.id
        await s.commit()

    await set_binding(uid, "https://git.example.com/alice/widget.git", pid)

    await svc.create_snippet(
        uid, name="cov_make_app", code="def make_app():\n    pass\n",
        language="python", repo="Widget", path="src/app.py",
        symbol="make_app", project_id=pid,
    )
    await svc.create_snippet(
        uid, name="cov_btn", code=".btn {\n  color: red;\n}\n",
        language="css", repo="Widget", path="web", symbol=".btn",
        project_id=pid,
    )
    return {"uid": uid, "pid": pid}


@pytest.mark.integration
async def test_coverage_measures_the_tree_exactly_and_caches(seeded):
    from scribe.services.coverage import (
        cached_coverage,
        compute_coverage,
        refresh_coverage,
    )

    uid, pid = seeded["uid"], seeded["pid"]
    selector = _selector(_tarball(TREE))

    coverage = await compute_coverage(uid, pid, selector=selector)
    assert coverage is not None
    assert coverage["total"] == 4
    assert coverage["recorded"] == 2
    assert coverage["estimate"] is True
    assert coverage["repos"] == [{
        "repo": "git.example.com/alice/widget", "ref": "main",
        "total": 4, "recorded": 2,
    }]
    assert coverage["largest_gaps"] == [{"dir": "src", "uncovered": 2, "total": 3}]

    # Nothing computed → nothing cached; refresh writes; the cache reads back
    # byte-equal, because enter_project will serve exactly this.
    assert await cached_coverage(uid, pid) is None
    stored = await refresh_coverage(uid, pid, selector=selector)
    assert (await cached_coverage(uid, pid)) == json.loads(json.dumps(stored))


@pytest.mark.integration
async def test_enter_project_surfaces_the_line_only_once_computed(seeded):
    from scribe.mcp._context import _user_id_ctx
    from scribe.mcp.tools.projects import enter_project
    from scribe.services.coverage import refresh_coverage

    uid, pid = seeded["uid"], seeded["pid"]
    token = _user_id_ctx.set(uid)
    try:
        # Forge-less / never-computed instance: the key is present, null, and
        # nothing else about the response changes.
        before = await enter_project(project_id=pid)
        assert before["pattern_coverage"] is None

        await refresh_coverage(uid, pid, selector=_selector(_tarball(TREE)))
        after = await enter_project(project_id=pid)
        line = after["pattern_coverage"]
        assert line.startswith(
            "pattern-library coverage: 2/4 shapes recorded (estimate, computed "
        )
        assert line.endswith("; largest gaps: src")
    finally:
        _user_id_ctx.reset(token)


@pytest.mark.integration
async def test_unservable_binding_measures_nothing(seeded):
    """A project bound only to a host the forge doesn't serve returns None —
    the same silence as no forge at all, never an error."""
    from scribe.services.coverage import compute_coverage
    from scribe.services.repo_bindings import set_binding

    from scribe.models import async_session
    from scribe.models.project import Project

    uid = seeded["uid"]
    async with async_session() as s:
        other = Project(user_id=uid, title="Elsewhere")
        s.add(other)
        await s.flush()
        other_pid = other.id
        await s.commit()
    await set_binding(uid, "https://github.com/somebody/else.git", other_pid)

    assert await compute_coverage(uid, other_pid, selector=_selector(_tarball(TREE))) is None