"""Test suite for station.py — SCST1/1.0 Asserts that the two-phase ordering in SPEC.md section 4.4 is actually enforced, not merely documented. The suite is written adversarially: most tests assert that something is REFUSED, because the value of this module is entirely in what it will not do. Every test in the "gate" section corresponds to a path that plain `attestation.py` would have permitted. If a test in that section stops failing when the guard is removed, the guard is decorative and the test is worthless — so the guards were removed one at a time to confirm each one breaks a test. See the sabotage log at the bottom of this file. Run: python test_station.py """ from __future__ import annotations import json import os import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import attestation as att import grading as gr import station as st from station import ( HOLDER, INSPECTOR, Phase, Station, AlreadySubmittedError, GradeNotPublishedError, GradeRoundVoidError, GradeTokenMismatchError, GradeUnresolvedError, PhaseOrderError, RevealMismatchError, RubricMismatchError, StationError, ) CHECKS = 0 FAILS = [] def check(cond, label): global CHECKS CHECKS += 1 if not cond: FAILS.append(label) def raises(exc, fn, label, *a, **k): """Assert fn raises exc. Also asserts it is not a bare Exception escape.""" global CHECKS CHECKS += 1 try: fn(*a, **k) except exc: return except Exception as e: # wrong error type is still a failure FAILS.append(f"{label}: raised {type(e).__name__} not {exc.__name__}") return FAILS.append(f"{label}: did not raise {exc.__name__}") # --------------------------------------------------------------------------- # Fixtures — the same reference unit the rubric vectors use # --------------------------------------------------------------------------- def sheet(**over): s = gr.blank_sheet() unknown = set(over) - set(s) if unknown: raise KeyError(f"not a rubric field: {sorted(unknown)}") s.update(over) return s FLEET_GOOD = dict( scratches_ge_10mm=2, cleaned_labels_removed=True, all_hardware_pass=True, ports_total=6, os_licensed_configured=True, software_test_results_available=True, battery_design_mwh=57000, battery_full_charge_mwh=52000, battery_cycles=310, storage_type="nvme", power_on_hours=9800, tpm_version="2.0", secure_boot_capable=True, secure_boot_enabled=True, uefi_not_legacy=True, iommu_present=True, dma_ports_lockable=True, firmware_latest_applied=True, vendor_support_active=True, cpu_microcode_current=True, supported_os_available=True, full_disk_encryption=True, wipe_standard="purge", wipe_verified=True, wipe_certificate_id="SAN-2026-11402", ) GV1 = sheet(**FLEET_GOOD) # RV pairs, identical to make_grade_vectors.py RV1_H = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 3, "scratches_ge_10mm": 8}) RV1_I = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 4, "scratches_ge_10mm": 11}) RV2_H = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 2, "scratches_ge_10mm": 2}) RV2_I = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 2, "scratches_ge_10mm": 2, "dents_ge_6mm": 1}) RV3_H = sheet(**FLEET_GOOD) RV3_I = sheet(**{**FLEET_GOOD, "battery_full_charge_mwh": 44000, "power_on_hours": 9801}) # GV-5: immaculate but unsanitized. Grades R. GV5 = sheet(**{**FLEET_GOOD, "wipe_standard": "none", "wipe_verified": False, "wipe_certificate_id": "", "scratches_ge_10mm": 0}) def fresh(rid="R-TEST", aid="UNIT-TEST"): return Station(round_id=rid, asset_id=aid) def phase1(station, h_sheet, i_sheet): station.commit_sheet(HOLDER, h_sheet) station.commit_sheet(INSPECTOR, i_sheet) station.reveal_sheet(HOLDER, h_sheet) station.reveal_sheet(INSPECTOR, i_sheet) return station.reconcile() # --------------------------------------------------------------------------- # 1. Happy path # --------------------------------------------------------------------------- s = fresh() rec = phase1(s, GV1, dict(GV1)) check(rec.status == "agreed", "1.1 identical sheets agree") check(s.phase is Phase.GRADE_REVEALED, "1.2 phase after reconcile") b = s.publish_grade() check(b.grade_code == "F6/C6/S4", f"1.3 agreed grade is F6/C6/S4, got {b.grade_code}") check(b.tier == "A", f"1.4 tier A, got {b.tier}") check(b.rubric_hash == gr.rubric_hash(), "1.5 binding carries live rubric hash") check(b.grade_token == gr.grade_token(GV1), "1.6 token matches agreed sheet") check(s.phase is Phase.GRADE_PUBLISHED, "1.7 phase is grade_published") bph = s.open_price_window() check(s.phase is Phase.PRICE_OPEN, "1.8 price window opens after grade") s.commit_price(HOLDER, 40500) s.commit_price(INSPECTOR, 41000) s.reveal_prices() settled = s.settle() check(settled.outcome == "auto_settled", "1.9 inside tolerance auto-settles") check(settled.amount_cents == 40750, f"1.10 midpoint of 40500/41000 is 40750, got {settled.amount_cents}") chain = att.Chain() entry = s.seal(chain) check(chain.verify(), "1.11 chain verifies") check(s.phase is Phase.SEALED, "1.12 sealed") check(entry["body"]["grade_binding"]["grade_token"] == b.grade_token, "1.13 sealed record carries the grade token") check(entry["body"]["grade_binding"]["grade_code"] == "F6/C6/S4", "1.14 sealed record carries the grade code") check(entry["body"]["bound_policy_hash"] == bph, "1.15 sealed record carries the bound policy hash") # --------------------------------------------------------------------------- # 2. THE GATE — price cannot precede grade # # Every check here is a path attestation.py alone would allow. # --------------------------------------------------------------------------- # 2.1 from a freshly opened station s = fresh() raises(GradeNotPublishedError, s.open_price_window, "2.1 price window refused with no sheets at all") # 2.2 after commits but before reveals s = fresh() s.commit_sheet(HOLDER, GV1) s.commit_sheet(INSPECTOR, dict(GV1)) raises(GradeNotPublishedError, s.open_price_window, "2.2 price window refused after commits, before reveals") # 2.3 after reveals but before reconcile s = fresh() s.commit_sheet(HOLDER, GV1) s.commit_sheet(INSPECTOR, dict(GV1)) s.reveal_sheet(HOLDER, GV1) s.reveal_sheet(INSPECTOR, dict(GV1)) raises(GradeNotPublishedError, s.open_price_window, "2.3 price window refused after reveals, before reconcile") # 2.4 after reconcile but before publish_grade — the subtle one. The grade is # known at this point; it is not yet published. The window must still refuse, # because "the station could compute it" is not the same as "both parties are # bound to it". s = fresh() phase1(s, GV1, dict(GV1)) check(s.reconciliation.agreed_grade is not None, "2.4a grade is derivable at this point") raises(GradeNotPublishedError, s.open_price_window, "2.4 price window refused after reconcile, before publish_grade") # 2.5 price commit without an open window s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() raises(PhaseOrderError, s.commit_price, "2.5 price commit refused before window opens", HOLDER, 40500) # 2.6 price commit on a station that never ran phase 1 at all s = fresh() raises(PhaseOrderError, s.commit_price, "2.6 price commit refused with no grade round", HOLDER, 40500) # 2.7 settle without prices s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() s.open_price_window() raises(PhaseOrderError, s.settle, "2.7 settle refused before reveals") raises(PhaseOrderError, s.reveal_prices, "2.8 reveal refused before commits") # 2.9 publish_grade before reconcile s = fresh() s.commit_sheet(HOLDER, GV1) s.commit_sheet(INSPECTOR, dict(GV1)) s.reveal_sheet(HOLDER, GV1) s.reveal_sheet(INSPECTOR, dict(GV1)) raises(PhaseOrderError, s.publish_grade, "2.9 publish_grade refused before reconcile") # --------------------------------------------------------------------------- # 3. Void rounds are terminal — no override path # --------------------------------------------------------------------------- s = fresh() rec = phase1(s, RV3_H, RV3_I) check(rec.status == "void", "3.1 machine conflict voids") check("power_on_hours" in rec.machine_conflicts, "3.2 power_on_hours named as a machine conflict") check(s.phase is Phase.VOID, "3.3 station enters VOID") raises(GradeRoundVoidError, s.open_price_window, "3.4 void blocks the price window") raises(GradeRoundVoidError, s.publish_grade, "3.5 void blocks publish_grade") raises(GradeRoundVoidError, s.commit_price, "3.6 void blocks price commit", HOLDER, 40500) raises(GradeRoundVoidError, s.settle, "3.7 void blocks settle") raises(GradeRoundVoidError, s.reconcile, "3.8 void blocks re-reconcile") # A void round voids even when the two derived grades would have agreed. # Class M equality is checked before any grade is derived, so this must void # despite both sheets landing on the same letters. a = sheet(**{**FLEET_GOOD, "power_on_hours": 9800}) b2 = sheet(**{**FLEET_GOOD, "power_on_hours": 9805}) check(gr.grade(a).code == gr.grade(b2).code, "3.9a both sheets derive the same grade") s = fresh() rec = phase1(s, a, b2) check(rec.status == "void", "3.9 machine conflict voids even when derived grades agree") # --------------------------------------------------------------------------- # 4. Material divergence blocks until resolved # --------------------------------------------------------------------------- s = fresh() rec = phase1(s, RV2_H, RV2_I) check(rec.status == "material", "4.1 RV-2 is material") check(rec.pivotal_fields == ["dents_ge_6mm"], f"4.2 pivotal field is dents_ge_6mm, got {rec.pivotal_fields}") check(s.phase is Phase.UNRESOLVED, "4.3 station enters UNRESOLVED") raises(GradeUnresolvedError, s.open_price_window, "4.4 material divergence blocks the price window") raises(GradeUnresolvedError, s.commit_price, "4.5 material divergence blocks price commit", HOLDER, 40500) # Re-inspection resolves it: a fresh round on agreed re-read values proceeds. s = fresh() rec = phase1(s, RV2_I, dict(RV2_I)) check(rec.status == "agreed", "4.6 re-inspection to a shared value agrees") s.publish_grade() s.open_price_window() check(s.phase is Phase.PRICE_OPEN, "4.7 resolved round opens the price window") # --------------------------------------------------------------------------- # 5. Immaterial divergence proceeds, on a canonical merged sheet # --------------------------------------------------------------------------- s = fresh() rec = phase1(s, RV1_H, RV1_I) check(rec.status == "immaterial", "5.1 RV-1 is immaterial") check(rec.pivotal_fields == [], "5.2 no pivotal fields when immaterial") b = s.publish_grade() merged = gr.conservative_merge(RV1_H, RV1_I) check(b.grade_token == gr.grade_token(merged), "5.3 token is over the conservative merge, not either party's sheet") check(b.grade_token != gr.grade_token(RV1_H), "5.4 token is not the optimistic party's own token") # In RV-1 the inspector happens to be worse on EVERY differing field, so the # conservative merge equals the inspector's sheet and the tokens coincide. # That is the rule working, not a collision: the party reporting the worse # condition sets the record when the disagreement is immaterial. check(b.grade_token == gr.grade_token(RV1_I), "5.5 when one party is worse on every field, the merge is that sheet") check(all(merged[k] == RV1_I[k] for k in gr.FIELDS), "5.5b merge equals the uniformly-worse sheet") # The canonicalization only earns its keep when each party is worse on a # DIFFERENT field, because then the merge matches neither sheet and a naive # "use the holder's token" rule would publish a token the inspector never # derived. Constructed so both sheets still derive the same grade. X_H = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 4, "scratches_ge_10mm": 8}) X_I = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 3, "scratches_ge_10mm": 11}) check(gr.grade(X_H).code == gr.grade(X_I).code, "5.5c crossed divergence still derives one grade") sx = fresh() recx = phase1(sx, X_H, X_I) check(recx.status == "immaterial", "5.5d crossed divergence is immaterial") bx = sx.publish_grade() mx = gr.conservative_merge(X_H, X_I) check(mx["dents_ge_2mm"] == 4 and mx["scratches_ge_10mm"] == 11, "5.5e merge takes the worse value from each side independently") check(bx.grade_token == gr.grade_token(mx), "5.5f token is over the crossed merge") check(bx.grade_token != gr.grade_token(X_H) and bx.grade_token != gr.grade_token(X_I), "5.5g crossed merge token matches neither party's sheet") check(gr.grade(merged).code == rec.agreed_grade.code, "5.6 merged sheet derives the agreed grade") check(b.reconciliation_status == "immaterial", "5.7 binding records that divergence occurred") s.open_price_window() check(s.phase is Phase.PRICE_OPEN, "5.8 immaterial divergence still settles") # Determinism: the two parties, in either order, reach the same token. s2 = fresh() phase1(s2, RV1_H, RV1_I) check(s2.publish_grade().grade_token == b.grade_token, "5.9 token is deterministic across runs") # --------------------------------------------------------------------------- # 6. The binding is load-bearing: a price cannot carry an unagreed grade # --------------------------------------------------------------------------- def priced_station(): """A station sitting at PRICE_OPEN on the reference unit.""" x = fresh() phase1(x, GV1, dict(GV1)) x.publish_grade() x.open_price_window() return x # One station per case, so removing the guard reports a named failure instead # of cascading into an unrelated error further down the file. for label, role, bogus in [ ("6.1 a better grade than agreed is refused", HOLDER, "F6/C6/S4X"), ("6.2 a worse grade than agreed is refused", INSPECTOR, "F3/C1/S0"), ("6.3 the retired A1-D4 token space is refused", HOLDER, "B2"), ("6.3b an empty grade is refused", HOLDER, ""), ]: raises(GradeTokenMismatchError, priced_station().commit_price, label, role, 40500, bogus) # The agreed grade, passed explicitly, is accepted. s = priced_station() s.commit_price(HOLDER, 40500, "F6/C6/S4") check(HOLDER in s.price_commits, "6.4 the agreed grade is accepted explicitly") # And the commitment really contains it. check(s.price_reveals[HOLDER]["grade"] == "F6/C6/S4", "6.5 commitment body carries the agreed grade") check(att.verify_reveal(s.price_commits[HOLDER], s.price_reveals[HOLDER]), "6.6 the price commitment verifies") # --------------------------------------------------------------------------- # 7. Sealed-sheet discipline # --------------------------------------------------------------------------- # 7.1 no reveal before BOTH parties commit — otherwise the second party copies s = fresh() s.commit_sheet(HOLDER, GV1) raises(PhaseOrderError, s.reveal_sheet, "7.1 reveal refused while counterparty is still writing", HOLDER, GV1) # 7.2 no amending a sealed sheet s = fresh() s.commit_sheet(HOLDER, GV1) raises(AlreadySubmittedError, s.commit_sheet, "7.2 a sealed sheet cannot be re-committed", HOLDER, GV5) # 7.3 a tampered reveal is caught s = fresh() s.commit_sheet(HOLDER, GV1) s.commit_sheet(INSPECTOR, dict(GV1)) tampered = sheet(**{**FLEET_GOOD, "dents_ge_2mm": 0, "battery_cycles": 10}) raises(RevealMismatchError, s.reveal_sheet, "7.3 a sheet edited after sealing is rejected", HOLDER, tampered) # 7.4 double reveal, while the round is still in the reveal phase s = fresh() s.commit_sheet(HOLDER, GV1) s.commit_sheet(INSPECTOR, dict(GV1)) s.reveal_sheet(HOLDER, GV1) raises(AlreadySubmittedError, s.reveal_sheet, "7.4 a party cannot reveal twice", HOLDER, GV1) # 7.4b once both have revealed, the phase itself has moved on, so a late # reveal is refused as an ordering error rather than a duplicate. s = fresh() phase1(s, GV1, dict(GV1)) raises(PhaseOrderError, s.reveal_sheet, "7.4b reveal refused once the reveal phase has closed", HOLDER, GV1) # 7.5 double price commit s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade(); s.open_price_window() s.commit_price(HOLDER, 40500) raises(AlreadySubmittedError, s.commit_price, "7.5 a party cannot re-commit a price", HOLDER, 39000) # 7.6 unknown role s = fresh() raises(StationError, s.commit_sheet, "7.6 unknown role rejected", "auditor", GV1) # 7.7 phases cannot be re-run s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() raises(PhaseOrderError, s.publish_grade, "7.7 grade cannot be published twice") s.open_price_window() raises(PhaseOrderError, s.open_price_window, "7.8 the price window cannot be reopened") # --------------------------------------------------------------------------- # 8. Backward compatibility — the published SCDA1 constants must not move # # This is the reason the binding lives in station.py instead of Policy. # --------------------------------------------------------------------------- LEGACY_POLICY_HASH = ( "f5bbc16aaba884177e6581aa528aa6b84c489aa842e592a9998909154ed5c0c1" ) check(att.policy_hash(att.Policy()) == LEGACY_POLICY_HASH, "8.1 SPEC.md policy_hash is unchanged by the station") s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() bph1 = s.open_price_window() check(bph1 != LEGACY_POLICY_HASH, "8.2 the bound policy hash is distinct from the legacy hash") check(len(bph1) == 64, "8.3 bound policy hash is a sha256 hex digest") # The bound hash must MOVE when the grade moves — otherwise it is not binding. s = fresh() phase1(s, GV5, dict(GV5)) s.publish_grade() bph2 = s.open_price_window() check(bph2 != bph1, "8.4 a different grade produces a different bound policy hash") # And be stable for the same inputs. s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() check(s.open_price_window() == bph1, "8.5 bound policy hash is deterministic") # A different tolerance must also move it. s = Station(round_id="R-TEST", asset_id="UNIT-TEST", policy=att.Policy(tolerance_bp=500)) phase1(s, GV1, dict(GV1)) s.publish_grade() check(s.open_price_window() != bph1, "8.6 policy changes still move the bound hash") # Settlement arithmetic is untouched: SPEC.md section 10 vectors still hold # when driven through the station. for h_c, i_c, idx, expect, label in [ (40500, 41000, None, 40750, "TV-1 midpoint"), (42000, 30000, 30000, None, "TV-5 escalation against the index"), ]: s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade(); s.open_price_window() s.commit_price(HOLDER, h_c); s.commit_price(INSPECTOR, i_c) s.reveal_prices() got = s.settle(index_cents=idx) direct = att.settle(h_c, i_c, att.Policy(), index_cents=idx) check(got.as_dict() == direct.as_dict(), f"8.7 {label}: station settlement equals attestation.settle") if expect is not None: check(got.amount_cents == expect, f"8.8 {label} amount") # An escalating round with no index value must refuse rather than guess. s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade(); s.open_price_window() s.commit_price(HOLDER, 42000); s.commit_price(INSPECTOR, 30000) s.reveal_prices() raises(ValueError, s.settle, "8.9 escalation with no index value refuses to settle") # --------------------------------------------------------------------------- # 9. Rubric binding # --------------------------------------------------------------------------- raises(RubricMismatchError, Station, "9.1 a station on the wrong rubric refuses to open", round_id="R", asset_id="U", expected_rubric_hash="00" * 32) s = Station(round_id="R", asset_id="U", expected_rubric_hash=gr.rubric_hash()) check(s.phase is Phase.GRADE_OPEN, "9.2 the correct rubric hash opens") check(s.binding is None, "9.3 no binding before phase 1") # --------------------------------------------------------------------------- # 10. A recovery-tier unit still gets a price round # # Recovery is not zero value — parts and materials still settle. The gate is # about ORDER, not about refusing to transact low grades. Asserting this stops # a future contributor from "improving" the gate into a value judgement. # --------------------------------------------------------------------------- s = fresh() rec = phase1(s, GV5, dict(GV5)) b = s.publish_grade() check(b.tier == "R", f"10.1 unsanitized unit is tier R, got {b.tier}") check(b.grade_code.endswith("S0"), "10.2 unsanitized unit is S0") s.open_price_window() s.commit_price(HOLDER, 3600) s.commit_price(INSPECTOR, 3600) s.reveal_prices() r = s.settle() check(r.amount_cents == 3600, "10.3 a recovery unit still settles") check(s.seal(att.Chain())["body"]["grade_binding"]["tier"] == "R", "10.4 the record shows it settled as recovery") # --------------------------------------------------------------------------- # 11. Transcript and state are honest about the phase # --------------------------------------------------------------------------- s = fresh() phase1(s, RV3_H, RV3_I) state = s.state() check(state["phase"] == "void", "11.1 state reports void") check(state["grade_binding"] is None, "11.2 no binding on a void round") check(state["reconciliation"]["status"] == "void", "11.3 reconciliation in state") check("VOID" in s.transcript(), "11.4 transcript records the void") s = fresh() phase1(s, GV1, dict(GV1)) s.publish_grade() check(json.loads(json.dumps(s.state()))["grade_binding"]["grade_code"] == "F6/C6/S4", "11.5 state is JSON-serializable") # --------------------------------------------------------------------------- # 12. run_round convenience wrapper honours the same gates # --------------------------------------------------------------------------- full = st.run_round("R-9", "UNIT-9", GV1, dict(GV1), 40500, 41000) check(full.phase is Phase.SEALED, "12.1 run_round completes a legal round") check(full.binding.grade_code == "F6/C6/S4", "12.2 run_round grades correctly") raises(GradeRoundVoidError, st.run_round, "12.3 run_round raises rather than returning a partial void round", "R-10", "UNIT-10", RV3_H, RV3_I, 40500, 41000) raises(GradeUnresolvedError, st.run_round, "12.4 run_round raises on material divergence", "R-11", "UNIT-11", RV2_H, RV2_I, 40500, 41000) # --------------------------------------------------------------------------- # Report # --------------------------------------------------------------------------- if FAILS: print(f"FAIL — {len(FAILS)} of {CHECKS} checks failed:") for f in FAILS: print(" -", f) sys.exit(1) print(f"PASS — {CHECKS} checks: two-phase ordering enforced, void and") print(" unresolved rounds terminal, grade binding load-bearing,") print(" SCDA1 policy_hash unchanged") print(f" station = {st.STATION_VERSION}") print(f" rubric_hash = {gr.rubric_hash()}") print(f" legacy policy_hash = {att.policy_hash(att.Policy())}") # --------------------------------------------------------------------------- # Sabotage log — evidence these tests are not vacuous # # Each guard in station.py was removed on its own and the suite re-run, to # confirm the guard is load-bearing rather than decorative: # # guard removed tests that broke # ----------------------------------------- ------------------------------ # `if self.binding is None: raise` in 2.1, 2.2, 2.3, 2.4 # open_price_window # VOID branch in _require 3.5, 3.6, 3.7, 3.8, 12.3 # (3.4 survives: open_price_window carries # its own explicit VOID check, so the # gate is defended twice on purpose) # UNRESOLVED branch in _require 4.5, 12.4 # (4.4 survives for the same reason as # 3.4 — open_price_window checks both # blocked phases itself before calling # _require) # grade_code != agreed in commit_price 6.1, 6.2, 6.3, 6.3b # both-committed gate in reveal_sheet 7.1 # _ct_eq check in reveal_sheet 7.3 # conservative_merge in publish_grade 5.3, 5.4, 5.5, 5.5f, 5.5g # # Results recorded 29 Aug 2026. Re-run this exercise if the guards are # refactored; a guard with no failing test behind it is not a guard. # ---------------------------------------------------------------------------