"""Validate provenance DAG validation and information propagation.
The tests cover public behavior, differentiability, validation, and stable
scientific identity in the supported certification regime.
"""
import pytest
from beartype.typing import Any
from hypothesis import given
from hypothesis import strategies as st
from diffpes.certify import (
build_provenance,
effective_information,
invalidated_claims,
lineage,
validate_provenance,
)
from diffpes.types import make_transformation_record
def _record(
transformation_id: Any,
parents: Any,
outputs: Any,
*,
preserves: Any = (),
introduces: Any = (),
destroys: Any = (),
invalidates: Any = (),
) -> Any:
return make_transformation_record(
transformation_id=transformation_id,
transformation_version="1.0.0",
parent_ids=parents,
output_ids=outputs,
preserves=preserves,
introduces=introduces,
destroys=destroys,
invalidates_claims=invalidates,
parameters_checksum="none",
)
[docs]
class TestValidateProvenance:
"""Verify :func:`~diffpes.certify.validate_provenance`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.validate_provenance`
"""
[docs]
def test_empty_graph_is_valid(self) -> None:
"""Verify an empty declared graph has no structural errors.
The case uses explicit inputs in the supported certification regime.
It checks the public result or the documented failure state.
Notes
-----
The test builds and independently validates the minimal graph.
"""
graph: Any
graph = build_provenance(())
assert validate_provenance(graph).valid
[docs]
@given(
st.lists(
st.integers(min_value=0, max_value=31), min_size=1, max_size=12
)
)
def test_generated_acyclic_graphs_validate(
self, parent_choices: Any
) -> None:
"""Validate generated DAGs independently of their supplied record 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.
"""
index: Any
choice: Any
records: Any
parents: Any
graph: Any
report: Any
records = []
for index, choice in enumerate(parent_choices):
parents = () if index == 0 else (f"node-{choice % index}",)
records.append(
_record(
f"org.diffpes.transform.generated.step_{index}",
parents,
(f"node-{index}",),
preserves=("axis",),
introduces=(f"stage-{index}",),
)
)
graph = build_provenance(tuple(reversed(records)))
report = validate_provenance(graph)
assert report.valid, report.errors
assert set(graph.topological_order) == {
f"node-{index}" for index in range(len(records))
}
[docs]
class TestBuildProvenance:
"""Verify :func:`~diffpes.certify.build_provenance`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.build_provenance`
"""
[docs]
def test_cycle_is_rejected_and_remains_inspectable_in_nonstrict_mode(
self,
) -> None:
"""Detect a multi-step cycle without hanging topological analysis.
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.
"""
first: Any
second: Any
graph: Any
report: Any
first = _record(
"org.diffpes.transform.test.cycle_a",
("b",),
("a",),
)
second = _record(
"org.diffpes.transform.test.cycle_b",
("a",),
("b",),
)
with pytest.raises(ValueError, match="cycle"):
build_provenance((first, second))
graph = build_provenance((first, second), strict=False)
report = validate_provenance(graph)
assert not report.valid
assert any("cycle" in error for error in report.errors)
[docs]
def test_missing_parent_and_multiple_producer_are_rejected(self) -> None:
"""Require every non-source parent and every output producer to be unique.
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.
"""
missing: Any
first: Any
second: Any
missing = _record(
"org.diffpes.transform.test.missing",
("not-declared",),
("result",),
)
with pytest.raises(ValueError, match="missing parents"):
build_provenance((missing,))
first = _record("org.diffpes.transform.test.one", (), ("same",))
second = _record("org.diffpes.transform.test.two", (), ("same",))
with pytest.raises(ValueError, match="multiple transformations"):
build_provenance((first, second))
[docs]
class TestLineage:
"""Verify :func:`~diffpes.certify.lineage`.
The cases cover the public behavior in the supported certification regime.
:see: :func:`~diffpes.certify.lineage`
"""
[docs]
def test_unknown_node_inspection_is_rejected(self) -> None:
"""Avoid silently returning empty state for a misspelled artifact ID.
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.
"""
source: Any
graph: Any
source = _record("org.diffpes.transform.test.known", (), ("known",))
graph = build_provenance((source,))
with pytest.raises(KeyError, match="unknown provenance node"):
effective_information(graph, "unknown")
with pytest.raises(KeyError, match="unknown provenance node"):
lineage(graph, "unknown")
[docs]
class TestInvalidatedClaims:
"""Verify :func:`~diffpes.certify.invalidated_claims`.
The cases cover sorted claim invalidation at a derived result node.
:see: :func:`~diffpes.certify.invalidated_claims`
"""
[docs]
def test_node_reports_inherited_invalidated_claims(self) -> None:
"""Return all claim identities invalidated along the node lineage.
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.
"""
record: Any
graph: Any
record = _record(
"org.diffpes.transform.test.invalidate",
(),
("result",),
invalidates=("org.diffpes.claim.test.invalid",),
)
graph = build_provenance((record,))
assert invalidated_claims(graph, "result") == (
"org.diffpes.claim.test.invalid",
)