test_review_queue_page.py

"""Review Queue page against the TEAM schema (AppTest, local Postgres).

The material tables the Space reads follow the InDeS 34-column schema: 9
legacy columns + image/image_url + the hardening columns, and NO id column
(origin/figure_id only after the migration). The page used to SELECT a
hard-coded list starting with `id`, every table raised UndefinedColumn, the
handler swallowed it and the page said "queue empty" while hundreds of rows
were flagged -- from the first deploy (13 Aug 2026) until this fix.

States exercised, each with flagged rows present where the state allows it:
  A. 34-column table (pre-migration: hardening columns, no origin/figure_id)
  B. after pg_mirror.ensure_schema (origin/figure_id/dedup2 present)
  C. legacy 11-column table (no status column at all)  -> empty, no error
  D. one material table missing                        -> "queue unreadable"
"""
import os
import sys

os.environ.setdefault("AGENT_WORK_DIR", "/tmp/agent_test")
os.environ["GEMINI_API_KEY"] = "test-key-not-used"
os.environ["AGENT_USE_NTRS"] = "0"      # tests stub the lanes; never touch the network
os.environ["AGENT_USE_S2"] = "0"
os.environ["AGENT_USE_OPENALEX_TOPICS"] = "0"
os.environ["AGENT_USE_DATASHEETS"] = "0"
os.environ["AGENT_SEED_QUERY_GRID"] = "0"
os.environ["AGENT_KEEPALIVE_URL"] = ""
# The page test controls the schema state itself; the boot migration must
# not silently upgrade state A into state B.
os.environ["AUTO_MIGRATE_MATERIAL_SCHEMA"] = "0"

from streamlit.testing.v1 import AppTest  # noqa: E402

import pg_mirror  # noqa: E402
from migrate import EXTRA_COLUMNS  # noqa: E402
from agent import agdb  # noqa: E402

TABLES = ("Polymers", "Fibers", "Composites_materials")
LEGACY = ("material_name text, material_abbreviation text, section text, "
          "property_name text, value text, unit text, english text, "
          "test_condition text, comments text, image bytea, image_url text")
fails = []


def check(name, cond):
    print(("PASS " if cond else "FAIL ") + name)
    if not cond:
        fails.append(name)


# Columns the live tables gained only through migrations after 1 Sep 2026
# (figure stage: origin/figure_id; figure linking: the three link columns).
POST_SEP1 = ("origin", "figure_id", "figure_ref", "figure_link_score", "figure_link_signals",
             # processing route per property (prompt 2.1, Oct 2026)
             "process_type", "process_name", "process_conditions", "process_quote",
             "process_page", "process_status")


def rebuild(conn, hardening: bool, with_origin: bool):
    with conn.cursor() as cur:
        for t in TABLES:
            cur.execute(f'DROP TABLE IF EXISTS "{t}" CASCADE')
            cur.execute(f'CREATE TABLE "{t}" ({LEGACY})')
            if hardening:
                for name, typ in EXTRA_COLUMNS:
                    if name in ("image", "image_url"):
                        continue            # InDeS legacy columns, already there
                    if not with_origin and name in POST_SEP1:
                        continue
                    cur.execute(f'ALTER TABLE "{t}" ADD COLUMN {name} {pg_mirror._pg_type(typ)}')
    conn.commit()


def seed_flagged(conn):
    """Two flagged rows + one ok row per table, like the live DB."""
    with conn.cursor() as cur:
        for t in TABLES:
            for i, status in enumerate(("unit_review", "out_of_range", "ok")):
                # Distinct on the pipeline dedup grain (material_key,
                # value_raw), as real rows are.
                cur.execute(
                    f'INSERT INTO "{t}" (material_name, material_key, section, property_name, '
                    f"value, value_raw, unit, source_pdf, page, status, flag_reason, "
                    f"source_sha1, extracted_at) "
                    f"VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)",
                    (f"{t} material {i}", f"{t.lower()}|{i}", "Mechanical",
                     "Tensile Strength", str(1000 + i), str(1000 + i), "MPa",
                     f"{t.lower()}_doc.pdf", 1 + i, status,
                     "" if status == "ok" else f"{status} test", "a" * 40,
                     f"2026-09-2{i}T00:00:00"))
    conn.commit()


def run_page():
    at = AppTest.from_file("page_files/Review_Queue.py", default_timeout=120)
    at.run()
    return at


def page_text(at) -> str:
    """Everything the page rendered as text: html blocks, markdown, errors."""
    parts = []
    for kind in ("html", "markdown", "error", "warning", "caption"):
        try:
            for el in at.get(kind):
                parts.append(str(getattr(el, "value", "") or getattr(el, "body", "")))
        except Exception:
            pass
    return "\n".join(parts)


def flagged_rows_shown(at) -> int:
    try:
        n = 0
        for df in at.dataframe:
            if "status" in list(df.value.columns):
                n += len(df.value)
        return n
    except Exception:
        return -1


conn = agdb.connect()
try:
    agdb.ensure_agent_schema(conn)

    # --- A: 34-column team schema, unmigrated ------------------------------
    rebuild(conn, hardening=True, with_origin=False)
    seed_flagged(conn)
    with conn.cursor() as cur:
        cur.execute("SELECT count(*) FROM information_schema.columns WHERE table_name='Polymers'")
        ncols = cur.fetchone()[0]
    check("state A tables have exactly 34 columns and no id", ncols == 34)
    at = run_page()
    txt = page_text(at)
    check("A: page renders without exception", not at.exception)
    check("A: page does NOT say 'queue empty'", "queue empty" not in txt and "Review queue is empty" not in txt)
    check("A: header pill says 6 flagged", "6 flagged" in txt)
    check("A: no 'queue unreadable' error", "queue unreadable" not in txt and not at.error)
    check("A: dataframe shows the 6 flagged rows", flagged_rows_shown(at) == 6)

    # --- B: after the migration --------------------------------------------
    added = pg_mirror.ensure_schema(conn)
    check("B: migration added origin/figure_id + the link columns on every table",
          all(set(added[t]) >= set(POST_SEP1) for t in TABLES))
    at = run_page()
    txt = page_text(at)
    check("B: page renders without exception", not at.exception)
    check("B: header pill says 6 flagged", "6 flagged" in txt)
    check("B: dataframe shows the 6 flagged rows", flagged_rows_shown(at) == 6)
    check("B: origin column present and 'text'",
          any("origin" in list(df.value.columns) and set(df.value["origin"]) == {"text"}
              for df in at.dataframe))

    # --- C: legacy 11-column tables (no status column) ---------------------
    rebuild(conn, hardening=False, with_origin=False)
    at = run_page()
    txt = page_text(at)
    check("C: page renders without exception", not at.exception)
    check("C: legacy tables -> 'queue empty', not an error",
          "queue empty" in txt and not at.error)

    # --- D: a material table is missing ------------------------------------
    rebuild(conn, hardening=True, with_origin=True)
    seed_flagged(conn)
    with conn.cursor() as cur:
        cur.execute('DROP TABLE "Fibers" CASCADE')
    conn.commit()
    at = run_page()
    txt = page_text(at)
    check("D: page renders without exception", not at.exception)
    check("D: missing table is reported in an error banner and the header pill",
          any("Fibers" in str(e.value) for e in at.error) and "unreadable" in txt)
    check("D: rows from the readable tables are still listed", flagged_rows_shown(at) == 4)
finally:
    conn.close()

print("\n%d checks failed" % len(fails))
sys.exit(1 if fails else 0)