tests/test_ui.py

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import Any

from gcmd_classifier.models import (
    ArticleClassificationOutcome,
    ArticleProcessingStatus,
    ArticleRecord,
    ArticleResult,
    ClassificationFinalStatus,
    ClassificationRecord,
    DeterministicValidationResult,
    OutputError,
    OutputWarning,
    ProcessingMetadata,
    ReviewStatus,
    SupportType,
)
from gcmd_classifier.ui import gradio_app
from gcmd_classifier.vocabulary import load_vocabulary

FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
PROTOTYPE_PATH = Path("prototype/app_hf_poc.py")


class FakeComponent:
    instances: list[FakeComponent] = []

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        self.args = args
        self.kwargs = kwargs
        self.click_kwargs: dict[str, Any] | None = None
        type(self).instances.append(self)

    def click(self, **kwargs: Any) -> None:
        self.click_kwargs = kwargs


class FakeLayout:
    def __init__(self, **kwargs: Any) -> None:
        self.kwargs = kwargs

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc, traceback) -> None:
        return None


class FakeBlocks(FakeLayout):
    launched = 0
    queued = 0

    def queue(self):
        type(self).queued += 1
        return self

    def launch(self, **kwargs: Any) -> None:
        self.launch_kwargs = kwargs
        type(self).launched += 1


def _fake_gradio_module() -> SimpleNamespace:
    FakeComponent.instances = []
    return SimpleNamespace(
        Blocks=FakeBlocks,
        Group=FakeLayout,
        Row=FakeLayout,
        Textbox=FakeComponent,
        Number=FakeComponent,
        Markdown=FakeComponent,
        Dataframe=FakeComponent,
        Button=FakeComponent,
        JSON=FakeComponent,
    )


def _no_classification_result(article: ArticleRecord) -> ArticleResult:
    return ArticleResult(
        DOI=article.DOI,
        Title=article.Title,
        Year=article.Year,
        Abstract=article.Abstract,
        processing_status=ArticleProcessingStatus.COMPLETED,
        classification_outcome=ArticleClassificationOutcome.NOT_CLASSIFIED,
        classifications=(),
        no_classification_reason="No Topic selected.",
        review_status=ReviewStatus.NOT_REQUIRED,
        processing_metadata=ProcessingMetadata(cache_used=False),
    )


def _classification_record() -> ClassificationRecord:
    return ClassificationRecord(
        UUID="vl2-carbon-dioxide",
        name="CARBON DIOXIDE",
        level="Variable_Level_2",
        canonical_path="ATMOSPHERE > ATMOSPHERIC CHEMISTRY > CARBON > CARBON DIOXIDE",
        path_components=(
            "ATMOSPHERE",
            "ATMOSPHERIC CHEMISTRY",
            "CARBON",
            "CARBON DIOXIDE",
        ),
        topic="ATMOSPHERE",
        term="ATMOSPHERIC CHEMISTRY",
        parent_uuid="vl1-atmosphere-carbon",
        branch_id="topic:t/term:x/variable:a",
        classifier_evidence="The article discusses atmospheric carbon dioxide.",
        support_type=SupportType.EXPLICIT,
        reason_for_stopping="Deepest supported concept.",
        deterministic_validation=DeterministicValidationResult(valid=True),
        final_status=ClassificationFinalStatus.ACCEPTED,
        review_required=False,
        review_status=ReviewStatus.NOT_REQUIRED,
    )


def _classified_result(article: ArticleRecord) -> ArticleResult:
    return ArticleResult(
        DOI=article.DOI,
        Title=article.Title,
        Year=article.Year,
        Abstract=article.Abstract,
        processing_status=ArticleProcessingStatus.COMPLETED,
        classification_outcome=ArticleClassificationOutcome.CLASSIFIED,
        classifications=(_classification_record(),),
        review_status=ReviewStatus.NOT_REQUIRED,
        warnings=(OutputWarning(code="TEST_WARNING", message="A warning.", stage="test"),),
        errors=(OutputError(code="TEST_ERROR", message="An error.", stage="test"),),
        processing_metadata=ProcessingMetadata(cache_used=False),
    )


def test_gradio_app_imports_without_gradio_or_api_keys() -> None:
    module = importlib.import_module("gcmd_classifier.ui.gradio_app")

    assert hasattr(module, "create_demo")


def test_create_demo_constructs_gradio_interface(monkeypatch) -> None:
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())

    demo = gradio_app.create_demo(vocabulary=load_vocabulary(FIXTURE_PATH))

    assert isinstance(demo, FakeBlocks)
    assert "results-table" in gradio_app.GRADIO_CSS
    dataframes = [
        component
        for component in FakeComponent.instances
        if component.kwargs.get("headers") == gradio_app.CLASSIFICATION_TABLE_COLUMNS
    ]
    assert dataframes
    assert dataframes[0].kwargs["wrap"] is True
    assert dataframes[0].kwargs["elem_id"] == "classification-results-table"
    assert "overflow-wrap: anywhere" in gradio_app.GRADIO_CSS
    assert "text-overflow: unset" in gradio_app.GRADIO_CSS
    assert "overflow: visible" in gradio_app.GRADIO_CSS
    assert "height: auto" in gradio_app.GRADIO_CSS


def test_ui_module_does_not_classify_or_call_model_at_import_time(monkeypatch) -> None:
    calls = {"classified": 0}

    def fake_classify(**kwargs):
        calls["classified"] += 1
        return _no_classification_result(kwargs["article"])

    monkeypatch.setattr(gradio_app.pipeline_service, "classify_article", fake_classify)
    importlib.reload(gradio_app)

    assert calls["classified"] == 0


def test_root_app_imports_without_launching_server(monkeypatch) -> None:
    FakeBlocks.launched = 0
    FakeBlocks.queued = 0
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    sys.modules.pop("app", None)

    module = importlib.import_module("app")

    assert isinstance(module.demo, FakeBlocks)
    assert FakeBlocks.launched == 0
    assert FakeBlocks.queued == 0


def test_root_app_imports_when_spaces_is_not_installed(monkeypatch) -> None:
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    monkeypatch.delitem(sys.modules, "spaces", raising=False)
    sys.modules.pop("app", None)

    module = importlib.import_module("app")

    assert module._zerogpu_startup_probe() == "ok"


def test_root_app_defines_zerogpu_probe_with_spaces_decorator(monkeypatch) -> None:
    decorated = []

    class FakeSpaces:
        @staticmethod
        def GPU(function):
            decorated.append(function.__name__)
            function._fake_gpu_decorated = True
            return function

    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    monkeypatch.setitem(sys.modules, "spaces", FakeSpaces)
    sys.modules.pop("app", None)

    module = importlib.import_module("app")

    assert decorated == ["_zerogpu_startup_probe"]
    assert module._zerogpu_startup_probe._fake_gpu_decorated is True


def test_root_app_launch_uses_spaces_compatible_settings(monkeypatch) -> None:
    FakeBlocks.launched = 0
    FakeBlocks.queued = 0
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    monkeypatch.setenv("GRADIO_SERVER_NAME", "127.0.0.1")
    monkeypatch.setenv("GRADIO_SERVER_PORT", "9999")
    sys.modules.pop("app", None)
    module = importlib.import_module("app")

    module.launch()

    assert FakeBlocks.queued == 1
    assert FakeBlocks.launched == 1
    assert module.demo.launch_kwargs == {
        "css": module.GRADIO_CSS,
        "server_name": "127.0.0.1",
        "server_port": 9999,
        "share": False,
        "prevent_thread_lock": False,
    }


def test_root_app_uses_spaces_default_server_settings(monkeypatch) -> None:
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    monkeypatch.delenv("GRADIO_SERVER_NAME", raising=False)
    monkeypatch.delenv("GRADIO_SERVER_PORT", raising=False)
    sys.modules.pop("app", None)
    module = importlib.import_module("app")

    assert module.gradio_server_name() == "0.0.0.0"
    assert module.gradio_server_port() == 7860


def test_root_app_is_thin_launcher_without_classification_logic() -> None:
    text = Path("app.py").read_text()

    assert "from gcmd_classifier.ui.gradio_app import GRADIO_CSS, create_demo" in text
    assert "classify_article" not in text
    assert "route_topics" not in text
    assert "OpenAI" not in text
    assert "@spaces.GPU" in text
    assert "def _zerogpu_startup_probe" in text
    assert "demo.queue().launch" in text
    assert "server_name=gradio_server_name()" in text
    assert "server_port=gradio_server_port()" in text
    assert "share=False" in text
    assert "prevent_thread_lock=False" in text


def test_ui_calls_pipeline_service(monkeypatch) -> None:
    calls = {"count": 0}

    def fake_classify(**kwargs):
        calls["count"] += 1
        return _no_classification_result(kwargs["article"])

    monkeypatch.setattr(gradio_app.pipeline_service, "classify_article", fake_classify)

    summary, table, payload, diagnostics = gradio_app.run_demo_classification(
        Title="A title",
        Abstract="",
        DOI="10.example/ui",
        Year=2025,
        vocabulary=load_vocabulary(FIXTURE_PATH),
        model_client_factory=lambda settings: object(),
    )

    assert calls["count"] == 1
    assert "not_classified" in summary
    assert table == []
    assert payload["DOI"] == "10.example/ui"
    assert diagnostics["errors"] == []


def test_no_classification_result_is_formatted_correctly() -> None:
    article = ArticleRecord(DOI="10.example/no", Title="No", Year=2025, Abstract="")
    summary = gradio_app.format_classification_summary(_no_classification_result(article))

    assert "not_classified" in summary
    assert "No Topic selected." in summary


def test_classified_result_includes_uuid_and_canonical_path() -> None:
    article = ArticleRecord(DOI="10.example/yes", Title="Yes", Year=2025, Abstract="Text.")
    summary = gradio_app.format_classification_summary(_classified_result(article))

    assert "vl2-carbon-dioxide" in summary
    assert "ATMOSPHERE > ATMOSPHERIC CHEMISTRY > CARBON > CARBON DIOXIDE" in summary
    assert "Variable_Level_2" in summary
    assert "The article discusses atmospheric carbon dioxide." in summary


def test_compact_summary_includes_status_model_and_review_counts() -> None:
    article = ArticleRecord(DOI="10.example/summary", Title="Summary", Year=2025, Abstract="Text.")
    result = _classified_result(article).model_copy(
        update={
            "processing_metadata": ProcessingMetadata(
                model_provider="fake",
                model_name="fake-model",
            ),
            "classifications": (
                _classification_record().model_copy(update={"review_required": True}),
            ),
        }
    )

    summary = gradio_app.format_compact_summary(result)

    assert "processing_status" in summary
    assert "classification_outcome" in summary
    assert "model_provider:** `fake`" in summary
    assert "model_name:** `fake-model`" in summary
    assert "classifications:** `1`" in summary
    assert "requiring_review:** `1`" in summary


def test_classification_table_rows_include_only_demo_relevant_fields() -> None:
    article = ArticleRecord(DOI="10.example/table", Title="Table", Year=2025, Abstract="Text.")
    result = _classified_result(article)

    rows = gradio_app.classification_table_rows(result)

    assert gradio_app.CLASSIFICATION_TABLE_COLUMNS == [
        "GCMD Keyword Path",
        "Evidence",
        "Support",
        "Review Required",
    ]
    assert rows == [
        [
            "ATMOSPHERE > ATMOSPHERIC CHEMISTRY > CARBON > CARBON DIOXIDE",
            "The article discusses atmospheric carbon dioxide.",
            "explicit",
            False,
        ]
    ]


def test_errors_and_warnings_are_displayed() -> None:
    article = ArticleRecord(
        DOI="10.example/diagnostics",
        Title="Diagnostics",
        Year=2025,
        Abstract="",
    )
    result = _classified_result(article)

    summary = gradio_app.format_classification_summary(result)
    diagnostics = gradio_app.diagnostics_payload(result)

    assert "TEST_WARNING" in summary
    assert "TEST_ERROR" in summary
    assert diagnostics["warnings"][0]["code"] == "TEST_WARNING"
    assert diagnostics["errors"][0]["code"] == "TEST_ERROR"


def test_empty_abstract_is_accepted_by_ui_input_path(monkeypatch) -> None:
    seen = {"abstract": None}

    def fake_classify(**kwargs):
        article = kwargs["article"]
        seen["abstract"] = article.Abstract
        return _no_classification_result(article)

    monkeypatch.setattr(gradio_app.pipeline_service, "classify_article", fake_classify)

    summary, table, payload, _ = gradio_app.run_demo_classification(
        Title="Title only",
        Abstract="",
        DOI="10.example/title-only",
        Year=None,
        vocabulary=load_vocabulary(FIXTURE_PATH),
        model_client_factory=lambda settings: object(),
    )

    assert seen["abstract"] == ""
    assert table == []
    assert payload["Abstract"] == ""
    assert "not_classified" in summary


def test_prototype_app_remains_unchanged_during_ui_import(monkeypatch) -> None:
    before = PROTOTYPE_PATH.read_bytes()
    monkeypatch.setitem(sys.modules, "gradio", _fake_gradio_module())
    gradio_app.create_demo(vocabulary=load_vocabulary(FIXTURE_PATH))
    after = PROTOTYPE_PATH.read_bytes()

    assert after == before