"""Schema tests.""" from datetime import date import pytest from coastwise.schemas import ( AnswerStatus, CoastalRegion, EvidenceType, FreshnessStatus, HarvestStatus, OfficialSource, OfficialSourceType, PhotoCandidate, PhotoCandidateResult, QueryContext, ReferenceImageSet, RefreshResult, RefreshStatus, RuleCard, SafetyResult, SourceBackedAnswer, SourceChunk, SourceSnapshot, StatewideSourceCoverage, StructuredRegulationFact, StructuredValidationSet, RequestedFactType, LocationContext, QuestionInterpretation, UserIntent, ) def test_controlled_vocabularies_use_specified_values(): assert HarvestStatus.DO_NOT_HARVEST.value == "do_not_harvest" assert HarvestStatus.VERIFY_BEFORE_HARVEST.value == "verify_before_harvest" assert HarvestStatus.RULES_APPLY.value == "rules_apply" assert HarvestStatus.OBSERVE_ONLY.value == "observe_only" assert UserIntent.THINKING_OF_HARVEST.value == "thinking_of_harvest" assert CoastalRegion.POINT_ARENA_TO_PIGEON_POINT.value == "point_arena_to_pigeon_point" assert OfficialSourceType.CDFW_OCEAN_REGULATIONS.value == "cdfw_ocean_regulations" assert RefreshStatus.UNCHANGED.value == "unchanged" assert FreshnessStatus.STALE.value == "stale" assert EvidenceType.SOURCE_SNIPPET.value == "source_snippet" assert RequestedFactType.MINIMUM_SIZE.value == "minimum_size" assert AnswerStatus.NO_CACHE.value == "no_cache" def test_source_backed_entities_keep_evidence_and_status_explicit(): source = OfficialSource( id="cdfw_ocean", title="CDFW Ocean Sport Fishing", source_type=OfficialSourceType.CDFW_OCEAN_REGULATIONS, url="https://wildlife.ca.gov/Fishing/Ocean/Regulations", scope="California ocean sport fishing", statewide_required=True, required_for_mvp=True, ) coverage = StatewideSourceCoverage( id="statewide_ca_ocean", required_region_keys=("northern", "san_francisco"), required_source_ids=("cdfw_ocean",), general_source_ids=("cdfw_ocean",), cdph_source_ids=("cdph_shellfish",), ) snapshot = SourceSnapshot( source_id=source.id, url=source.url, title=source.title, retrieved_at="2026-06-14T12:00:00+00:00", content_hash="abc123", raw_text="Lingcod minimum size is 22 inches total length.", origin="seed", refresh_status=RefreshStatus.UNCHANGED, ) chunk = SourceChunk( id="cdfw_ocean:lingcod:1", source_id=source.id, source_url=source.url, source_title=source.title, retrieved_at=snapshot.retrieved_at, heading="Lingcod", text="Lingcod minimum size is 22 inches total length.", tokens=("lingcod", "minimum", "size"), species_or_category_terms=("lingcod",), ) fact = StructuredRegulationFact( id="lingcod_min_size", species_or_category="lingcod", display_name="Lingcod", aliases=("ling cod",), area_or_scope="California ocean", fact_type=RequestedFactType.MINIMUM_SIZE, value="22", units="inches total length", source_id=source.id, source_url=source.url, retrieved_at=snapshot.retrieved_at, supporting_chunk_id=chunk.id, supporting_snippet=chunk.text, validation_set=True, ) validation_set = StructuredValidationSet( id="mvp_validation", required_questions=("what's the min size of lingcod?",), required_species_or_categories=("lingcod",), required_fact_types=(RequestedFactType.MINIMUM_SIZE,), required_sources=(source.id,), required_statewide_coverage_id=coverage.id, ) location = LocationContext( raw_question_location=None, raw_location_input="Pacifica", matched_place="Pacifica", area_or_scope="San Francisco coast", confidence="recognized", warning=None, region_key="san_francisco", ) interpretation = QuestionInterpretation( question="what's the min size of lingcod?", species_or_category="lingcod", requested_fact_type=RequestedFactType.MINIMUM_SIZE, location_context=location, scope="california_ocean", confidence="high", ) answer = SourceBackedAnswer( status=AnswerStatus.ANSWERED, direct_answer="Lingcod minimum size is 22 inches total length.", evidence_type=EvidenceType.STRUCTURED_FACT, species_or_category=fact.display_name, location_context=location, source_links=(source.url,), source_titles=(source.title,), retrieved_at=snapshot.retrieved_at, freshness_status=FreshnessStatus.CURRENT, refresh_status=RefreshStatus.UNCHANGED, supporting_snippets=(chunk.text,), uncertainty_notes=(), next_safe_action="Verify current CDFW guidance before harvest.", ) refresh = RefreshResult( started_at="2026-06-14T12:00:00+00:00", completed_at="2026-06-14T12:00:01+00:00", source_results={source.id: RefreshStatus.UNCHANGED}, updated_source_ids=(), unchanged_source_ids=(source.id,), failed_source_ids=(), cache_preserved_source_ids=(), user_message="Sources checked. No changes found.", ) assert validation_set.required_statewide_coverage_id == coverage.id assert interpretation.location_context.matched_place == "Pacifica" assert answer.source_links == (source.url,) assert refresh.source_results[source.id] is RefreshStatus.UNCHANGED def test_cached_evidence_packet_keeps_sources_and_excerpts_explicit(): from datetime import datetime, timezone from coastwise.schemas import CachedEvidencePacket, CachedEvidenceSource source = CachedEvidenceSource( source_id="cdfw_crabs", title="CDFW Crabs", url="https://wildlife.ca.gov/Conservation/Marine/Invertebrates/Crabs", retrieved_at=datetime(2026, 6, 14, 12, tzinfo=timezone.utc), freshness_status="current", excerpt="Saved CDFW crab information references Dungeness crab season.", ) packet = CachedEvidencePacket( question="is dungenese crab in season for fort Bragg", species_or_category="dungeness_crab", display_name="Dungeness crab", requested_fact_type=RequestedFactType.SEASON, matched_place="Fort Bragg", region_key="mendocino", sources=(source,), deterministic_next_safe_action="Use saved official source details as decision support; do not treat CoastWise as permission to harvest.", ) assert packet.sources[0].title == "CDFW Crabs" assert packet.requested_fact_type is RequestedFactType.SEASON def test_rule_card_requires_source_links_except_unknown_fallback(): with pytest.raises(ValueError, match="source_urls"): RuleCard( key="mussels", display_name="Mussels", category="shellfish", aliases=("mussel",), covered_species=("California mussel",), region_keys=("sf_coast_point_arena_to_pigeon_point",), photo_supported=True, default_status=HarvestStatus.VERIFY_BEFORE_HARVEST, short_answer="Verify before harvest.", rule_notes=("Check current rules.",), safety_warnings=(), source_names=("CDFW",), source_urls=(), cache_date=date(2026, 6, 1), ) unknown = RuleCard( key="unknown", display_name="Unknown organism", category="unknown", aliases=("unknown",), covered_species=(), region_keys=("sf_coast_point_arena_to_pigeon_point",), photo_supported=False, default_status=HarvestStatus.OBSERVE_ONLY, short_answer="Observe only.", rule_notes=(), safety_warnings=("Search by known name or verify with official sources.",), source_names=(), source_urls=(), cache_date=date(2026, 6, 1), ) assert unknown.default_status is HarvestStatus.OBSERVE_ONLY def test_reference_image_and_photo_candidate_boundaries_are_validated(): with pytest.raises(ValueError, match="reference_count"): ReferenceImageSet( key="mussels_refs", rule_card_key="mussels", display_name="Mussels", reference_count=0, source_or_permission_notes="Owned demo images.", representative_scope="Common mussel-like shellfish examples.", review_status="approved", limitations="Visual similarity only.", ) candidate = PhotoCandidate( rule_card_key="mussels", display_name="Mussels", confidence=0.74, match_basis="reference_similarity", reference_set_key="mussels_refs", alternatives=(), limitations="Visually similar, not confirmed ID.", ) result = PhotoCandidateResult( candidates=(candidate,), fallback_status=None, message="Visually similar candidate.", reference_set_keys=("mussels_refs",), image_retained=False, ) assert result.candidates[0].match_basis == "reference_similarity" with pytest.raises(ValueError, match="image_retained"): PhotoCandidateResult( candidates=(), fallback_status=HarvestStatus.OBSERVE_ONLY, message="Observe only.", reference_set_keys=(), image_retained=True, ) def test_query_context_and_safety_result_keep_sources_and_status_explicit(): context = QueryContext( intent=UserIntent.THINKING_OF_HARVEST, region=CoastalRegion.POINT_ARENA_TO_PIGEON_POINT, county_or_beach="Pacifica", today=date(2026, 6, 13), ) assert context.county_or_beach == "Pacifica" with pytest.raises(ValueError, match="source_links"): SafetyResult( status=HarvestStatus.RULES_APPLY, display_name="Lingcod", summary="Rules apply.", candidate_notice=None, rule_notes=("Check current size and bag rules.",), advisory_warnings=(), source_links=(), cache_dates=(date(2026, 6, 1),), stale=False, next_safe_action="Verify official rules before harvest.", ) source = OfficialSource( name="CDFW Ocean Sport Fishing", url="https://wildlife.ca.gov/Fishing/Ocean/Regulations", source_type="cdfw_regulation", retrieved_on=date(2026, 6, 1), ) assert source.url.startswith("https://") def test_model_question_interpretation_validates_allowlisted_shape(): from coastwise.schemas import ModelQuestionInterpretation, SourceBackedIntent interpretation = ModelQuestionInterpretation( species_or_category="dungeness_crab", location_name="Mendocino", region_key="mendocino", intent="permission_status", requested_fact_types=("season", "closure"), confidence="high", notes="Normalized typo species and location.", ) assert interpretation.intent is SourceBackedIntent.PERMISSION_STATUS assert interpretation.requested_fact_types == ( RequestedFactType.SEASON, RequestedFactType.CLOSURE, ) assert interpretation.confidence == "high" assert interpretation.species_or_category == "dungeness_crab" assert interpretation.location_name == "Mendocino" assert interpretation.region_key == "mendocino" assert interpretation.notes == "Normalized typo species and location." def test_model_question_interpretation_rejects_unsupported_confidence(): from coastwise.schemas import ModelQuestionInterpretation, SourceBackedIntent with pytest.raises(ValueError, match="model interpretation confidence"): ModelQuestionInterpretation( species_or_category=None, location_name=None, region_key=None, intent=SourceBackedIntent.GENERAL, requested_fact_types=(RequestedFactType.GENERAL_SUMMARY,), confidence="certain", )