stack/tests/lead/test_explore_pages.py

"""The explorer pages render against a small mirror and do not write it."""

from __future__ import annotations

import sys
from pathlib import Path

from tests.lead.test_explore import _mirror
from zot.db import TYPE_MAP, Db
from zot.schema import create_db


def _run(source: str):
    from streamlit.testing.v1 import AppTest

    at = AppTest.from_string(source, default_timeout=90)
    at.run()
    assert not at.exception, at.exception
    return at


def _blob(at) -> str:
    chunks: list[str] = []
    for name in (
        "markdown",
        "caption",
        "warning",
        "error",
        "info",
        "success",
        "text",
        "title",
        "header",
        "subheader",
        "metric",
        "text_area",
    ):
        for item in getattr(at, name):
            chunks.append(str(getattr(item, "value", "") or ""))
            chunks.append(str(getattr(item, "label", "") or ""))
    return "\n".join(part for part in chunks if part)


def _widget(at, kind: str, label: str):
    kinds = ("button", "form_submit_button") if kind == "button" else (kind,)
    found = [
        item for name in kinds for item in getattr(at, name) if item.label == label
    ]
    assert found, f"no {kind} labeled {label}"
    return found[0]


def _mirror_path(tmp_path: Path) -> str:
    con = _mirror(tmp_path)
    path = str(tmp_path / "public.duckdb")
    con.close()
    return path


def test_coverage_filters_the_log_and_a_missing_file_is_an_error(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))
    at = _run("from lead.explore.views import coverage\ncoverage()\n")
    assert "loaded" in _blob(at)
    assert at.dataframe
    picked = _widget(at, "selectbox", "Status")
    picked.select("loaded").run()
    assert not at.exception, at.exception
    _widget(at, "text_input", "Source or path contains").input("qcew").run()
    assert not at.exception, at.exception
    assert at.dataframe

    monkeypatch.setenv("LEAD_PUBLIC_DB", str(tmp_path / "absent.duckdb"))
    missing = _run("from lead.explore.views import writeup\nwriteup()\n")
    assert "No mirror" in _blob(missing)


def test_a_locked_mirror_and_any_other_open_error_are_shown(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", str(tmp_path / "locked.duckdb"))

    def locked(_path):
        raise RuntimeError("Could not set lock on file")

    monkeypatch.setattr("lead.explore.views.open_mirror", locked)
    at = _run("from lead.explore.views import coverage\ncoverage()\n")
    assert "locked by a writer" in _blob(at)

    def broken(_path):
        raise RuntimeError("header said no")

    monkeypatch.setattr("lead.explore.views.open_mirror", broken)
    at = _run("from lead.explore.views import coverage\ncoverage()\n")
    assert "header said no" in _blob(at)


def test_tables_place_series_and_sql_read_the_mirror(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))
    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    assert browse.dataframe
    _widget(browse, "text_input", "Filter names").input("zzzz").run()
    assert not browse.exception, browse.exception
    assert "No table name matches." in _blob(browse)

    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    _widget(browse, "selectbox", "Table").select("bls.qcew").run()
    assert not browse.exception, browse.exception
    _widget(browse, "text_input", "Year").input("nope")
    _widget(browse, "button", "Show 100 rows").click().run()
    assert "Year must be a number." in _blob(browse)

    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    _widget(browse, "selectbox", "Table").select("bls.qcew").run()
    _widget(browse, "selectbox", "Geography column").select("geo_fips")
    _widget(browse, "text_input", "Geography value").input("01001")
    _widget(browse, "button", "Show 100 rows").click().run()
    assert "needs a year" in _blob(browse)

    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    _widget(browse, "selectbox", "Table").select("bls.qcew").run()
    _widget(browse, "text_input", "Year").input("2020")
    _widget(browse, "button", "Profile").click().run()
    assert not browse.exception, browse.exception
    assert browse.dataframe

    place = _run("from lead.explore.views import place\nplace()\n")
    _widget(place, "button", "Look up").click().run()
    assert "Enter a value." in _blob(place)
    _widget(place, "text_input", "Value").input("0100")
    _widget(place, "text_input", "Year (required for a heavy table)").input("year")
    _widget(place, "button", "Look up").click().run()
    assert "Year must be a number." in _blob(place)

    place = _run("from lead.explore.views import place\nplace()\n")
    _widget(place, "text_input", "Value").input("01001")
    _widget(place, "text_input", "Year (required for a heavy table)").input("2020")
    _widget(place, "button", "Look up").click().run()
    assert not place.exception, place.exception
    assert place.dataframe
    assert "tables have this value" in _blob(place)

    series = _run("from lead.explore.views import series\nseries()\n")
    _widget(series, "button", "Plot").click().run()
    assert not series.exception, series.exception
    assert series.dataframe
    assert "Last minus first" in _blob(series)

    reader = _run("from lead.explore.views import reader\nreader()\n")
    _widget(reader, "button", "Run").click().run()
    assert not reader.exception, reader.exception
    assert reader.dataframe
    _widget(reader, "text_area", "Statement").input("DROP TABLE bls.qcew")
    _widget(reader, "button", "Run").click().run()
    assert "only a read is allowed" in _blob(reader)


def test_each_page_says_when_the_mirror_is_missing(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", str(tmp_path / "absent.duckdb"))
    at = _run(
        "from lead.explore.views import browse, place, series, reader\n"
        "browse()\nplace()\nseries()\nreader()\n"
    )
    assert _blob(at).count("No mirror") == 4


def test_show_rows_and_a_heavy_profile_without_a_year(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))
    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    _widget(browse, "selectbox", "Table").select("lead_public.cms_demo").run()
    _widget(browse, "button", "Show 100 rows").click().run()
    assert not browse.exception, browse.exception
    assert browse.dataframe

    browse = _run("from lead.explore.views import browse\nbrowse()\n")
    _widget(browse, "selectbox", "Table").select("bls.qcew").run()
    _widget(browse, "button", "Profile").click().run()
    assert "needs a year" in _blob(browse)


def test_place_and_series_surface_a_query_error(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))

    def place_refused(*_args, **_kwargs):
        raise ValueError("place refused")

    monkeypatch.setattr("lead.explore.views.place_summary", place_refused)
    place = _run("from lead.explore.views import place\nplace()\n")
    _widget(place, "text_input", "Value").input("01001")
    _widget(place, "button", "Look up").click().run()
    assert "place refused" in _blob(place)

    def series_refused(*_args, **_kwargs):
        raise ValueError("series refused")

    monkeypatch.setattr("lead.explore.views.series_points", series_refused)
    series = _run("from lead.explore.views import series\nseries()\n")
    _widget(series, "button", "Plot").click().run()
    assert "series refused" in _blob(series)


def test_an_empty_mirror_has_nothing_to_browse(tmp_path, monkeypatch):
    path = tmp_path / "empty.duckdb"
    import duckdb

    duckdb.connect(str(path)).close()
    monkeypatch.setenv("LEAD_PUBLIC_DB", str(path))
    source = (
        "from lead.explore.views import browse, place, series\n"
        "browse()\nplace()\nseries()\n"
    )
    at = _run(source)
    text = _blob(at)
    assert "no tables" in text
    assert "No table carries a geography column." in text
    assert "No table has both series_id" in text


def test_a_heavy_series_is_typed_and_an_unknown_id_is_empty(tmp_path, monkeypatch):
    con = _mirror(tmp_path)
    con.execute(
        """
        UPDATE lead_public.load_log
        SET row_count = 2000000
        WHERE table_name = 'bls.ces' AND status = 'loaded'
        """
    )
    path = str(tmp_path / "public.duckdb")
    con.close()
    monkeypatch.setenv("LEAD_PUBLIC_DB", path)
    at = _run("from lead.explore.views import series\nseries()\n")
    assert "not listed from a scan" in _blob(at)
    _widget(at, "text_input", "series_id").input("NOPE")
    _widget(at, "button", "Plot").click().run()
    assert "No rows for that id." in _blob(at)
    _widget(at, "text_input", "series_id").input("CES1")
    _widget(at, "button", "Plot").click().run()
    assert not at.exception, at.exception
    assert "Last minus first" in _blob(at)


def test_a_series_list_that_fails_asks_for_the_id(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))

    def fail(*_args, **_kwargs):
        from lead.explore.catalog import NeedsFilter

        raise NeedsFilter("bls.ces")

    monkeypatch.setattr("lead.explore.views.series_ids", fail)
    at = _run("from lead.explore.views import series\nseries()\n")
    assert "needs a year or a series" in _blob(at)
    assert _widget(at, "text_input", "series_id")


def test_the_shell_links_osf_and_reconnects(tmp_path, monkeypatch):
    monkeypatch.setenv("LEAD_PUBLIC_DB", _mirror_path(tmp_path))
    at = _run("from lead.explore.app import main\nmain()\n")
    urls = [str(getattr(item, "url", "")) for item in at.link_button]
    assert any("osf.io/8wq3m" in url for url in urls) or "osf.io/8wq3m" in _blob(at)
    _widget(at, "button", "Reconnect").click().run()
    assert not at.exception, at.exception


def test_module_main_forwards_streamlit_and_keeps_a_chosen_theme(monkeypatch):
    from bcda.ui import brand
    from lead.explore import __main__ as entry

    seen: dict = {}

    def exit_three():
        seen["argv"] = list(sys.argv)
        raise SystemExit(3)

    monkeypatch.setattr("streamlit.web.cli.main", exit_three)
    assert entry.main(["--server.port", "1"]) == 3
    argv = seen["argv"]
    assert argv[:2] == ["streamlit", "run"]
    assert argv[3:5] == ["--theme.base", str(brand.THEME_PATH)]

    def already():
        seen["argv"] = list(sys.argv)

    monkeypatch.setattr("streamlit.web.cli.main", already)
    assert entry.main(["--theme.base", "plain"]) == 0
    assert seen["argv"].count("--theme.base") == 1
    assert entry.main(["--theme.base=plain"]) == 0

    def exit_none():
        raise SystemExit(None)

    monkeypatch.setattr("streamlit.web.cli.main", exit_none)
    assert entry.main([]) == 0


def _zotero(tmp_path: Path, *, attachment: Path | None) -> Path:
    library = tmp_path / "zotero.sqlite"
    create_db(str(library)).close()
    with Db(str(library)) as db:
        kept = db.create_item(TYPE_MAP["journalArticle"], key="AAAAAAAA")
        db.set_field(kept, "title", "A predecessor of the frailty index")
        db.set_field(kept, "abstractNote", "Medicare cohort, factor before outcome.")
        db.set_field(kept, "date", "2020")
        db.set_field(kept, "DOI", "10.0000/example")
        db.add_creators(kept, [("Ada", "Lovelace")])
        db.sync_tags(kept, ["project:lead-history"])
        if attachment is not None:
            db.add_attachment(kept, path=str(attachment), key="BBBBBBBB")
        db.commit()
    return library


def _labels(monkeypatch, tmp_path: Path, library: Path) -> None:
    from lead.explore import review_page

    monkeypatch.setattr(review_page, "_ZOTERO", str(library))
    monkeypatch.setattr(review_page, "_STORAGE", str(tmp_path / "storage"))
    monkeypatch.setattr(review_page, "_LABELS", str(tmp_path / "labels.sqlite"))


def test_the_label_page_saves_a_decision_and_shows_a_full_text(tmp_path, monkeypatch):
    html = tmp_path / "paper.html"
    html.write_text(
        "<html><body><p>The exposure precedes the outcome.</p></body></html>"
    )
    _labels(monkeypatch, tmp_path, _zotero(tmp_path, attachment=html))
    at = _run("from lead.explore.review_page import review\nreview()\n")
    assert "Ada Lovelace" in _blob(at)
    assert "Nothing unlabeled" not in _blob(at)
    _widget(at, "radio", "Decision").set_value("include")
    _widget(at, "text_area", "Why, in a sentence that names the criterion").input(
        "Names a predecessor of a registry target."
    )
    _widget(at, "button", "Save label").click().run()
    assert not at.exception, at.exception
    assert "Nothing unlabeled on this pass." in _blob(at)

    from lead.review.store import Store

    saved = Store(tmp_path / "labels.sqlite")
    assert saved.latest("lead-history", "title_abstract", 1) == {"AAAAAAAA": "include"}
    _widget(at, "radio", "Pass").set_value(2).run()
    assert not at.exception, at.exception
    assert "Ada Lovelace" in _blob(at)

    fresh = _run("from lead.explore.review_page import review\nreview()\n")
    _widget(fresh, "selectbox", "Stage").select("full_text").run()
    assert not fresh.exception, fresh.exception
    assert "exposure precedes" in _blob(fresh).lower()


def test_pass_two_waits_and_a_bad_exclude_is_not_saved(tmp_path, monkeypatch):
    _labels(monkeypatch, tmp_path, _zotero(tmp_path, attachment=None))
    at = _run("from lead.explore.review_page import review\nreview()\n")
    _widget(at, "radio", "Pass").set_value(2).run()
    assert "Pass 2 waits until pass 1 has a label." in _blob(at)

    at = _run("from lead.explore.review_page import review\nreview()\n")
    _widget(at, "selectbox", "Stage").select("full_text").run()
    assert "No full text is attached." in _blob(at)
    _widget(at, "radio", "Decision").set_value("exclude")
    _widget(at, "text_area", "Why, in a sentence that names the criterion").input(
        "No temporal order."
    )
    _widget(at, "button", "Save label").click().run()
    assert "codebook" in _blob(at)


def test_a_missing_library_is_an_error_and_the_actor_is_the_sso_header(
    tmp_path, monkeypatch
):
    from lead.explore import review_page

    monkeypatch.setattr(review_page, "_ZOTERO", str(tmp_path / "missing.sqlite"))
    monkeypatch.setattr(review_page, "_LABELS", str(tmp_path / "labels.sqlite"))
    at = _run("from lead.explore.review_page import review\nreview()\n")
    assert at.error

    class _Context:
        def __init__(self, headers):
            self.headers = headers

    monkeypatch.setattr(
        review_page.st,
        "context",
        _Context({"X-Auth-Request-Email": "kert@example"}),
        raising=False,
    )
    assert review_page._actor() == "kert@example"
    monkeypatch.setattr(
        review_page.st,
        "context",
        _Context({"X-Auth-Request-User": "kert"}),
        raising=False,
    )
    assert review_page._actor() == "kert"
    monkeypatch.setattr(review_page.st, "context", _Context({}), raising=False)
    assert review_page._actor() == "reviewer"


def test_an_unreadable_attachment_is_a_warning(tmp_path, monkeypatch):
    blob = tmp_path / "scan.txt"
    blob.write_text("not html")
    _labels(monkeypatch, tmp_path, _zotero(tmp_path, attachment=blob))
    at = _run("from lead.explore.review_page import review\nreview()\n")
    _widget(at, "selectbox", "Stage").select("full_text").run()
    assert "no text came out of it" in _blob(at)


def test_two_passes_that_disagree_are_counted(tmp_path, monkeypatch):
    from lead.review.store import Store

    library = _zotero(tmp_path, attachment=None)
    labels = tmp_path / "labels.sqlite"
    store = Store(labels)
    for pass_n, decision in ((1, "include"), (2, "exclude")):
        store.record(
            zotero_key="AAAAAAAA",
            project="lead-history",
            stage="title_abstract",
            pass_n=pass_n,
            decision=decision,
            reason="off-topic" if decision == "exclude" else "",
            rationale=f"Pass {pass_n}.",
            actor="kert",
        )
    from lead.explore import review_page

    monkeypatch.setattr(review_page, "_ZOTERO", str(library))
    monkeypatch.setattr(review_page, "_STORAGE", str(tmp_path / "storage"))
    monkeypatch.setattr(review_page, "_LABELS", str(labels))
    at = _run("from lead.explore.review_page import review\nreview()\n")
    assert "disagree" in _blob(at)
    assert "Nothing unlabeled on this pass." in _blob(at)