diff --git a/chronicle/jurisdictions/__init__.py b/chronicle/jurisdictions/__init__.py index 91d5fe0..28ea239 100644 --- a/chronicle/jurisdictions/__init__.py +++ b/chronicle/jurisdictions/__init__.py @@ -1,3 +1,3 @@ """Jurisdiction-specific Chronicle loaders and source adapters.""" -__all__ = ["uk", "us"] +__all__ = ["be", "uk", "us"] diff --git a/chronicle/jurisdictions/be/__init__.py b/chronicle/jurisdictions/be/__init__.py new file mode 100644 index 0000000..eb52173 --- /dev/null +++ b/chronicle/jurisdictions/be/__init__.py @@ -0,0 +1,15 @@ +"""Belgium-specific publisher geography sources and adapters.""" + +from chronicle.jurisdictions.be.geography import ( + NISCodeCrosswalk, + NISCodeTranslation, + NISCrosswalkError, + NISCrosswalkLookupError, +) + +__all__ = [ + "NISCodeCrosswalk", + "NISCodeTranslation", + "NISCrosswalkError", + "NISCrosswalkLookupError", +] diff --git a/chronicle/jurisdictions/be/geography.py b/chronicle/jurisdictions/be/geography.py new file mode 100644 index 0000000..c6dbcaf --- /dev/null +++ b/chronicle/jurisdictions/be/geography.py @@ -0,0 +1,180 @@ +"""Publisher-backed NIS geography translations for Belgian source facts. + +Chronicle preserves the geography identities asserted by Statbel. Consumers +remain responsible for selecting facts and enforcing their own geography-vintage +join contracts. +""" + +from __future__ import annotations + +import csv +from dataclasses import dataclass +from datetime import date +from pathlib import Path +from typing import Iterable + + +class NISCrosswalkError(ValueError): + """Raised when a NIS crosswalk is malformed or internally ambiguous.""" + + +class NISCrosswalkLookupError(NISCrosswalkError): + """Raised when a requested NIS translation is not publisher-declared.""" + + +@dataclass(frozen=True) +class NISCodeTranslation: + """One publisher-declared NIS identity translation, without fact values.""" + + source_nis: str + source_name: str + source_vintage: str + target_nis: str + target_name: str + target_vintage: str + effective_date: date + relationship: str + source_url: str + + +class NISCodeCrosswalk: + """A validated, immutable index of publisher-declared NIS code changes.""" + + def __init__(self, rows: Iterable[NISCodeTranslation]) -> None: + translations = tuple(rows) + if not translations: + raise NISCrosswalkError("NIS crosswalk must contain at least one row") + + by_source: dict[tuple[str, str, str], NISCodeTranslation] = {} + for row in translations: + _validate_translation(row) + key = (row.source_vintage, row.target_vintage, row.source_nis) + prior = by_source.get(key) + if prior is not None: + raise NISCrosswalkError( + "Duplicate NIS crosswalk mapping for " + f"{row.source_nis!r} from {row.source_vintage!r} to " + f"{row.target_vintage!r}" + ) + by_source[key] = row + + self._rows = translations + self._by_source = by_source + + @classmethod + def from_csv(cls, path: str | Path) -> NISCodeCrosswalk: + """Load and validate the complete publisher-backed crosswalk CSV.""" + + input_path = Path(path) + with input_path.open(newline="", encoding="utf-8") as stream: + reader = csv.DictReader(stream) + required = { + "source_nis", + "source_name", + "source_vintage", + "target_nis", + "target_name", + "target_vintage", + "effective_date", + "relationship", + "source_url", + } + missing = sorted(required - set(reader.fieldnames or ())) + if missing: + raise NISCrosswalkError( + "NIS crosswalk is missing columns: " + ", ".join(missing) + ) + rows = [] + for line_number, payload in enumerate(reader, start=2): + try: + rows.append( + NISCodeTranslation( + source_nis=payload["source_nis"].strip(), + source_name=payload["source_name"].strip(), + source_vintage=payload["source_vintage"].strip(), + target_nis=payload["target_nis"].strip(), + target_name=payload["target_name"].strip(), + target_vintage=payload["target_vintage"].strip(), + effective_date=date.fromisoformat( + payload["effective_date"].strip() + ), + relationship=payload["relationship"].strip(), + source_url=payload["source_url"].strip(), + ) + ) + except (AttributeError, KeyError, TypeError, ValueError) as error: + raise NISCrosswalkError( + f"Invalid NIS crosswalk row at line {line_number}: {error}" + ) from error + return cls(rows) + + @property + def rows(self) -> tuple[NISCodeTranslation, ...]: + """Return all validated publisher rows in source order.""" + + return self._rows + + def translate( + self, + source_nis: str, + *, + source_vintage: str, + target_vintage: str, + ) -> NISCodeTranslation: + """Return one declared code translation or fail on missing coverage.""" + + key = (source_vintage, target_vintage, source_nis) + try: + return self._by_source[key] + except KeyError as error: + raise NISCrosswalkLookupError( + "No NIS crosswalk row for " + f"{source_nis!r} from {source_vintage!r} to {target_vintage!r}" + ) from error + + def translation_plan( + self, + source_ids: Iterable[str], + *, + source_vintage: str, + target_vintage: str, + ) -> tuple[NISCodeTranslation, ...]: + """Compile identity translations without reconciling or summing values.""" + + return tuple( + self.translate( + source_id, + source_vintage=source_vintage, + target_vintage=target_vintage, + ) + for source_id in source_ids + ) + + +def _validate_translation(row: NISCodeTranslation) -> None: + for field_name in ("source_nis", "target_nis"): + code = getattr(row, field_name) + if len(code) != 5 or not code.isascii() or not code.isdigit(): + raise NISCrosswalkError( + f"{field_name} must be a five-digit NIS code: {code!r}" + ) + for field_name in ( + "source_name", + "source_vintage", + "target_name", + "target_vintage", + ): + if not getattr(row, field_name).strip(): + raise NISCrosswalkError(f"{field_name} must be non-empty") + if row.relationship not in {"merged", "unchanged"}: + raise NISCrosswalkError(f"Unsupported NIS relationship: {row.relationship!r}") + if row.relationship == "unchanged" and row.source_nis != row.target_nis: + raise NISCrosswalkError( + "An unchanged NIS relationship must preserve the code: " + f"{row.source_nis!r} -> {row.target_nis!r}" + ) + if not row.source_url.startswith("https://statbel.fgov.be/"): + raise NISCrosswalkError( + "NIS crosswalk rows must cite a public Statbel publisher URL: " + f"{row.source_url!r}" + ) diff --git a/chronicle/sources/__init__.py b/chronicle/sources/__init__.py index 6dde257..d842219 100644 --- a/chronicle/sources/__init__.py +++ b/chronicle/sources/__init__.py @@ -17,6 +17,15 @@ validate_source_cells, ) from .models import SourceFile, SourceReference +from .offline_fetch import ( + OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION, + OfflineFetchArtifact, + OfflineFetchManifest, + OfflineFetchManifestError, + OfflineFetchR2Location, + load_offline_fetch_manifest, + validate_offline_fetch_manifest, +) from .specs import ( CellSelectorSpec, SourceRecord, @@ -35,6 +44,11 @@ __all__ = [ "CellSelectorSpec", + "OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION", + "OfflineFetchArtifact", + "OfflineFetchManifest", + "OfflineFetchManifestError", + "OfflineFetchR2Location", "SourceArtifactMetadata", "SourceArtifact", "SourceCell", @@ -55,6 +69,7 @@ "compile_source_record_set_specs", "decode_delimited_text", "load_source_cells_jsonl", + "load_offline_fetch_manifest", "query_sources", "resolve_cell_selector", "resolve_source_record", @@ -66,4 +81,5 @@ "source_cells_from_xlsx", "source_regions_from_record_set_spec", "validate_source_cells", + "validate_offline_fetch_manifest", ] diff --git a/chronicle/sources/offline_fetch.py b/chronicle/sources/offline_fetch.py new file mode 100644 index 0000000..6686203 --- /dev/null +++ b/chronicle/sources/offline_fetch.py @@ -0,0 +1,643 @@ +"""Validation for deterministic offline source-artifact fetch manifests.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from ipaddress import ip_address +import json +from pathlib import Path, PurePosixPath +import re +from typing import Any, NoReturn +from urllib.parse import urlsplit + + +OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION = "ledger.offline_fetch_manifest.v1" + +_LOWERCASE_SHA256 = re.compile(r"[0-9a-f]{64}") +_MISSING = object() +_MANIFEST_FIELDS = frozenset( + {"schema_version", "generated_for", "artifacts", "final_validation"} +) +_ARTIFACT_FIELDS = frozenset( + { + "package_id", + "url", + "expected_filename", + "destination_path", + "manifest_path", + "manifest_year", + "discovery_note", + "post_download_steps", + "expected_sha256", + "source_package_path", + # Existing v1 extensions used by FETCH-MANIFEST.json. + "r2", + "replaces_synthetic_fixture", + "already_archived", + "archive_member", + "note_new_vintage_sha256", + } +) +_R2_FIELDS = frozenset( + {"provider", "bucket", "key", "uri", "key_template", "uri_template"} +) + + +class OfflineFetchManifestError(ValueError): + """Raised when an offline fetch manifest violates its deterministic contract.""" + + +@dataclass(frozen=True) +class OfflineFetchR2Location: + """A validated R2 location or content-addressed location template.""" + + provider: str + bucket: str + key: str | None + uri: str | None + key_template: str | None + uri_template: str | None + + +@dataclass(frozen=True) +class OfflineFetchArtifact: + """One publisher artifact requested through an offline handoff.""" + + package_id: str + url: str + expected_filename: str + destination_path: str + manifest_path: str + manifest_year: int + discovery_note: str | None + post_download_steps: tuple[str, ...] + expected_sha256: str | None = None + source_package_path: str | None = None + r2: OfflineFetchR2Location | None = None + replaces_synthetic_fixture: bool | None = None + already_archived: bool | None = None + archive_member: str | None = None + note_new_vintage_sha256: str | None = None + + +@dataclass(frozen=True) +class OfflineFetchManifest: + """A validated offline publisher-artifact fetch handoff.""" + + schema_version: str + generated_for: str + artifacts: tuple[OfflineFetchArtifact, ...] + final_validation: tuple[str, ...] = () + + +def load_offline_fetch_manifest( + path: str | Path, + *, + require_discovery_notes: bool = False, +) -> OfflineFetchManifest: + """Load and validate an offline fetch manifest from JSON.""" + + manifest_path = Path(path) + try: + payload = json.loads(manifest_path.read_text(encoding="utf-8")) + except OSError as exc: + raise OfflineFetchManifestError( + f"Could not read offline fetch manifest {manifest_path}: {exc}" + ) from exc + except json.JSONDecodeError as exc: + raise OfflineFetchManifestError( + f"Offline fetch manifest {manifest_path} is not valid JSON: {exc}" + ) from exc + + return validate_offline_fetch_manifest( + payload, + source=str(manifest_path), + require_discovery_notes=require_discovery_notes, + ) + + +def validate_offline_fetch_manifest( + payload: object, + *, + source: str = "offline fetch manifest", + require_discovery_notes: bool = False, +) -> OfflineFetchManifest: + """Validate a parsed ``ledger.offline_fetch_manifest.v1`` mapping.""" + + if not isinstance(payload, Mapping): + _fail(source, "$", "must be a JSON object") + _reject_unknown_fields(payload, _MANIFEST_FIELDS, source=source, location="$") + + schema_version = _required_string(payload, "schema_version", source=source) + if schema_version != OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION: + _fail( + source, + "schema_version", + f"must be {OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION!r}", + ) + + generated_for = _required_string(payload, "generated_for", source=source) + raw_artifacts = payload.get("artifacts") + if not isinstance(raw_artifacts, list) or not raw_artifacts: + _fail(source, "artifacts", "must be a nonempty list") + final_validation = _optional_string_list( + payload, + "final_validation", + source=source, + parent=None, + ) + + artifacts: list[OfflineFetchArtifact] = [] + seen_destinations: dict[str, int] = {} + seen_manifest_paths: dict[str, int] = {} + for index, raw_artifact in enumerate(raw_artifacts): + location = f"artifacts[{index}]" + if not isinstance(raw_artifact, Mapping): + _fail(source, location, "must be a JSON object") + _reject_unknown_fields( + raw_artifact, + _ARTIFACT_FIELDS, + source=source, + location=location, + ) + + package_id = _required_string( + raw_artifact, + "package_id", + source=source, + parent=location, + ) + url = _https_url(raw_artifact, source=source, parent=location) + expected_filename = _expected_filename( + raw_artifact, + source=source, + parent=location, + ) + destination_path = _repo_path( + raw_artifact, + "destination_path", + source=source, + parent=location, + required_prefix=("db", "data"), + ) + if PurePosixPath(destination_path).name != expected_filename: + _fail( + source, + f"{location}.destination_path", + "must end with expected_filename", + ) + manifest_path = _repo_path( + raw_artifact, + "manifest_path", + source=source, + parent=location, + required_prefix=("db", "data"), + ) + if not manifest_path.endswith(".yaml"): + _fail( + source, + f"{location}.manifest_path", + "must identify a YAML manifest", + ) + if ( + PurePosixPath(manifest_path).parent + != PurePosixPath(destination_path).parent + ): + _fail( + source, + f"{location}.manifest_path", + "must be beside destination_path", + ) + if destination_path == manifest_path: + _fail( + source, + f"{location}.destination_path", + "must not overwrite manifest_path", + ) + manifest_year = raw_artifact.get("manifest_year", _MISSING) + if type(manifest_year) is not int: + _fail( + source, + f"{location}.manifest_year", + "must be an integer", + ) + + discovery_note = _optional_string( + raw_artifact, + "discovery_note", + source=source, + parent=location, + required=require_discovery_notes, + ) + post_download_steps = _nonempty_string_list( + raw_artifact, + "post_download_steps", + source=source, + parent=location, + ) + expected_sha256 = _optional_sha256( + raw_artifact, + "expected_sha256", + source=source, + parent=location, + ) + + source_package_path = None + if "source_package_path" in raw_artifact: + source_package_path = _repo_path( + raw_artifact, + "source_package_path", + source=source, + parent=location, + required_prefix=("packages",), + required_basename="source_package.yaml", + ) + + r2 = _optional_r2(raw_artifact, source=source, parent=location) + replaces_synthetic_fixture = _optional_bool( + raw_artifact, + "replaces_synthetic_fixture", + source=source, + parent=location, + ) + already_archived = _optional_bool( + raw_artifact, + "already_archived", + source=source, + parent=location, + ) + archive_member = _optional_basename( + raw_artifact, + "archive_member", + source=source, + parent=location, + ) + note_new_vintage_sha256 = _optional_sha256( + raw_artifact, + "note_new_vintage_sha256", + source=source, + parent=location, + ) + + prior_index = seen_destinations.get(destination_path) + if prior_index is not None: + _fail( + source, + location, + (f"duplicates destination_path from artifacts[{prior_index}]"), + ) + prior_manifest_index = seen_manifest_paths.get(destination_path) + if prior_manifest_index is not None: + _fail( + source, + f"{location}.destination_path", + ( + "must not overwrite manifest_path from " + f"artifacts[{prior_manifest_index}]" + ), + ) + prior_destination_index = seen_destinations.get(manifest_path) + if prior_destination_index is not None: + _fail( + source, + f"{location}.manifest_path", + ( + "must not identify destination_path from " + f"artifacts[{prior_destination_index}]" + ), + ) + seen_destinations[destination_path] = index + seen_manifest_paths.setdefault(manifest_path, index) + + artifacts.append( + OfflineFetchArtifact( + package_id=package_id, + url=url, + expected_filename=expected_filename, + destination_path=destination_path, + manifest_path=manifest_path, + manifest_year=manifest_year, + discovery_note=discovery_note, + post_download_steps=post_download_steps, + expected_sha256=expected_sha256, + source_package_path=source_package_path, + r2=r2, + replaces_synthetic_fixture=replaces_synthetic_fixture, + already_archived=already_archived, + archive_member=archive_member, + note_new_vintage_sha256=note_new_vintage_sha256, + ) + ) + + return OfflineFetchManifest( + schema_version=schema_version, + generated_for=generated_for, + artifacts=tuple(artifacts), + final_validation=final_validation, + ) + + +def _required_string( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str | None = None, +) -> str: + location = f"{parent}.{field}" if parent else field + value = payload.get(field, _MISSING) + if not isinstance(value, str) or not value.strip(): + _fail(source, location, "must be a nonempty string") + return value + + +def _optional_string( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str, + required: bool, +) -> str | None: + if field not in payload: + if required: + _fail(source, f"{parent}.{field}", "must be a nonempty string") + return None + return _required_string(payload, field, source=source, parent=parent) + + +def _https_url( + payload: Mapping[str, Any], + *, + source: str, + parent: str, +) -> str: + url = _required_string(payload, "url", source=source, parent=parent) + if _has_control_characters(url): + _fail(source, f"{parent}.url", "must not contain control characters") + try: + parsed = urlsplit(url) + _ = parsed.port + except ValueError as exc: + _fail(source, f"{parent}.url", f"must be a valid HTTPS URL: {exc}") + if ( + parsed.scheme != "https" + or not parsed.netloc + or not parsed.hostname + or any(character.isspace() for character in url) + ): + _fail(source, f"{parent}.url", "must be an HTTPS URL with a host") + if parsed.username is not None or parsed.password is not None: + _fail(source, f"{parent}.url", "must not contain user information") + + hostname = parsed.hostname.rstrip(".").casefold() + if hostname == "localhost" or hostname.endswith(".localhost"): + _fail(source, f"{parent}.url", "must use a public publisher host") + try: + address = ip_address(hostname) + except ValueError: + pass + else: + if not address.is_global: + _fail(source, f"{parent}.url", "must use a public IP address") + return url + + +def _expected_filename( + payload: Mapping[str, Any], + *, + source: str, + parent: str, +) -> str: + filename = _required_string( + payload, + "expected_filename", + source=source, + parent=parent, + ) + if _has_control_characters(filename): + _fail( + source, + f"{parent}.expected_filename", + "must not contain control characters", + ) + if ( + "/" in filename + or "\\" in filename + or filename in {".", ".."} + or PurePosixPath(filename).name != filename + ): + _fail(source, f"{parent}.expected_filename", "must be a basename") + return filename + + +def _repo_path( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str, + required_prefix: tuple[str, ...], + required_basename: str | None = None, +) -> str: + value = _required_string(payload, field, source=source, parent=parent) + location = f"{parent}.{field}" + if _has_control_characters(value): + _fail(source, location, "must not contain control characters") + if "\\" in value: + _fail(source, location, "must use a relative POSIX repository path") + + raw_parts = value.split("/") + path = PurePosixPath(value) + if path.is_absolute(): + _fail(source, location, "must be relative") + if any(part in {"", ".", ".."} for part in raw_parts): + _fail(source, location, "must not contain empty or traversal segments") + if path.parts[: len(required_prefix)] != required_prefix: + required_root = "/".join(required_prefix) + _fail(source, location, f"must be under {required_root}/") + if len(path.parts) <= len(required_prefix): + _fail(source, location, "must identify a file below the required root") + if required_basename is not None and path.name != required_basename: + _fail(source, location, f"must end with {required_basename}") + return path.as_posix() + + +def _nonempty_string_list( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str | None, +) -> tuple[str, ...]: + value = payload.get(field, _MISSING) + location = f"{parent}.{field}" if parent else field + if not isinstance(value, list) or not value: + _fail(source, location, "must be a nonempty list") + for index, item in enumerate(value): + if not isinstance(item, str) or not item.strip(): + _fail(source, f"{location}[{index}]", "must be a nonempty string") + return tuple(value) + + +def _optional_sha256( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str, +) -> str | None: + if field not in payload: + return None + value = payload[field] + location = f"{parent}.{field}" + if ( + not isinstance(value, str) + or _LOWERCASE_SHA256.fullmatch(value) is None + or value == "0" * 64 + ): + _fail( + source, + location, + "must be a nonzero lowercase 64-character hexadecimal SHA-256", + ) + return value + + +def _optional_string_list( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str | None, +) -> tuple[str, ...]: + if field not in payload: + return () + return _nonempty_string_list( + payload, + field, + source=source, + parent=parent, + ) + + +def _optional_bool( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str, +) -> bool | None: + if field not in payload: + return None + value = payload[field] + if type(value) is not bool: + _fail(source, f"{parent}.{field}", "must be a boolean") + return value + + +def _optional_basename( + payload: Mapping[str, Any], + field: str, + *, + source: str, + parent: str, +) -> str | None: + if field not in payload: + return None + value = _required_string(payload, field, source=source, parent=parent) + if ( + _has_control_characters(value) + or "/" in value + or "\\" in value + or value in {".", ".."} + or PurePosixPath(value).name != value + ): + _fail(source, f"{parent}.{field}", "must be a safe basename") + return value + + +def _optional_r2( + payload: Mapping[str, Any], + *, + source: str, + parent: str, +) -> OfflineFetchR2Location | None: + if "r2" not in payload: + return None + value = payload["r2"] + location = f"{parent}.r2" + if not isinstance(value, Mapping): + _fail(source, location, "must be a JSON object") + _reject_unknown_fields(value, _R2_FIELDS, source=source, location=location) + + provider = _required_string(value, "provider", source=source, parent=location) + if provider != "r2": + _fail(source, f"{location}.provider", "must be 'r2'") + bucket = _required_string(value, "bucket", source=source, parent=location) + key = _optional_string(value, "key", source=source, parent=location, required=False) + uri = _optional_string(value, "uri", source=source, parent=location, required=False) + key_template = _optional_string( + value, + "key_template", + source=source, + parent=location, + required=False, + ) + uri_template = _optional_string( + value, + "uri_template", + source=source, + parent=location, + required=False, + ) + + if (key is None) != (uri is None): + _fail(source, location, "must provide key and uri together") + if (key_template is None) != (uri_template is None): + _fail(source, location, "must provide key_template and uri_template together") + if key is None and key_template is None: + _fail(source, location, "must provide a location or location template") + for field_name, item in ( + ("bucket", bucket), + ("key", key), + ("uri", uri), + ("key_template", key_template), + ("uri_template", uri_template), + ): + if item is not None and _has_control_characters(item): + _fail( + source, + f"{location}.{field_name}", + "must not contain control characters", + ) + + return OfflineFetchR2Location( + provider=provider, + bucket=bucket, + key=key, + uri=uri, + key_template=key_template, + uri_template=uri_template, + ) + + +def _reject_unknown_fields( + payload: Mapping[Any, Any], + allowed: frozenset[str], + *, + source: str, + location: str, +) -> None: + unknown = sorted(repr(field) for field in payload if field not in allowed) + if unknown: + _fail(source, location, "contains unknown fields: " + ", ".join(unknown)) + + +def _has_control_characters(value: str) -> bool: + return any(ord(character) < 32 or ord(character) == 127 for character in value) + + +def _fail(source: str, location: str, message: str) -> NoReturn: + raise OfflineFetchManifestError(f"{source}: {location} {message}") diff --git a/tests/test_chronicle_belgium_geography.py b/tests/test_chronicle_belgium_geography.py new file mode 100644 index 0000000..d2a432c --- /dev/null +++ b/tests/test_chronicle_belgium_geography.py @@ -0,0 +1,91 @@ +"""Tests for Belgium's publisher-backed NIS crosswalk contract.""" + +from __future__ import annotations + +from dataclasses import replace +from pathlib import Path + +import pytest + +from chronicle.jurisdictions.be import ( + NISCodeCrosswalk, + NISCrosswalkError, + NISCrosswalkLookupError, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] +CROSSWALK_PATH = ( + REPO_ROOT + / "db" + / "data" + / "statbel" + / "nis_2025_commune_crosswalk" + / "statbel_nis_2025_commune_crosswalk.csv" +) + + +def test_nis_crosswalk_loads_publisher_rows_and_merged_codes(): + crosswalk = NISCodeCrosswalk.from_csv(CROSSWALK_PATH) + + assert len(crosswalk.rows) == 581 + assert { + row.source_nis + for row in crosswalk.rows + if row.target_nis == "82039" and row.relationship == "merged" + } == {"82003", "82005"} + assert ( + crosswalk.translate( + "11007", + source_vintage="nis_2019_2024", + target_vintage="nis_2025", + ).target_nis + == "11002" + ) + + +def test_nis_translation_plan_preserves_many_to_one_identity_rows(): + crosswalk = NISCodeCrosswalk.from_csv(CROSSWALK_PATH) + + plan = crosswalk.translation_plan( + ["11056", "46003", "46013"], + source_vintage="nis_2019_2024", + target_vintage="nis_2025", + ) + + assert [row.source_nis for row in plan] == ["11056", "46003", "46013"] + assert [row.target_nis for row in plan] == ["46030", "46030", "46030"] + + +def test_nis_translation_fails_when_crosswalk_has_no_declared_row(): + crosswalk = NISCodeCrosswalk.from_csv(CROSSWALK_PATH) + + with pytest.raises(NISCrosswalkLookupError, match="No NIS crosswalk row"): + crosswalk.translate( + "99999", + source_vintage="nis_2019_2024", + target_vintage="nis_2025", + ) + + +def test_nis_crosswalk_reports_truncated_csv_rows_as_contract_errors(tmp_path): + path = tmp_path / "truncated.csv" + path.write_text( + "source_nis,source_name,source_vintage,target_nis,target_name," + "target_vintage,effective_date,relationship,source_url\n" + "11001,Aartselaar,nis_2019_2024,11001,Aartselaar,nis_2025," + "2019-01-01,unchanged\n", + encoding="utf-8", + ) + + with pytest.raises(NISCrosswalkError, match="Invalid.*line 2"): + NISCodeCrosswalk.from_csv(path) + + +@pytest.mark.parametrize("conflicting", [False, True]) +def test_nis_crosswalk_rejects_every_duplicate_source_key(conflicting): + row = NISCodeCrosswalk.from_csv(CROSSWALK_PATH).rows[0] + duplicate = replace(row, target_name="Conflicting name") if conflicting else row + + with pytest.raises(NISCrosswalkError, match="Duplicate NIS crosswalk mapping"): + NISCodeCrosswalk([row, duplicate]) diff --git a/tests/test_chronicle_offline_fetch.py b/tests/test_chronicle_offline_fetch.py new file mode 100644 index 0000000..6e39991 --- /dev/null +++ b/tests/test_chronicle_offline_fetch.py @@ -0,0 +1,299 @@ +"""Tests for deterministic offline artifact-fetch handoffs.""" + +from __future__ import annotations + +from copy import deepcopy +import json +from pathlib import Path + +import pytest + +from chronicle.sources.offline_fetch import ( + OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION, + OfflineFetchManifestError, + load_offline_fetch_manifest, + validate_offline_fetch_manifest, +) + + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def _artifact() -> dict[str, object]: + return { + "package_id": "statbel-fiscal-income-2023", + "url": "https://statbel.fgov.be/open-data/fiscal-income-2023.csv", + "expected_filename": "fiscal-income-2023.csv", + "destination_path": ( + "db/data/statbel/fiscal_income_2023/fiscal-income-2023.csv" + ), + "manifest_path": "db/data/statbel/fiscal_income_2023/manifest.yaml", + "manifest_year": 2023, + "discovery_note": "Resolve through the Statbel machine-readable catalog.", + "source_package_path": ( + "packages/statbel/fiscal_income_2023/source_package.yaml" + ), + "post_download_steps": [ + "Compute and record the publisher artifact SHA-256.", + "Run the package validator.", + ], + } + + +def _manifest() -> dict[str, object]: + return { + "schema_version": OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION, + "generated_for": "Open Belgian aggregate source packages", + "artifacts": [_artifact()], + } + + +def test_load_accepts_an_omitted_unknown_sha256(tmp_path): + path = tmp_path / "FETCH-MANIFEST-BE.json" + path.write_text(json.dumps(_manifest()), encoding="utf-8") + + manifest = load_offline_fetch_manifest(path, require_discovery_notes=True) + + assert manifest.schema_version == OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION + assert manifest.generated_for == "Open Belgian aggregate source packages" + assert len(manifest.artifacts) == 1 + assert manifest.artifacts[0].expected_sha256 is None + assert manifest.artifacts[0].manifest_year == 2023 + assert manifest.artifacts[0].post_download_steps == ( + "Compute and record the publisher artifact SHA-256.", + "Run the package validator.", + ) + + +def test_existing_v1_fetch_manifest_remains_valid(): + manifest = load_offline_fetch_manifest(REPO_ROOT / "FETCH-MANIFEST.json") + + assert len(manifest.artifacts) == 3 + assert manifest.final_validation + assert manifest.artifacts[0].r2 is not None + assert manifest.artifacts[0].r2.bucket == "ledger-raw" + assert manifest.artifacts[0].replaces_synthetic_fixture is True + assert manifest.artifacts[-1].already_archived is True + assert manifest.artifacts[-1].archive_member == "FY25.xlsx" + assert manifest.artifacts[-1].manifest_path.endswith( + "manifest_fy2025_monthly_source_package.yaml" + ) + + +def test_discovery_notes_are_optional_by_default_but_can_be_required(): + payload = _manifest() + del payload["artifacts"][0]["discovery_note"] + + manifest = validate_offline_fetch_manifest(payload) + + assert manifest.artifacts[0].discovery_note is None + with pytest.raises(OfflineFetchManifestError, match="discovery_note"): + validate_offline_fetch_manifest(payload, require_discovery_notes=True) + + +@pytest.mark.parametrize( + "sha256", + [ + "pending", + "0" * 64, + "A" * 64, + ], +) +def test_rejects_unpinned_sha256_sentinels_and_nonlowercase_hashes(sha256): + payload = _manifest() + payload["artifacts"][0]["expected_sha256"] = sha256 + + with pytest.raises(OfflineFetchManifestError, match="expected_sha256"): + validate_offline_fetch_manifest(payload) + + +def test_accepts_a_lowercase_nonzero_sha256(): + payload = _manifest() + payload["artifacts"][0]["expected_sha256"] = "a1" * 32 + + manifest = validate_offline_fetch_manifest(payload) + + assert manifest.artifacts[0].expected_sha256 == "a1" * 32 + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("destination_path", "/tmp/publisher.csv"), + ("destination_path", "db/data/statbel/../publisher.csv"), + ("manifest_path", "/db/data/statbel/manifest.yaml"), + ("manifest_path", "db/data/statbel/../../manifest.yaml"), + ("source_package_path", "/packages/statbel/source_package.yaml"), + ( + "source_package_path", + "packages/statbel/../source_package.yaml", + ), + ], +) +def test_rejects_absolute_and_traversing_repository_paths(field, value): + payload = _manifest() + payload["artifacts"][0][field] = value + + with pytest.raises(OfflineFetchManifestError, match=field): + validate_offline_fetch_manifest(payload) + + +@pytest.mark.parametrize( + "url", + [ + "https://user:password@example.com/data.csv", + "https://localhost/data.csv", + "https://publisher.localhost/data.csv", + "https://127.0.0.1/data.csv", + "https://10.0.0.1/data.csv", + "https://[::1]/data.csv", + ], +) +def test_rejects_userinfo_and_nonpublic_fetch_hosts(url): + payload = _manifest() + payload["artifacts"][0]["url"] = url + + with pytest.raises(OfflineFetchManifestError, match="url"): + validate_offline_fetch_manifest(payload) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("expected_filename", "publisher\x00.csv"), + ( + "destination_path", + "db/data/statbel/fiscal_income_2023/publisher\n.csv", + ), + ( + "manifest_path", + "db/data/statbel/fiscal_income_2023/manifest\x7f.yaml", + ), + ( + "source_package_path", + "packages/statbel/fiscal_income_2023\x00/source_package.yaml", + ), + ], +) +def test_rejects_control_characters_in_filenames_and_paths(field, value): + payload = _manifest() + payload["artifacts"][0][field] = value + + with pytest.raises(OfflineFetchManifestError, match=field): + validate_offline_fetch_manifest(payload) + + +def test_rejects_unknown_manifest_and_artifact_fields(): + payload = _manifest() + payload["expected_sha25"] = "typo" + + with pytest.raises(OfflineFetchManifestError, match="expected_sha25"): + validate_offline_fetch_manifest(payload) + + payload = _manifest() + payload["artifacts"][0]["expected_sha25"] = "typo" + + with pytest.raises(OfflineFetchManifestError, match="expected_sha25"): + validate_offline_fetch_manifest(payload) + + +def test_rejects_duplicate_destinations(): + payload = _manifest() + payload["artifacts"].append(deepcopy(payload["artifacts"][0])) + + with pytest.raises(OfflineFetchManifestError, match="duplicates destination_path"): + validate_offline_fetch_manifest(payload) + + payload = _manifest() + duplicate = deepcopy(payload["artifacts"][0]) + duplicate["package_id"] = "another-package" + payload["artifacts"].append(duplicate) + + with pytest.raises(OfflineFetchManifestError, match="duplicates destination_path"): + validate_offline_fetch_manifest(payload) + + +def test_rejects_destination_filename_and_manifest_directory_mismatches(): + payload = _manifest() + payload["artifacts"][0]["destination_path"] = ( + "db/data/statbel/fiscal_income_2023/not-the-expected-file.csv" + ) + + with pytest.raises(OfflineFetchManifestError, match="expected_filename"): + validate_offline_fetch_manifest(payload) + + payload = _manifest() + payload["artifacts"][0]["manifest_path"] = ( + "db/data/statbel/another_package/manifest.yaml" + ) + + with pytest.raises(OfflineFetchManifestError, match="beside destination_path"): + validate_offline_fetch_manifest(payload) + + +def test_rejects_destination_paths_that_can_overwrite_manifests(): + payload = _manifest() + artifact = payload["artifacts"][0] + artifact["expected_filename"] = "manifest.yaml" + artifact["destination_path"] = artifact["manifest_path"] + + with pytest.raises(OfflineFetchManifestError, match="overwrite manifest_path"): + validate_offline_fetch_manifest(payload) + + payload = _manifest() + first = payload["artifacts"][0] + second = deepcopy(first) + second["package_id"] = "another-package" + second["expected_filename"] = "manifest.yaml" + second["destination_path"] = first["manifest_path"] + second["manifest_path"] = "db/data/statbel/fiscal_income_2023/another-manifest.yaml" + payload["artifacts"].append(second) + + with pytest.raises( + OfflineFetchManifestError, + match="overwrite manifest_path from artifacts\\[0\\]", + ): + validate_offline_fetch_manifest(payload) + + payload["artifacts"].reverse() + with pytest.raises( + OfflineFetchManifestError, + match="identify destination_path from artifacts\\[0\\]", + ): + validate_offline_fetch_manifest(payload) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("url", "http://statbel.fgov.be/data.csv"), + ("expected_filename", "nested/data.csv"), + ("destination_path", "packages/statbel/data.csv"), + ("manifest_path", "db/data/statbel/artifact.json"), + ("manifest_year", True), + ("post_download_steps", []), + ("source_package_path", "db/data/statbel/source_package.yaml"), + ], +) +def test_rejects_malformed_required_artifact_fields(field, value): + payload = _manifest() + payload["artifacts"][0][field] = value + + with pytest.raises(OfflineFetchManifestError, match=field): + validate_offline_fetch_manifest(payload) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("schema_version", "ledger.offline_fetch_manifest.v2"), + ("generated_for", ""), + ("artifacts", []), + ], +) +def test_rejects_malformed_manifest_fields(field, value): + payload = _manifest() + payload[field] = value + + with pytest.raises(OfflineFetchManifestError, match=field): + validate_offline_fetch_manifest(payload)