257 lines
9.2 KiB
Python
257 lines
9.2 KiB
Python
"""Validation for the Git-reviewed Evidence authoring workspace."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import unicodedata
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import yaml
|
|
from pydantic import ValidationError, field_validator, model_validator
|
|
|
|
from tht.evidence.canonical import (
|
|
CuratedEvidence,
|
|
StrictModel,
|
|
is_evidence_id,
|
|
load_curated_tree,
|
|
validate_source_file,
|
|
)
|
|
|
|
MAX_AUTHORING_FILE_BYTES = 10 * 1024 * 1024
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidationFinding:
|
|
severity: Literal["error", "warning"]
|
|
code: str
|
|
path: str
|
|
message: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ValidationReport:
|
|
findings: tuple[ValidationFinding, ...]
|
|
|
|
@property
|
|
def publishable(self) -> bool:
|
|
return not any(finding.severity == "error" for finding in self.findings)
|
|
|
|
|
|
class ManifestSource(StrictModel):
|
|
sha256: str
|
|
units: tuple[str, ...]
|
|
|
|
@field_validator("sha256")
|
|
@classmethod
|
|
def _validate_sha256(cls, value: str) -> str:
|
|
if not value.startswith("sha256:") or len(value) != 71:
|
|
raise ValueError("sha256 must be a sha256 digest")
|
|
try:
|
|
int(value.removeprefix("sha256:"), 16)
|
|
except ValueError as error:
|
|
raise ValueError("sha256 must be a sha256 digest") from error
|
|
return value
|
|
|
|
@field_validator("units")
|
|
@classmethod
|
|
def _validate_units(cls, value: tuple[str, ...]) -> tuple[str, ...]:
|
|
if any(not is_evidence_id(unit) for unit in value):
|
|
raise ValueError("units must use stable evidence identifiers")
|
|
return value
|
|
|
|
|
|
class EvidenceManifest(StrictModel):
|
|
schema_version: Literal[1]
|
|
pipeline_version: Literal["evidence-authoring-v1"]
|
|
sources: dict[str, ManifestSource]
|
|
orphans: tuple[str, ...]
|
|
|
|
@field_validator("sources")
|
|
@classmethod
|
|
def _validate_sources(cls, value: dict[str, ManifestSource]) -> dict[str, ManifestSource]:
|
|
for source_path in value:
|
|
validate_source_file(source_path)
|
|
return value
|
|
|
|
@field_validator("orphans")
|
|
@classmethod
|
|
def _validate_orphans(cls, value: tuple[str, ...]) -> tuple[str, ...]:
|
|
if any(not is_evidence_id(unit) for unit in value):
|
|
raise ValueError("orphans must use stable evidence identifiers")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def _validate_unit_membership(self) -> EvidenceManifest:
|
|
seen: set[str] = set()
|
|
for source in self.sources.values():
|
|
duplicate = seen.intersection(source.units)
|
|
if duplicate:
|
|
raise ValueError("a unit may belong to only one source")
|
|
seen.update(source.units)
|
|
if len(set(self.orphans)) != len(self.orphans):
|
|
raise ValueError("orphans must be unique")
|
|
return self
|
|
|
|
|
|
def load_manifest(path: Path) -> EvidenceManifest:
|
|
"""Load the managed, versioned authoring manifest."""
|
|
try:
|
|
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeDecodeError, yaml.YAMLError) as error:
|
|
raise ValueError("manifest cannot be read") from error
|
|
try:
|
|
return EvidenceManifest.model_validate(raw)
|
|
except ValidationError as error:
|
|
raise ValueError("manifest is invalid") from error
|
|
|
|
|
|
def dump_manifest(manifest: EvidenceManifest) -> str:
|
|
"""Serialize the manifest deterministically for Git review."""
|
|
return yaml.safe_dump(manifest.model_dump(mode="json"), allow_unicode=True, sort_keys=True)
|
|
|
|
|
|
def validate_workspace_evidence(workspace_root: Path) -> ValidationReport:
|
|
"""Validate the curated corpus without writing the workspace."""
|
|
evidence_root = workspace_root / "evidence"
|
|
findings: list[ValidationFinding] = []
|
|
manifest_path = evidence_root / "manifest.yaml"
|
|
if not manifest_path.is_file():
|
|
return ValidationReport((ValidationFinding(
|
|
severity="error",
|
|
code="manifest_missing",
|
|
path="manifest.yaml",
|
|
message="The managed Evidence manifest is missing.",
|
|
),))
|
|
try:
|
|
manifest = load_manifest(manifest_path)
|
|
except ValueError:
|
|
return ValidationReport((ValidationFinding(
|
|
severity="error",
|
|
code="manifest_invalid",
|
|
path="manifest.yaml",
|
|
message="The managed Evidence manifest is invalid.",
|
|
),))
|
|
for orphan in manifest.orphans:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="orphaned_unit",
|
|
path="manifest.yaml",
|
|
message=f"Orphaned Evidence {orphan} must be resolved before publication.",
|
|
))
|
|
try:
|
|
documents = load_curated_tree(evidence_root / "curated")
|
|
except (OSError, ValidationError, ValueError):
|
|
return ValidationReport(tuple(findings + [ValidationFinding(
|
|
severity="error",
|
|
code="curated_invalid",
|
|
path="curated",
|
|
message="A Curated Evidence document is invalid or cannot be read.",
|
|
)]))
|
|
source_texts = {
|
|
source_path: _validate_manifest_source(evidence_root, source_path, source, findings)
|
|
for source_path, source in manifest.sources.items()
|
|
}
|
|
seen_ids: set[str] = set()
|
|
for evidence in documents:
|
|
if evidence.id in seen_ids:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="duplicate_evidence_id",
|
|
path="curated",
|
|
message=f"Evidence id {evidence.id} appears more than once.",
|
|
))
|
|
seen_ids.add(evidence.id)
|
|
findings.extend(_validate_unit(manifest, evidence, source_texts.get(evidence.provenance.source_file)))
|
|
for source in manifest.sources.values():
|
|
for unit in source.units:
|
|
if unit not in seen_ids:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="manifest_unit_without_curated",
|
|
path="manifest.yaml",
|
|
message=f"Manifest Evidence {unit} has no Curated document.",
|
|
))
|
|
return ValidationReport(tuple(findings))
|
|
|
|
|
|
def _validate_manifest_source(
|
|
evidence_root: Path,
|
|
source_path: str,
|
|
manifest_source: ManifestSource,
|
|
findings: list[ValidationFinding],
|
|
) -> str | None:
|
|
path = evidence_root / source_path
|
|
if not path.is_file():
|
|
findings.append(ValidationFinding("error", "source_missing", source_path,
|
|
"The manifest source does not exist."))
|
|
return None
|
|
if path.stat().st_size > MAX_AUTHORING_FILE_BYTES:
|
|
findings.append(ValidationFinding("error", "source_oversized", source_path,
|
|
"The manifest source exceeds the authoring size limit."))
|
|
return None
|
|
try:
|
|
source = _normalize(path.read_text(encoding="utf-8"))
|
|
except UnicodeDecodeError:
|
|
findings.append(ValidationFinding("error", "source_unreadable", source_path,
|
|
"The manifest source is not valid UTF-8."))
|
|
return None
|
|
digest = "sha256:" + hashlib.sha256(source.encode("utf-8")).hexdigest()
|
|
if digest != manifest_source.sha256:
|
|
findings.append(ValidationFinding("error", "source_hash_mismatch", source_path,
|
|
"The manifest hash does not match the normalized source."))
|
|
return source
|
|
|
|
|
|
def _validate_unit(
|
|
manifest: EvidenceManifest, evidence: CuratedEvidence, source: str | None,
|
|
) -> list[ValidationFinding]:
|
|
path = evidence.provenance.source_file
|
|
findings: list[ValidationFinding] = []
|
|
manifest_source = manifest.sources.get(path)
|
|
if manifest_source is None:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="manifest_source_missing",
|
|
path="manifest.yaml",
|
|
message="The manifest does not contain the provenance source.",
|
|
))
|
|
else:
|
|
if manifest_source.sha256 != evidence.provenance.source_sha256:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="manifest_source_hash_mismatch",
|
|
path="manifest.yaml",
|
|
message="The manifest hash does not match the Evidence provenance.",
|
|
))
|
|
if evidence.id not in manifest_source.units:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="manifest_unit_missing",
|
|
path="manifest.yaml",
|
|
message="The manifest does not link the Evidence unit to its source.",
|
|
))
|
|
if source is None:
|
|
return findings
|
|
for excerpt in evidence.provenance.supporting_excerpts:
|
|
if _normalize(excerpt) not in source:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="supporting_excerpt_missing",
|
|
path=path,
|
|
message="A supporting excerpt is absent from the normalized source.",
|
|
))
|
|
for item in evidence.review_items:
|
|
findings.append(ValidationFinding(
|
|
severity="error",
|
|
code="unresolved_review_item",
|
|
path=path,
|
|
message=f"Review item {item.code} must be resolved before publication.",
|
|
))
|
|
return findings
|
|
|
|
|
|
def _normalize(text: str) -> str:
|
|
return unicodedata.normalize("NFC", text.replace("\r\n", "\n").replace("\r", "\n"))
|