Skip to content

Commit 2c6aff8

Browse files
authored
Create test_review_workflow_readiness.py
1 parent 080b08d commit 2c6aff8

1 file changed

Lines changed: 315 additions & 0 deletions

File tree

Lines changed: 315 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,315 @@
1+
from __future__ import annotations
2+
3+
import pytest
4+
5+
from ix_autonomy_assurance_case_runtime.authority import ReviewActor
6+
from ix_autonomy_assurance_case_runtime.contracts import (
7+
ContractValueError,
8+
EvidenceStatus,
9+
ReviewDisposition,
10+
)
11+
from ix_autonomy_assurance_case_runtime.evidence import EvidenceBundle, EvidenceRecord
12+
from ix_autonomy_assurance_case_runtime.prototype_readiness import (
13+
PrototypeClaimLevel,
14+
PrototypeReadinessDecision,
15+
)
16+
from ix_autonomy_assurance_case_runtime.review_workflow import (
17+
ReviewAuthorityBinding,
18+
ReviewAuthorityScope,
19+
ReviewDissentRecord,
20+
ReviewDissentSeverity,
21+
ReviewFinding,
22+
ReviewFindingSeverity,
23+
ReviewFindingStatus,
24+
ReviewSignoffRecord,
25+
ReviewWorkflowRecord,
26+
ReviewWorkflowStatus,
27+
)
28+
from ix_autonomy_assurance_case_runtime.review_workflow_readiness import (
29+
REVIEW_WORKFLOW_CAPABILITY_ID,
30+
ReviewWorkflowLayerReadinessEvaluator,
31+
ReviewWorkflowReadinessDecision,
32+
ReviewWorkflowReadinessFinding,
33+
ReviewWorkflowReadinessFindingSeverity,
34+
ReviewWorkflowReadinessFindingSource,
35+
)
36+
37+
38+
def _actor(actor_id: str = "reviewer-001") -> ReviewActor:
39+
return ReviewActor(
40+
actor_id=actor_id,
41+
role="assurance-reviewer",
42+
display_name="Assurance Reviewer",
43+
)
44+
45+
46+
def _binding(
47+
*,
48+
actor_id: str = "reviewer-001",
49+
scopes: tuple[ReviewAuthorityScope, ...] = (
50+
ReviewAuthorityScope.ASSURANCE_CASE,
51+
ReviewAuthorityScope.SCENARIO_CAMPAIGN,
52+
ReviewAuthorityScope.MONITORING,
53+
),
54+
can_sign: bool = True,
55+
can_waive: bool = True,
56+
can_record_dissent: bool = True,
57+
) -> ReviewAuthorityBinding:
58+
return ReviewAuthorityBinding(
59+
binding_id=f"binding-{actor_id}",
60+
actor=_actor(actor_id),
61+
authority_scopes=scopes,
62+
can_sign=can_sign,
63+
can_waive=can_waive,
64+
can_record_dissent=can_record_dissent,
65+
)
66+
67+
68+
def _finding(
69+
*,
70+
finding_id: str = "finding-runtime-evidence-001",
71+
status: ReviewFindingStatus = ReviewFindingStatus.CLOSED,
72+
severity: ReviewFindingSeverity = ReviewFindingSeverity.HIGH,
73+
opened_by_actor_id: str = "reviewer-001",
74+
scope: ReviewAuthorityScope = ReviewAuthorityScope.SCENARIO_CAMPAIGN,
75+
) -> ReviewFinding:
76+
return ReviewFinding(
77+
finding_id=finding_id,
78+
scope=scope,
79+
severity=severity,
80+
status=status,
81+
title="Runtime evidence reviewed",
82+
rationale="Campaign evidence supports the bounded runtime behavior claim.",
83+
opened_by_actor_id=opened_by_actor_id,
84+
opened_at_utc="2026-05-12T12:00:00Z",
85+
requirement_ids=("req-runtime-boundary",),
86+
hazard_ids=("hazard-runtime-boundary",),
87+
evidence_bundle_ids=("ev-review-finding-001",),
88+
source_record_ids=("campaign-run-001",),
89+
)
90+
91+
92+
def _signoff(
93+
*,
94+
actor_id: str = "reviewer-001",
95+
scope: ReviewAuthorityScope = ReviewAuthorityScope.ASSURANCE_CASE,
96+
disposition: ReviewDisposition = ReviewDisposition.APPROVED,
97+
evidence_bundle_ids: tuple[str, ...] = ("ev-review-signoff-001",),
98+
) -> ReviewSignoffRecord:
99+
return ReviewSignoffRecord(
100+
signoff_id="signoff-reviewer-001",
101+
workflow_id="workflow-review-001",
102+
actor=_actor(actor_id),
103+
scope=scope,
104+
disposition=disposition,
105+
rationale="Evidence, campaign, monitoring, and provenance posture reviewed.",
106+
signed_at_utc="2026-05-12T13:00:00Z",
107+
evidence_bundle_ids=evidence_bundle_ids,
108+
)
109+
110+
111+
def _dissent(
112+
*,
113+
severity: ReviewDissentSeverity = ReviewDissentSeverity.CONCERN,
114+
) -> ReviewDissentRecord:
115+
return ReviewDissentRecord(
116+
dissent_id="dissent-reviewer-001",
117+
workflow_id="workflow-review-001",
118+
actor=_actor("reviewer-002"),
119+
scope=ReviewAuthorityScope.MONITORING,
120+
severity=severity,
121+
rationale="Monitoring confidence should be reviewed again before wider claims.",
122+
recorded_at_utc="2026-05-12T13:10:00Z",
123+
evidence_bundle_ids=("ev-review-dissent-001",),
124+
related_finding_ids=("finding-runtime-evidence-001",),
125+
)
126+
127+
128+
def _workflow(
129+
*,
130+
status: ReviewWorkflowStatus = ReviewWorkflowStatus.COMPLETED,
131+
findings: tuple[ReviewFinding, ...] | None = None,
132+
signoffs: tuple[ReviewSignoffRecord, ...] | None = None,
133+
dissents: tuple[ReviewDissentRecord, ...] = (),
134+
bindings: tuple[ReviewAuthorityBinding, ...] | None = None,
135+
) -> ReviewWorkflowRecord:
136+
return ReviewWorkflowRecord(
137+
workflow_id="workflow-review-001",
138+
case_id="case-runtime-001",
139+
title="Runtime assurance review",
140+
status=status,
141+
authority_bindings=(
142+
bindings
143+
if bindings is not None
144+
else (_binding(), _binding(actor_id="reviewer-002"))
145+
),
146+
findings=findings if findings is not None else (_finding(),),
147+
signoffs=signoffs if signoffs is not None else (_signoff(),),
148+
dissents=dissents,
149+
evidence_bundle_ids=("ev-review-workflow-001",),
150+
system_id="system-runtime-001",
151+
deployment_id="deploy-runtime-001",
152+
)
153+
154+
155+
def _bundle(bundle_id: str, *, hashed: bool = True) -> EvidenceBundle:
156+
bundle = EvidenceBundle(
157+
bundle_id=bundle_id,
158+
case_id="case-runtime-001",
159+
records=(
160+
EvidenceRecord(
161+
evidence_id=f"record-{bundle_id}",
162+
kind="review-workflow",
163+
source="unit-test",
164+
payload={"bundle_id": bundle_id},
165+
status=EvidenceStatus.ACCEPTED,
166+
),
167+
),
168+
)
169+
if hashed:
170+
return bundle.with_computed_hashes()
171+
return bundle
172+
173+
174+
def _bundles(*, unhashed: str | None = None) -> tuple[EvidenceBundle, ...]:
175+
bundle_ids = (
176+
"ev-review-workflow-001",
177+
"ev-review-finding-001",
178+
"ev-review-signoff-001",
179+
"ev-review-dissent-001",
180+
)
181+
return tuple(_bundle(bundle_id, hashed=bundle_id != unhashed) for bundle_id in bundle_ids)
182+
183+
184+
def test_review_workflow_readiness_completes_clean_workflow() -> None:
185+
report = ReviewWorkflowLayerReadinessEvaluator(
186+
evidence_bundles=_bundles(),
187+
).evaluate(_workflow())
188+
189+
assert report.decision is ReviewWorkflowReadinessDecision.COMPLETE
190+
assert report.is_complete()
191+
assert report.completed_capability_ids() == (REVIEW_WORKFLOW_CAPABILITY_ID,)
192+
assert report.blocker_count == 0
193+
assert report.warning_count == 0
194+
assert report.summary() == (
195+
"review-workflow-readiness: complete "
196+
"(1 finding(s), 1 signoff(s), 0 dissent(s), 3 evidence bundle(s), "
197+
"0 blocker(s), 0 warning(s), capability=review-workflow)"
198+
)
199+
200+
201+
def test_review_workflow_readiness_feeds_prototype_claim_gate() -> None:
202+
report = ReviewWorkflowLayerReadinessEvaluator(
203+
evidence_bundles=_bundles(),
204+
).evaluate(_workflow())
205+
206+
prototype_report = report.prototype_readiness_report(
207+
PrototypeClaimLevel.SERIOUS_OPEN_SOURCE_PROTOTYPE,
208+
existing_completed_capability_ids=(
209+
"registry-layer",
210+
"policy-pack-engine",
211+
"framework-crosswalks",
212+
"signed-provenance",
213+
"telemetry-adapters",
214+
"scenario-campaign-runner",
215+
"monitoring-incidents",
216+
),
217+
)
218+
219+
assert prototype_report.decision is PrototypeReadinessDecision.BLOCK
220+
assert REVIEW_WORKFLOW_CAPABILITY_ID in prototype_report.completed_capability_ids
221+
222+
223+
def test_review_workflow_readiness_blocks_uncompleted_workflow() -> None:
224+
report = ReviewWorkflowLayerReadinessEvaluator(
225+
evidence_bundles=_bundles(),
226+
).evaluate(_workflow(status=ReviewWorkflowStatus.IN_REVIEW))
227+
228+
assert report.decision is ReviewWorkflowReadinessDecision.BLOCKED
229+
assert any(
230+
finding.finding_id == "workflow-workflow-review-001-not-acceptance-ready"
231+
for finding in report.findings_for_workflow("workflow-review-001")
232+
)
233+
234+
235+
def test_review_workflow_readiness_blocks_missing_accepting_signoff() -> None:
236+
report = ReviewWorkflowLayerReadinessEvaluator(
237+
evidence_bundles=_bundles(),
238+
).evaluate(_workflow(signoffs=()))
239+
240+
assert report.decision is ReviewWorkflowReadinessDecision.BLOCKED
241+
assert any(
242+
finding.source is ReviewWorkflowReadinessFindingSource.SIGNOFF
243+
for finding in report.findings
244+
)
245+
246+
247+
def test_review_workflow_readiness_blocks_unresolved_review_finding() -> None:
248+
report = ReviewWorkflowLayerReadinessEvaluator(
249+
evidence_bundles=_bundles(),
250+
).evaluate(_workflow(findings=(_finding(status=ReviewFindingStatus.OPEN),)))
251+
252+
assert report.decision is ReviewWorkflowReadinessDecision.BLOCKED
253+
assert any(
254+
finding.finding_id == "finding-finding-runtime-evidence-001-unresolved-blocker"
255+
for finding in report.findings_for_review_finding("finding-runtime-evidence-001")
256+
)
257+
258+
259+
def test_review_workflow_readiness_blocks_blocking_dissent() -> None:
260+
report = ReviewWorkflowLayerReadinessEvaluator(
261+
evidence_bundles=_bundles(),
262+
).evaluate(
263+
_workflow(
264+
dissents=(_dissent(severity=ReviewDissentSeverity.BLOCKING_OBJECTION),),
265+
)
266+
)
267+
268+
assert report.decision is ReviewWorkflowReadinessDecision.BLOCKED
269+
assert any(
270+
finding.finding_id == "dissent-dissent-reviewer-001-blocking"
271+
for finding in report.findings_for_dissent("dissent-reviewer-001")
272+
)
273+
274+
275+
def test_review_workflow_readiness_limited_for_evidence_warnings() -> None:
276+
report = ReviewWorkflowLayerReadinessEvaluator(
277+
evidence_bundles=_bundles(unhashed="ev-review-finding-001"),
278+
).evaluate(_workflow())
279+
280+
assert report.decision is ReviewWorkflowReadinessDecision.LIMITED
281+
assert report.warning_count == 2
282+
assert report.findings_for_evidence_bundle("ev-review-finding-001")
283+
284+
285+
def test_review_workflow_readiness_blocks_validation_authority_findings() -> None:
286+
report = ReviewWorkflowLayerReadinessEvaluator(
287+
evidence_bundles=_bundles(),
288+
).evaluate(
289+
_workflow(
290+
bindings=(_binding(scopes=(ReviewAuthorityScope.MONITORING,),),),
291+
signoffs=(_signoff(scope=ReviewAuthorityScope.ASSURANCE_CASE),),
292+
)
293+
)
294+
295+
assert report.decision is ReviewWorkflowReadinessDecision.BLOCKED
296+
assert report.findings_for_actor("reviewer-001")
297+
298+
299+
def test_review_workflow_readiness_finding_validates_optional_identifiers() -> None:
300+
with pytest.raises(ContractValueError, match="needs a message"):
301+
ReviewWorkflowReadinessFinding(
302+
finding_id="finding-review-readiness-001",
303+
severity=ReviewWorkflowReadinessFindingSeverity.BLOCKER,
304+
source=ReviewWorkflowReadinessFindingSource.READINESS,
305+
message="",
306+
)
307+
308+
with pytest.raises(ContractValueError, match="signoff_id must not be blank"):
309+
ReviewWorkflowReadinessFinding(
310+
finding_id="finding-review-readiness-001",
311+
severity=ReviewWorkflowReadinessFindingSeverity.BLOCKER,
312+
source=ReviewWorkflowReadinessFindingSource.SIGNOFF,
313+
message="Bad signoff.",
314+
signoff_id="",
315+
)

0 commit comments

Comments
 (0)