tests/test_app.py

"""Gradio app integration tests."""

import json
import time
from dataclasses import replace
from datetime import datetime, timedelta, timezone

import gradio as gr
from PIL import Image

import app
from coastwise.source_cache import DEFAULT_SEED_CACHE_DIR, SourceCache, load_seed_cache


def test_manual_refresh_applies_to_active_cache_and_preserves_curated_facts():
    from coastwise.schemas import RefreshStatus

    app._SESSION_CACHE = None
    try:
        before = app._active_cache()
        assert not any(s.refresh_status is RefreshStatus.UPDATED for s in before.snapshots)

        app.refresh_official_sources_status(
            fetcher=lambda source: f"refreshed official source text for {source.id} 2026"
        )

        after = app._active_cache()
        # Refresh now actually changes what subsequent answers read.
        assert any(s.refresh_status is RefreshStatus.UPDATED for s in after.snapshots)
        # Curated structured facts are preserved, never auto-derived from fetched text.
        assert len(after.facts) == len(before.facts)
        assert after.facts == before.facts
    finally:
        app._SESSION_CACHE = None


def test_app_import_exposes_coastwise_identity_without_launching():
    assert app.APP_TITLE == "CoastWise"
    assert isinstance(app.demo, gr.Blocks)


def test_build_app_returns_blocks_with_lookup_shell():
    demo = app.build_app()

    assert isinstance(demo, gr.Blocks)
    assert demo.title == "CoastWise"


def test_source_backed_app_shell_exposes_primary_controls_without_legacy_inputs():
    assert "Question" in app.SOURCE_BACKED_CONTROL_LABELS
    assert "Beach or coastal area" not in app.SOURCE_BACKED_CONTROL_LABELS
    assert "Refresh official sources" in app.SOURCE_BACKED_CONTROL_LABELS
    assert "Intent" not in app.SOURCE_BACKED_CONTROL_LABELS
    assert "Region" not in app.SOURCE_BACKED_CONTROL_LABELS
    assert "Photo candidate" not in app.SOURCE_BACKED_CONTROL_LABELS


def test_source_backed_question_journey_renders_evidence_and_next_action():
    # Assert freshness against a controlled time inside the seed's 24h window so the
    # test stays deterministic regardless of the real date (seed dates drift over time).
    seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR)
    fresh_now = max(snapshot.retrieved_at for snapshot in seed.snapshots) + timedelta(hours=1)

    result = app.ask_source_backed_question("what's the min size of lingcod?", "")
    markdown = app.render_source_backed_answer_from_cache(
        "what's the min size of lingcod?", "", seed, fresh_now
    )

    assert result["direct_answer"]
    assert result["source_links"]
    assert "current" in markdown.lower()
    assert "22" in markdown
    assert "Source evidence" in markdown
    assert "Retrieved" in markdown
    assert "Next safe action" in markdown


def test_manual_refresh_status_uses_nontechnical_source_group_language():
    status = app.refresh_official_sources_status(fetcher=lambda _source: "unchanged text")

    assert "CDFW" in status["message"]
    assert "CDPH" in status["message"]
    assert "checked" in status["message"].lower()
    assert "traceback" not in status["message"].lower()


def test_app_renders_stale_and_no_cache_source_backed_states():
    now = datetime(2026, 6, 14, 13, tzinfo=timezone.utc)
    seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR)
    stale = SourceCache(
        sources=seed.sources,
        snapshots=tuple(replace(snapshot, retrieved_at=now - timedelta(hours=25)) for snapshot in seed.snapshots),
        chunks=seed.chunks,
        facts=seed.facts,
        advisory_facts=seed.advisory_facts,
    )

    stale_markdown = app.render_source_backed_answer_from_cache(
        "what's the min size of lingcod?",
        "",
        stale,
        now,
    )
    no_cache_markdown = app.render_source_backed_answer_from_cache(
        "what's the min size of lingcod?",
        "",
        SourceCache((), (), (), ()),
        now,
    )

    assert "stale" in stale_markdown.lower()
    assert "refresh official" in no_cache_markdown.lower()


def test_app_renders_cdph_warning_and_verification_for_mussels():
    markdown = app.render_source_backed_answer("can I eat mussels from Pacifica today?", "")
    lowered = markdown.lower()

    assert "cdph" in lowered
    assert "cooking does not reliably" in lowered
    assert "verify" in lowered
    assert "safe to eat" not in lowered


def test_app_renders_source_evidence_limits_and_partial_unsupported_states():
    now = datetime(2026, 6, 14, 13, tzinfo=timezone.utc)
    partial = app.render_source_backed_answer("what should I check for ocean rules near Monterey?", "Monterey")
    unsupported = app.render_source_backed_answer("what is the trout limit in Lake Tahoe?", "")
    no_cache = app.render_source_backed_answer_from_cache(
        "what's the min size of lingcod?",
        "",
        SourceCache((), (), (), ()),
        now,
    )
    combined = "\n".join((partial, unsupported, no_cache)).casefold()

    assert "source evidence" in partial.casefold()
    assert "uncertainty" in partial.casefold()
    assert "partial" in partial.casefold()
    assert "outside this mvp" in unsupported.casefold()
    assert "refresh official" in no_cache.casefold()
    assert "does not authorize harvest" in combined
    assert "session-only" in combined
    assert "source-backed evidence" in combined


def test_cached_answer_app_build_and_refresh_feedback_fit_latency_budgets():
    start = time.perf_counter()
    markdown = app.render_source_backed_answer("what's the min size of lingcod?", "")
    answer_seconds = time.perf_counter() - start

    start = time.perf_counter()
    demo = app.build_app()
    build_seconds = time.perf_counter() - start

    start = time.perf_counter()
    status = app.refresh_official_sources_status(fetcher=lambda _source: "unchanged official source text")
    refresh_seconds = time.perf_counter() - start

    assert "22" in markdown
    assert isinstance(demo, gr.Blocks)
    assert "checked" in status["message"].casefold()
    assert answer_seconds < 3
    assert build_seconds < 3
    assert refresh_seconds < 3


def test_lookup_species_returns_source_backed_result_dict():
    result = app.lookup_species("lingcod", "Just curious", "Point Arena to Pigeon Point", "")

    assert result["display_name"] == "Lingcod"
    assert result["status"] == "rules_apply"
    assert result["source_links"]
    assert "verify" in result["next_safe_action"].lower()


def test_lookup_species_unknown_defaults_to_observe_only():
    result = app.lookup_species("unknown blob", "Thinking of harvest", "Point Arena to Pigeon Point", "")

    assert result["display_name"] == "Unknown organism"
    assert result["status"] == "observe_only"
    assert "official" in result["next_safe_action"].lower()


def test_lookup_species_with_harvest_context_shows_mussel_quarantine_warning():
    result = app.lookup_species("mussels", "Thinking of harvest", "Point Arena to Pigeon Point", "")

    assert result["display_name"] == "Mussels"
    assert result["status"] == "do_not_harvest"
    assert any("quarantine" in warning.lower() for warning in result["advisory_warnings"])


def test_render_lookup_result_shows_advisory_warnings_before_rule_notes():
    markdown = app.render_lookup_result(
        "mussels",
        "Thinking of harvest",
        "Point Arena to Pigeon Point",
        "",
    )

    assert markdown.index("**Advisory and location warnings**") < markdown.index("**Rule notes**")
    assert "quarantine" in markdown.lower()


def test_lookup_species_includes_generated_explanation_and_workflow_trace():
    result = app.lookup_species("lingcod", "Just curious", "Point Arena to Pigeon Point", "")

    assert result["generated_explanation"]
    assert result["workflow_trace"]
    assert result["workflow_trace"]["rule_steps"]
    assert result["generated_explanation"]["fallback_used"] is True


def test_lookup_photo_candidate_returns_candidate_without_harvest_guidance():
    result = app.lookup_photo_candidate(Image.new("RGB", (64, 64), (25, 25, 30)))

    assert result["candidates"][0]["rule_card_key"] == "mussels"
    assert "not confirmed" in result["candidates"][0]["limitations"].lower()
    assert "harvest_status" not in result


def test_app_routes_questions_through_configured_interpreter(monkeypatch):
    payload = {
        "species_or_category": "dungeness_crab",
        "location_name": "Mendocino",
        "region_key": "mendocino",
        "intent": "permission_status",
        "requested_fact_types": ["season", "closure"],
        "confidence": "high",
    }
    monkeypatch.setattr(app, "SOURCE_BACKED_INTERPRETER", lambda _prompt: json.dumps(payload))

    result = app.ask_source_backed_question(
        "im in mandocino. is it legal to harvest dungenese crabs?", ""
    )

    assert result["matched_place"] == "Mendocino"
    assert "dungeness crab" in (result["species_or_category"] or "").casefold()