"""Validate deterministic immutable certification registries.
The tests cover public behavior, differentiability, validation, and stable
scientific identity in the supported certification regime.
"""
import subprocess
import sys
import pytest
from beartype.typing import Any
from diffpes.certify import (
freeze_registry,
get_model,
get_transformation,
list_handshakes,
list_models,
list_registered_models,
list_transformations,
packaged_model_card,
register_builtin_models,
register_handshake,
register_model,
register_transformation,
registry_manifest,
registry_snapshot,
render_model_card,
validate_handshake,
validate_registry,
validate_registry_manifest,
)
from diffpes.types import (
make_convention_ref,
make_forward_model_spec,
make_registration_handshake,
make_transformation_contract,
)
def _model_spec(name: str) -> Any:
return make_forward_model_spec(
model_id=f"org.diffpes.model.registry_test.{name}",
model_version="1.0.0",
observable_id="org.diffpes.observable.arpes.intensity",
implementation_ref=f"tests.registry:{name}",
differentiable_paths=("parameters.scale",),
)
[docs]
class TestValidateRegistry:
"""Verify :func:`~diffpes.certify.validate_registry`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.validate_registry`
"""
[docs]
def test_snapshot_is_structurally_valid(self) -> None:
"""Verify a snapshot satisfies ordering and checksum invariants.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
Recomputes validation against the current process-local registry.
"""
assert validate_registry().valid
[docs]
def test_registry_report_recomputes_structural_consistency(self) -> None:
"""Report a stable checksum and successful internal validation.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
report: Any
report = validate_registry()
assert report.valid, report.errors
assert report.model_count >= 0
assert report.transformation_count >= 0
assert report.checksum.startswith("crc32:canonical-1:registry:")
[docs]
class TestRegistrySnapshot:
"""Verify :func:`~diffpes.certify.registry_snapshot`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.registry_snapshot`
"""
[docs]
def test_models_are_sorted_and_resolved_independent_of_registration_order(
self,
) -> None:
"""Expose deterministic immutable snapshots after reverse-order inserts.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
zulu: Any
alpha: Any
ids: Any
snapshot: Any
zulu = _model_spec("zulu")
alpha = _model_spec("alpha")
register_model(zulu, lambda value: value)
register_model(alpha, lambda value: value)
ids = tuple(spec.model_id for spec in list_models())
assert ids == tuple(sorted(ids))
assert get_model(alpha.model_id, alpha.model_version).spec is alpha
snapshot = registry_snapshot()
assert snapshot.models == tuple(
sorted(
snapshot.models,
key=lambda item: (item.spec.model_id, item.spec.model_version),
)
)
with pytest.raises(TypeError):
snapshot.models[0] = snapshot.models[-1]
[docs]
class TestRegisterModel:
"""Verify :func:`~diffpes.certify.register_model`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.register_model`
"""
[docs]
def test_duplicate_model_identity_is_rejected_even_for_same_spec(
self,
) -> None:
"""Prevent import order from replacing an existing scientific identity.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
spec: Any
spec = _model_spec("duplicate")
register_model(spec, lambda value: value)
with pytest.raises(ValueError, match="duplicate model identity"):
register_model(spec, lambda value: value)
[docs]
@pytest.mark.parametrize(
("model_id", "model_version"),
[("invalid", "1.0.0"), ("org.diffpes.model.registry_test.bad", "v1")],
)
def test_invalid_model_identity_is_rejected(
self, model_id: Any, model_version: Any
) -> None:
"""Enforce permanent reverse-DNS IDs and semantic model versions.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
spec: Any
spec = make_forward_model_spec(
model_id=model_id,
model_version=model_version,
observable_id="org.diffpes.observable.arpes.intensity",
implementation_ref="tests.registry:invalid",
)
with pytest.raises(ValueError):
register_model(spec, lambda value: value)
[docs]
class TestGetModel:
"""Verify :func:`~diffpes.certify.get_model`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.get_model`
"""
[docs]
def test_unknown_registry_entries_raise_key_error(self) -> None:
"""Require an exact scientific identity and version on lookup.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
with pytest.raises(KeyError, match="unknown model"):
get_model("org.diffpes.model.registry_test.absent", "1.0.0")
with pytest.raises(KeyError, match="unknown transformation"):
get_transformation(
"org.diffpes.transform.registry_test.absent",
"1.0.0",
)
[docs]
class TestListModels:
"""Verify :func:`~diffpes.certify.list_models`.
The cases cover stable ordering without exposing executor callables.
:see: :func:`~diffpes.certify.list_models`
"""
[docs]
def test_model_specs_are_sorted(self) -> None:
"""Return model specifications in permanent identity order.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
models: Any
models = list_models()
identities: tuple[tuple[str, str], ...] = tuple(
(model.model_id, model.model_version) for model in models
)
assert identities == tuple(sorted(identities))
[docs]
class TestListRegisteredModels:
"""Verify :func:`~diffpes.certify.list_registered_models`.
The cases cover stable ordering of model specifications and executors.
:see: :func:`~diffpes.certify.list_registered_models`
"""
[docs]
def test_registered_bindings_are_sorted(self) -> None:
"""Return complete registered bindings in permanent identity order.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
models: Any
models = list_registered_models()
identities: tuple[tuple[str, str], ...] = tuple(
(model.spec.model_id, model.spec.model_version) for model in models
)
assert identities == tuple(sorted(identities))
[docs]
class TestFreezeRegistry:
"""Verify :func:`~diffpes.certify.freeze_registry`.
The case isolates the process-global freeze operation in a child process.
:see: :func:`~diffpes.certify.freeze_registry`
"""
[docs]
def test_freeze_rejects_later_registration(self) -> None:
"""Reject registry mutation after an application freezes its snapshot.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test compares the result with explicit numerical or structural assertions.
"""
program: str = """
from diffpes.certify import freeze_registry, register_model
from diffpes.types import make_forward_model_spec
freeze_registry()
spec = make_forward_model_spec(
'org.diffpes.model.registry_test.frozen',
'1.0.0',
'org.diffpes.observable.test.result',
'tests.registry:frozen',
)
try:
register_model(spec, lambda value: value)
except ValueError as exc:
assert 'frozen' in str(exc)
else:
raise AssertionError('registration unexpectedly succeeded')
"""
completed: subprocess.CompletedProcess[str] = subprocess.run(
[sys.executable, "-c", program],
check=False,
capture_output=True,
text=True,
)
assert completed.returncode == 0, completed.stderr
[docs]
class TestRegisterHandshake:
"""Verify :func:`~diffpes.certify.register_handshake`.
The case registers one owner without importing its scientific modules.
:see: :func:`~diffpes.certify.register_handshake`
"""
[docs]
def test_registers_one_exact_owner(self) -> None:
"""Register one unique owner handshake in process-local state.
The next registry snapshot must contain the same immutable record.
Notes
-----
The test uses an empty requirement set and a unique owner suffix.
"""
owner_id: str = f"plan-test-{len(list_handshakes())}"
handshake: Any = make_registration_handshake(owner_id)
register_handshake(handshake)
assert handshake in list_handshakes()
[docs]
class TestListHandshakes:
"""Verify :func:`~diffpes.certify.list_handshakes`.
The case checks deterministic owner ordering after registration.
:see: :func:`~diffpes.certify.list_handshakes`
"""
[docs]
def test_returns_owner_sorted_records(self) -> None:
"""Return all handshake declarations in sorted owner order.
Registry insertion order must not change the returned order.
Notes
-----
The test compares the owner sequence with its sorted copy.
"""
owners: tuple[str, ...] = tuple(
item.owner_id for item in list_handshakes()
)
assert owners == tuple(sorted(owners))
[docs]
class TestValidateHandshake:
"""Verify :func:`~diffpes.certify.validate_handshake`.
The case validates model, convention, and evidence references explicitly.
:see: :func:`~diffpes.certify.validate_handshake`
"""
[docs]
def test_reports_missing_then_complete_references(self) -> None:
"""Report missing evidence and then complete the same handshake.
The same declaration must become complete when evidence becomes available.
Notes
-----
The test registers one model and supplies its external evidence later.
"""
suffix: str = str(len(list_models()))
convention: Any = make_convention_ref(
f"org.diffpes.convention.registry_test.{suffix}",
"1.0.0",
"{}",
)
spec: Any = make_forward_model_spec(
model_id=f"org.diffpes.model.registry_test.handshake{suffix}",
model_version="1.0.0",
observable_id="org.diffpes.observable.test.result",
implementation_ref="tests.registry:handshake",
conventions=(convention,),
)
register_model(spec, lambda value: value)
handshake: Any = make_registration_handshake(
owner_id=f"plan-{suffix}",
model_refs=(f"{spec.model_id}@{spec.model_version}",),
convention_refs=(
f"{convention.convention_id}@{convention.version}",
),
evidence_ids=("evidence-plan",),
)
missing: Any = validate_handshake(handshake)
complete: Any = validate_handshake(
handshake,
evidence_ids=("evidence-plan",),
)
assert missing.missing_ids == ("evidence-plan",)
assert bool(complete.complete)
[docs]
def test_plan03_handshake_is_green_with_declared_evidence(self) -> None:
"""Verify the built-in Plan 03 handshake with exact evidence IDs.
The registered transformation contracts and supplied evidence must suffice.
Notes
-----
The test reads evidence IDs from the packaged handshake declaration.
"""
register_builtin_models()
manifest: dict[str, Any] = registry_manifest()
declaration: dict[str, Any] = manifest["handshakes"][0]
handshake: Any = next(
item
for item in list_handshakes()
if item.owner_id == "org.diffpes.plan.03"
)
report: Any = validate_handshake(
handshake,
evidence_ids=tuple(declaration["evidence_ids"]),
)
assert bool(report.complete), report.missing_ids
[docs]
class TestRegistryManifest:
"""Verify :func:`~diffpes.certify.registry_manifest`.
The case reads the packaged manifest without process-local mutation.
:see: :func:`~diffpes.certify.registry_manifest`
"""
[docs]
def test_manifest_has_versioned_builtins(self) -> None:
"""Read the schema and the radial model from package resources.
The manifest must contain explicit versions for stable identities.
Notes
-----
The test checks explicit stable identities, not registry insertion order.
"""
manifest: dict[str, Any] = registry_manifest()
assert manifest["schema_version"] == "1.0.0"
assert manifest["models"][0]["model_id"].endswith("tb_radial")
[docs]
class TestRenderModelCard:
"""Verify :func:`~diffpes.certify.render_model_card`.
The case renders Markdown only from the registered model specification.
:see: :func:`~diffpes.certify.render_model_card`
"""
[docs]
def test_card_contains_exact_model_identity(self) -> None:
"""Render the exact model ID and version in the card header.
The generated text must identify the registered radial model.
Notes
-----
The test uses the packaged radial model specification.
"""
register_builtin_models()
model: Any = get_model(
"org.diffpes.model.arpes.tb_radial",
"0.1.0",
).spec
card: str = render_model_card(model)
assert card.startswith("# org.diffpes.model.arpes.tb_radial")
assert "Version: `0.1.0`." in card
[docs]
class TestPackagedModelCard:
"""Verify :func:`~diffpes.certify.packaged_model_card`.
The case reads the generated radial model card from package resources.
:see: :func:`~diffpes.certify.packaged_model_card`
"""
[docs]
def test_packaged_card_matches_registry_render(self) -> None:
"""Compare the packaged card with a fresh registry-based rendering.
The complete Markdown outputs must match without manual fields.
Notes
-----
The test registers the built-ins and compares the full Markdown text.
"""
register_builtin_models()
model: Any = get_model(
"org.diffpes.model.arpes.tb_radial",
"0.1.0",
).spec
packaged: str = packaged_model_card(
model.model_id, model.model_version
)
assert packaged == render_model_card(model)
[docs]
class TestValidateRegistryManifest:
"""Verify :func:`~diffpes.certify.validate_registry_manifest`.
The case checks every packaged entry and generated model card for drift.
:see: :func:`~diffpes.certify.validate_registry_manifest`
"""
[docs]
def test_builtin_registry_has_no_packaged_drift(self) -> None:
"""Find no missing built-in entry or changed generated model card.
The validator must return an empty tuple after built-in registration.
Notes
-----
The test registers all built-ins before it validates the package files.
"""
register_builtin_models()
assert validate_registry_manifest() == ()