"""Traceable Knowledge Model evidence for DSW render regression."""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from .translation_repository import DswPreviewRuntime
from .yaml_config import YamlConfigError, load_yaml_file
EVIDENCE_SCHEMA_VERSION = 1
class RegressionEvidenceError(ValueError):
"""Raised when regression evidence configuration is invalid or stale."""
[docs]
@dataclass(frozen=True)
class KnowledgeModelEvidence:
"""Pinned provenance for one immutable Knowledge Model bundle."""
key: str
path: Path
package_id: str
version: str
metamodel_version: str
source_url: str
sha256: str
[docs]
@dataclass(frozen=True)
class RegressionEvidenceConfig:
"""Knowledge Model fixtures assigned to supported DSW runtimes."""
knowledge_models: dict[str, KnowledgeModelEvidence]
runtime_knowledge_models: dict[str, str]
[docs]
def knowledge_model_for_runtime(self, runtime: DswPreviewRuntime) -> KnowledgeModelEvidence:
"""Return the pinned Knowledge Model assigned to ``runtime``."""
fixture_key = self.runtime_knowledge_models.get(runtime.metamodel_key)
if fixture_key is None:
raise RegressionEvidenceError(
f"Runtime {runtime.metamodel_key!r} has no regression Knowledge Model assignment"
)
try:
return self.knowledge_models[fixture_key]
except KeyError as exc:
raise RegressionEvidenceError(
f"Runtime {runtime.metamodel_key!r} references unknown Knowledge Model "
f"fixture {fixture_key!r}"
) from exc
[docs]
def load_regression_evidence_config(path: Path) -> RegressionEvidenceConfig:
"""Load and validate pinned regression evidence from YAML."""
path = Path(path)
try:
payload = load_yaml_file(path)
except (OSError, YamlConfigError) as exc:
raise RegressionEvidenceError(str(exc)) from exc
if not isinstance(payload, dict):
raise RegressionEvidenceError(f"Regression evidence config {path} must be a mapping")
_reject_unknown_keys(
payload,
{"knowledge_models", "runtime_knowledge_models", "schema_version"},
"regression evidence config",
)
if payload.get("schema_version") != EVIDENCE_SCHEMA_VERSION:
raise RegressionEvidenceError(
f"Regression evidence schema_version must be {EVIDENCE_SCHEMA_VERSION}"
)
raw_models = payload.get("knowledge_models")
if not isinstance(raw_models, dict) or not raw_models:
raise RegressionEvidenceError("Regression evidence must define knowledge_models")
knowledge_models = {
_mapping_key(key, context="knowledge_models"): _load_knowledge_model(
key=_mapping_key(key, context="knowledge_models"),
payload=value,
base_dir=path.parent,
)
for key, value in raw_models.items()
}
raw_assignments = payload.get("runtime_knowledge_models")
if not isinstance(raw_assignments, dict) or not raw_assignments:
raise RegressionEvidenceError("Regression evidence must define runtime_knowledge_models")
assignments = {
_mapping_key(key, context="runtime_knowledge_models"): _required_string_value(
value,
context=f"runtime_knowledge_models.{key}",
)
for key, value in raw_assignments.items()
}
config = RegressionEvidenceConfig(
knowledge_models=knowledge_models,
runtime_knowledge_models=assignments,
)
_validate_assignment_references(config)
return config
[docs]
def validate_regression_evidence_config(
config: RegressionEvidenceConfig,
runtimes: tuple[DswPreviewRuntime, ...],
) -> None:
"""Validate runtime assignments and every pinned Knowledge Model bundle."""
runtime_keys = {runtime.metamodel_key for runtime in runtimes}
assignment_keys = set(config.runtime_knowledge_models)
missing = sorted(runtime_keys - assignment_keys)
stale = sorted(assignment_keys - runtime_keys)
if missing:
raise RegressionEvidenceError(
f"Missing regression Knowledge Model assignment for runtime(s): {', '.join(missing)}"
)
if stale:
raise RegressionEvidenceError(
f"Regression evidence references unknown runtime(s): {', '.join(stale)}"
)
for runtime in runtimes:
verify_knowledge_model_evidence(config.knowledge_model_for_runtime(runtime))
[docs]
def verify_knowledge_model_evidence(evidence: KnowledgeModelEvidence) -> None:
"""Verify one pinned bundle's checksum and declared package metadata."""
if not evidence.path.is_file():
raise RegressionEvidenceError(
f"Knowledge Model fixture {evidence.key!r} does not exist: {evidence.path}"
)
actual_sha256 = _sha256(evidence.path)
if actual_sha256 != evidence.sha256:
raise RegressionEvidenceError(
f"Knowledge Model fixture {evidence.key!r} checksum mismatch: "
f"expected {evidence.sha256}, got {actual_sha256}"
)
try:
payload = json.loads(evidence.path.read_text(encoding="utf-8"))
except json.JSONDecodeError as exc:
raise RegressionEvidenceError(
f"Knowledge Model fixture {evidence.key!r} is not valid JSON"
) from exc
if not isinstance(payload, dict):
raise RegressionEvidenceError(
f"Knowledge Model fixture {evidence.key!r} must contain a JSON object"
)
expected_fields = {
"id": evidence.package_id,
"metamodelVersion": evidence.metamodel_version,
"version": evidence.version,
}
for field, expected in expected_fields.items():
actual = str(payload.get(field, ""))
if actual != expected:
raise RegressionEvidenceError(
f"Knowledge Model fixture {evidence.key!r} declares {field}={actual!r}; "
f"expected {expected!r}"
)
def _load_knowledge_model(
*,
key: str,
payload: object,
base_dir: Path,
) -> KnowledgeModelEvidence:
if not isinstance(payload, dict):
raise RegressionEvidenceError(f"Knowledge Model fixture {key!r} must be a mapping")
_reject_unknown_keys(
payload,
{"metamodel_version", "package_id", "path", "sha256", "source_url", "version"},
f"knowledge_models.{key}",
)
raw_path = _required_string(payload, "path", context=f"knowledge_models.{key}")
path = Path(raw_path)
if not path.is_absolute():
path = (base_dir / path).resolve()
sha256 = _required_string(payload, "sha256", context=f"knowledge_models.{key}")
if len(sha256) != 64 or any(character not in "0123456789abcdef" for character in sha256):
raise RegressionEvidenceError(
f"knowledge_models.{key}.sha256 must be a lowercase SHA-256 digest"
)
return KnowledgeModelEvidence(
key=key,
path=path,
package_id=_required_string(payload, "package_id", context=f"knowledge_models.{key}"),
version=_required_string(payload, "version", context=f"knowledge_models.{key}"),
metamodel_version=_required_string(
payload,
"metamodel_version",
context=f"knowledge_models.{key}",
),
source_url=_required_string(payload, "source_url", context=f"knowledge_models.{key}"),
sha256=sha256,
)
def _validate_assignment_references(config: RegressionEvidenceConfig) -> None:
unknown = sorted(set(config.runtime_knowledge_models.values()) - set(config.knowledge_models))
if unknown:
raise RegressionEvidenceError(
f"Unknown Knowledge Model fixture assignment(s): {', '.join(unknown)}"
)
def _reject_unknown_keys(
payload: dict[object, object],
allowed: set[str],
context: str,
) -> None:
unknown = sorted(str(key) for key in payload if not isinstance(key, str) or key not in allowed)
if unknown:
raise RegressionEvidenceError(f"Unknown field(s) in {context}: {', '.join(unknown)}")
def _mapping_key(value: object, *, context: str) -> str:
if not isinstance(value, str) or not value:
raise RegressionEvidenceError(f"Expected non-empty string key in {context}")
return value
def _required_string(payload: dict[object, object], key: str, *, context: str) -> str:
return _required_string_value(payload.get(key), context=f"{context}.{key}")
def _required_string_value(value: object, *, context: str) -> str:
if not isinstance(value, str) or not value.strip():
raise RegressionEvidenceError(f"Expected non-empty string at {context}")
return value.strip()
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()