coastwise / tests /test_source_extract.py
Stephen S. Lee
feat: source-derived Q&A with statewide CA ocean species coverage (#1)
c1b067e unverified
Raw
History Blame Contribute Delete
2.78 kB
"""Source extraction tests."""
from coastwise.schemas import RefreshStatus, SourceSnapshot
from coastwise.source_extract import (
content_hash,
default_structured_validation_set,
extract_structured_facts,
extract_source_chunks,
normalize_source_text,
tokenize_source_text,
)
from coastwise.source_cache import DEFAULT_SEED_CACHE_DIR, load_seed_cache
def test_source_text_normalization_hashing_and_chunk_creation():
text = "<html><body><h1>Lingcod</h1><p>Minimum size: 22 inches total length.</p></body></html>"
normalized = normalize_source_text(text)
assert normalized == "Lingcod Minimum size: 22 inches total length."
assert content_hash(normalized) == content_hash("Lingcod Minimum size: 22 inches total length.")
assert "lingcod" in tokenize_source_text(normalized)
snapshot = SourceSnapshot(
source_id="cdfw_ocean",
url="https://wildlife.ca.gov/Fishing/Ocean/Regulations",
title="CDFW Ocean Sport Fishing",
retrieved_at="2026-06-14T12:00:00+00:00",
content_hash=content_hash(normalized),
raw_text=normalized,
origin="seed",
refresh_status=RefreshStatus.UNCHANGED,
)
chunks = extract_source_chunks(snapshot)
assert chunks
assert chunks[0].source_url == snapshot.url
assert "lingcod" in chunks[0].tokens
def test_extract_structured_validation_set_facts_from_seed_chunks():
cache = load_seed_cache(DEFAULT_SEED_CACHE_DIR)
result = extract_structured_facts(cache.chunks, default_structured_validation_set())
fact_ids = {fact.id for fact in result.facts}
assert {"lingcod_min_size", "lingcod_bag_limit", "dungeness_crab_bag_limit"} <= fact_ids
assert not result.issues
def test_extract_structured_facts_finds_explicit_season_and_gear_facts():
snapshot = SourceSnapshot(
source_id="cdfw_crabs",
url="https://wildlife.ca.gov/Conservation/Marine/Invertebrates/Crabs",
title="CDFW Crabs",
retrieved_at="2026-06-14T12:00:00+00:00",
content_hash="crabs-test",
raw_text=(
"Dungeness crab season status is closed for the Mendocino ocean region. "
"Dungeness crab trap and hoop net rules apply before harvest."
),
origin="seed",
refresh_status=RefreshStatus.UNCHANGED,
)
chunks = extract_source_chunks(snapshot)
result = extract_structured_facts(chunks, default_structured_validation_set())
facts = {(fact.species_or_category, fact.fact_type.value): fact for fact in result.facts}
assert facts[("dungeness_crab", "season")].region_key == "mendocino"
assert "closed" in facts[("dungeness_crab", "season")].value
assert "trap" in facts[("dungeness_crab", "gear_or_method")].gear_or_method_notes