"""Tests for GET /api/relations: discovery + content + live refresh after add_provenance."""
from __future__ import annotations


def test_schema_lists_personnel_with_columns(client):
    # /api/schema must include personnel (a SELECT-able table on the test
    # search_path) with its full column list, including the rewriter's
    # provsql tag, and exclude catalog / provsql-internal schemas.
    resp = client.get("/api/schema")
    assert resp.status_code == 200
    rows = resp.get_json()
    assert isinstance(rows, list) and rows, rows
    by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
    assert "provsql_test.personnel" in by_qname
    rel = by_qname["provsql_test.personnel"]
    assert rel["kind"] == "table"
    col_names = [c["name"] for c in rel["columns"]]
    assert col_names == ["id", "name", "position", "city", "classification", "provsql"]
    # The catalog and provsql schemas must be filtered out.
    schemas = {r["schema"] for r in rows}
    assert "pg_catalog" not in schemas
    assert "information_schema" not in schemas
    assert "provsql" not in schemas


def test_schema_marks_provenance_tracked_relations(client):
    """The schema endpoint flags relations carrying a `provsql uuid`
    column so the front-end can render the PROV pill. personnel is
    provenance-tracked by the conftest setup; pg_catalog tables are
    excluded outright, so we add a vanilla untracked table inline to
    check the negative case too."""
    setup = (
        "DROP TABLE IF EXISTS rel_test_plain;"
        " CREATE TABLE rel_test_plain (a int, b text)"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        rows = client.get("/api/schema").get_json()
        by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
        assert by_qname["provsql_test.personnel"]["has_provenance"] is True
        assert by_qname["provsql_test.rel_test_plain"]["has_provenance"] is False
    finally:
        client.post(
            "/api/exec",
            json={"sql": "DROP TABLE rel_test_plain", "mode": "circuit"},
        )


def test_schema_marks_views_propagating_provenance(client):
    """A view built on top of a provenance-tracked relation never carries
    a literal `provsql` column in pg_attribute, but ProvSQL's planner
    hook injects one into the rewritten output. The schema endpoint must
    still flag such views (CS2's `f` and `f_replicated` are the canonical
    case)."""
    setup = (
        "DROP VIEW IF EXISTS rel_test_view;"
        " DROP VIEW IF EXISTS rel_test_view_plain;"
        " DROP TABLE IF EXISTS rel_test_plain;"
        " CREATE TABLE rel_test_plain (a int, b text);"
        " CREATE VIEW rel_test_view        AS SELECT * FROM personnel;"
        " CREATE VIEW rel_test_view_plain  AS SELECT * FROM rel_test_plain;"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        rows = client.get("/api/schema").get_json()
        by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
        # View on a provenance-tracked table propagates provenance.
        assert by_qname["provsql_test.rel_test_view"]["has_provenance"] is True
        # View on a plain table does not.
        assert by_qname["provsql_test.rel_test_view_plain"]["has_provenance"] is False
    finally:
        client.post("/api/exec", json={
            "sql": (
                "DROP VIEW IF EXISTS rel_test_view;"
                " DROP VIEW IF EXISTS rel_test_view_plain;"
                " DROP TABLE IF EXISTS rel_test_plain;"
            ),
            "mode": "circuit",
        })


def test_schema_classifies_view_prov_kind(client):
    """Views have no ``provsql_table_info`` entry; ``prov_kind`` for a
    view comes from the planner-hook classifier (NOTICE on the view body
    captured by a LIMIT 0 probe).  A view that wraps a TID base table
    inherits TID ; a view over a BID base table inherits BID ; a view
    over a multi-source join classifies as OPAQUE."""
    setup = (
        "DROP VIEW IF EXISTS vk_over_tid;"
        " DROP VIEW IF EXISTS vk_over_bid;"
        " DROP VIEW IF EXISTS vk_over_join;"
        " DROP TABLE IF EXISTS vk_bid;"
        " CREATE TABLE vk_bid (k int, lbl text);"
        " SELECT repair_key('vk_bid', 'k');"
        " CREATE VIEW vk_over_tid  AS SELECT * FROM personnel;"
        " CREATE VIEW vk_over_bid  AS SELECT * FROM vk_bid;"
        " CREATE VIEW vk_over_join AS"
        "   SELECT p.id, b.k FROM personnel p, vk_bid b WHERE p.id = b.k;"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        rows = client.get("/api/schema").get_json()
        by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
        assert by_qname["provsql_test.vk_over_tid"]["prov_kind"]  == "tid"
        assert by_qname["provsql_test.vk_over_bid"]["prov_kind"]  == "bid"
        assert by_qname["provsql_test.vk_over_join"]["prov_kind"] == "opaque"
    finally:
        client.post("/api/exec", json={
            "sql": (
                "DROP VIEW IF EXISTS vk_over_tid;"
                " DROP VIEW IF EXISTS vk_over_bid;"
                " DROP VIEW IF EXISTS vk_over_join;"
                " DROP TABLE IF EXISTS vk_bid;"
            ),
            "mode": "circuit",
        })


def test_schema_surfaces_per_relation_prov_kind(client):
    """add_provenance / repair_key / provenance_guard write a per-table
    record into provsql_table_info marking each tracked relation as
    one of tid / bid / opaque.  The schema endpoint surfaces that as
    `prov_kind` so the front-end can render the PROV-TID / PROV-BID /
    PROV-OPAQUE badge.  pk_tid is a vanilla add_provenance, pk_bid
    flips to BID via repair_key, pk_opq starts as TID then becomes
    OPAQUE when a user INSERT supplies a literal provsql value (the
    provenance_guard trigger demotes the table).
    """
    setup = (
        "DROP TABLE IF EXISTS pk_tid, pk_bid, pk_opq;"
        " CREATE TABLE pk_tid (id int, lbl text);"
        " CREATE TABLE pk_bid (id int, k int, lbl text);"
        " CREATE TABLE pk_opq (id int, lbl text);"
        " SELECT add_provenance('pk_tid');"
        " SELECT repair_key('pk_bid', 'id');"
        " SELECT add_provenance('pk_opq');"
        " INSERT INTO pk_opq (id, lbl, provsql) "
        "    VALUES (1, 'x', '00000000-0000-0000-0000-000000000001');"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        rows = client.get("/api/schema").get_json()
        by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
        assert by_qname["provsql_test.pk_tid"]["prov_kind"] == "tid"
        assert by_qname["provsql_test.pk_bid"]["prov_kind"] == "bid"
        assert by_qname["provsql_test.pk_opq"]["prov_kind"] == "opaque"
        # An untracked relation must come back with prov_kind = None
        # (no per-table record) ; pk_untracked stands in for the
        # catalog-untracked case the front-end falls back on.
        client.post("/api/exec", json={
            "sql": "DROP TABLE IF EXISTS pk_untracked;"
                   " CREATE TABLE pk_untracked (a int)",
            "mode": "circuit",
        })
        rows2 = client.get("/api/schema").get_json()
        by_qname2 = {f"{r['schema']}.{r['table']}": r for r in rows2}
        assert by_qname2["provsql_test.pk_untracked"]["prov_kind"] is None
    finally:
        client.post("/api/exec", json={
            "sql": "DROP TABLE IF EXISTS pk_tid, pk_bid, pk_opq, pk_untracked;",
            "mode": "circuit",
        })


def test_schema_surfaces_provsql_column_types(client):
    """The schema endpoint must report the `random_variable` and
    `agg_token` types for columns that carry them, so the front-end can
    render the per-column RV / AGG pills in the schema panel. The
    fixture's search_path puts `provsql` before `public`, so
    format_type returns the unqualified form; we accept the qualified
    form too so the test stays robust to search_path drift."""
    setup = (
        "DROP TABLE IF EXISTS rel_test_provtypes;"
        " CREATE TABLE rel_test_provtypes ("
        "   id INT,"
        "   reading random_variable,"
        "   running agg_token"
        " );"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        rows = client.get("/api/schema").get_json()
        by_qname = {f"{r['schema']}.{r['table']}": r for r in rows}
        rel = by_qname["provsql_test.rel_test_provtypes"]
        col_types = {c["name"]: c["type"] for c in rel["columns"]}
        assert col_types["id"] == "integer"
        assert col_types["reading"] in ("random_variable", "provsql.random_variable")
        assert col_types["running"] in ("agg_token", "provsql.agg_token")
    finally:
        client.post(
            "/api/exec",
            json={"sql": "DROP TABLE rel_test_provtypes", "mode": "circuit"},
        )


def test_personnel_listed(client):
    resp = client.get("/api/relations")
    assert resp.status_code == 200
    relations = resp.get_json()
    by_name = {r["regclass"]: r for r in relations}
    assert "personnel" in by_name
    rel = by_name["personnel"]

    col_names = [c["name"] for c in rel["columns"]]
    # SELECT * column order matches the table's pg_attribute order with
    # the rewriter-injected provsql trailing.
    assert col_names == ["id", "name", "position", "city", "classification", "provsql"]
    assert len(rel["rows"]) == 7
    assert any(r["values"][1] == "John" for r in rel["rows"])


def test_each_row_has_provenance_uuid(client):
    relations = client.get("/api/relations").get_json()
    rel = next(r for r in relations if r["regclass"] == "personnel")
    for row in rel["rows"]:
        # uuid is a 36-char hyphenated string.
        assert isinstance(row["uuid"], str) and len(row["uuid"]) == 36
        # The provsql cell value matches the row's uuid identifier.
        assert row["values"][rel["prov_col"]] == row["uuid"]


def test_relations_truncation_when_table_exceeds_cap(test_dsn, tmp_path, monkeypatch):
    """When a relation has more rows than `max_sidebar_rows`, /api/relations
    must return exactly `max_sidebar_rows` rows, set truncated=True, and
    surface max_rows so the front-end can render the "showing N of ~T" hint.

    Personnel has 7 rows; we cap at 3 to force truncation."""
    from provsql_studio.app import create_app
    monkeypatch.setenv("PROVSQL_STUDIO_CONFIG_DIR", str(tmp_path / "studio_cfg"))
    app = create_app(
        dsn=f"{test_dsn} options='-c search_path=provsql_test,provsql,public'",
        max_sidebar_rows=3,
    )
    app.config.update(TESTING=True)
    client = app.test_client()
    relations = client.get("/api/relations").get_json()
    rel = next(r for r in relations if r["regclass"] == "personnel")
    assert rel["truncated"] is True
    assert rel["max_rows"] == 3
    assert len(rel["rows"]) == 3


def test_relations_no_truncation_when_under_cap(client):
    """Personnel has 7 rows; with the default cap (100) the response must
    not be flagged as truncated."""
    relations = client.get("/api/relations").get_json()
    rel = next(r for r in relations if r["regclass"] == "personnel")
    assert rel["truncated"] is False
    assert rel["max_rows"] == 100
    assert len(rel["rows"]) == 7


def test_relations_excludes_provenance_mappings(client):
    """Provenance-mapping-shaped relations -- tables with both `value` and
    `provenance uuid` columns -- play a different role from source data
    (they label input gates) and must not appear in the where-mode
    sidebar even when they happen to also carry a provsql column. The
    latter is the CTAS case: building a mapping with `CREATE TABLE foo
    AS SELECT x AS value, provenance() AS provenance FROM tracked` lets
    the planner hook inject a provsql column, which the older
    _RELATIONS_QUERY would have surfaced."""
    setup = (
        "DROP TABLE IF EXISTS rel_test_mapping;"
        # Mapping shape (value, provenance uuid) sourced from personnel
        # so the planner rewrite materializes a provsql column too.
        " CREATE TABLE rel_test_mapping AS"
        "   SELECT name AS value, provenance() AS provenance FROM personnel;"
        # And mark it tracked, just to be unambiguous about the case
        # under test (mapping-shape AND provsql column present).
        " SELECT add_provenance('rel_test_mapping'::regclass)"
    )
    resp = client.post("/api/exec", json={"sql": setup, "mode": "circuit"})
    assert resp.status_code == 200, resp.data
    try:
        relations = client.get("/api/relations").get_json()
        names = {r["regclass"] for r in relations}
        assert "rel_test_mapping" not in names, (
            "mapping-shape relation leaked into the where-mode sidebar"
        )
        # Personnel itself is still listed -- the filter shouldn't be
        # over-broad.
        assert "personnel" in names
    finally:
        client.post(
            "/api/exec",
            json={"sql": "DROP TABLE rel_test_mapping", "mode": "circuit"},
        )


def test_add_provenance_picks_up_new_relation(client, app):
    # Run add_provenance via /api/exec, then re-query relations and check
    # the new table shows up. Use a unique name so we don't clash with other tests.
    table = "rel_test_widget"

    # Drop any prior leftover and create the new table + provenance, all in one batch.
    setup_sql = (
        f"DROP TABLE IF EXISTS {table};"
        f" CREATE TABLE {table} (id INT, label TEXT);"
        f" INSERT INTO {table} VALUES (1, 'a'), (2, 'b');"
        f" SELECT add_provenance('{table}'::regclass)"
    )
    resp = client.post("/api/exec", json={"sql": setup_sql, "mode": "circuit"})
    assert resp.status_code == 200, resp.data

    try:
        relations = client.get("/api/relations").get_json()
        names = {r["regclass"] for r in relations}
        assert table in names
        rel = next(r for r in relations if r["regclass"] == table)
        col_names = [c["name"] for c in rel["columns"]]
        assert col_names == ["id", "label", "provsql"]
        assert len(rel["rows"]) == 2
    finally:
        client.post(
            "/api/exec",
            json={"sql": f"DROP TABLE {table}", "mode": "circuit"},
        )
