"""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()