Spaces:
Sleeping
Sleeping
File size: 3,257 Bytes
95941cd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """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
|