Spaces:
Running
Running
| """Source cache tests.""" | |
| from dataclasses import replace | |
| from datetime import datetime, timedelta, timezone | |
| from coastwise.source_cache import ( | |
| DEFAULT_SEED_CACHE_DIR, | |
| SourceCache, | |
| cache_status_summary, | |
| freshness_status, | |
| load_seed_cache, | |
| load_latest_cache, | |
| validate_seed_cache, | |
| write_runtime_cache, | |
| ) | |
| from coastwise.schemas import FreshnessStatus, RefreshStatus, SourceSnapshot | |
| from coastwise.source_registry import load_official_sources, statewide_source_coverage | |
| def test_seed_cache_loads_and_covers_required_statewide_sources(): | |
| cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) | |
| issues = validate_seed_cache(cache, load_official_sources(), statewide_source_coverage()) | |
| assert not [issue for issue in issues if issue.severity == "error"] | |
| assert "lingcod_min_size" in cache.facts_by_id | |
| assert "dungeness_crab_bag_limit" in cache.facts_by_id | |
| assert not hasattr(cache, "question_history") | |
| assert not hasattr(cache, "location_history") | |
| def test_freshness_status_uses_24_hour_threshold(): | |
| now = datetime(2026, 6, 14, 12, tzinfo=timezone.utc) | |
| assert freshness_status(now - timedelta(hours=24), now) is FreshnessStatus.CURRENT | |
| assert freshness_status(now - timedelta(hours=24, seconds=1), now) is FreshnessStatus.STALE | |
| def test_runtime_cache_overlays_valid_snapshot_and_preserves_seed_on_failure(tmp_path): | |
| seed = load_seed_cache(DEFAULT_SEED_CACHE_DIR) | |
| original = seed.snapshots_by_source_id["cdfw_ocean_regulations"] | |
| replacement = SourceSnapshot( | |
| source_id=original.source_id, | |
| url=original.url, | |
| title=original.title, | |
| retrieved_at="2026-06-14T13:00:00+00:00", | |
| content_hash="new-hash", | |
| raw_text="Updated official source text", | |
| origin="refresh", | |
| refresh_status=RefreshStatus.UPDATED, | |
| ) | |
| write_runtime_cache(tmp_path, seed, snapshots=(replacement,)) | |
| latest = load_latest_cache(seed, tmp_path) | |
| assert latest.snapshots_by_source_id[original.source_id].content_hash == "new-hash" | |
| preserved = load_latest_cache(seed, tmp_path) | |
| assert preserved.snapshots_by_source_id["cdfw_groundfish_summary"].content_hash == seed.snapshots_by_source_id["cdfw_groundfish_summary"].content_hash | |
| def test_cache_status_summary_reports_seed_runtime_stale_and_no_cache(): | |
| 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, | |
| ) | |
| assert cache_status_summary(seed, now)["usable_cache"] is True | |
| assert cache_status_summary(stale, now)["freshness_status"] == "stale" | |
| assert cache_status_summary(SourceCache((), (), (), ()), now)["freshness_status"] == "no_cache" | |
| def test_seed_cache_contains_real_query_saved_evidence_chunks(): | |
| cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR) | |
| text = " ".join(chunk.text.casefold() for chunk in cache.chunks) | |
| for term in ("dungeness crab", "rock crab", "cabezon", "half moon bay", "pacifica"): | |
| assert term in text | |