#!/usr/bin/env python
"""Put each key defect back, run the tests that guard it, and require them to FAIL — then restore the code.
python scripts/mutation_check.py # every mutation
python scripts/mutation_check.py tc1 # one
A test suite that still passes with the defect in place is not protecting the rule (AGENTS.md: verify by breaking).
"""
import pathlib
import subprocess
import sys
ROOT = pathlib.Path(__file__).resolve().parents[1]
SRC = ROOT / "src" / "brick_hr"
PY = sys.executable
MUTATIONS = {
"bpmn_dependency": ("the BPMN picture drops a dependency edge (a step looks free to start early)",
SRC / "domain" / "bpmn.py",
" for d in deps[n]:\n src =",
" for d in deps[n][1:]:\n src =",
["tests/test_rules.py::test_hr_11_every_gb_playbook_is_valid_bpmn_2_with_its_dependencies_as_flows"]),
"foundation_gate": ("a pack's legal gate can be dropped from a step (the right-to-work check before the account)",
SRC / "domain" / "deviations.py",
" if lost:\n cls = \"foundation\"",
" if lost:\n cls = \"variation\"",
["tests/test_process.py::test_dropping_a_pack_gate_is_the_kept_foundation_and_refused_at_save"]),
"stable_keys": ("BPMN ids go back to step positions (inserting a step shifts every later id)",
SRC / "domain" / "bpmn.py",
' tid = f"Task_{s.key}"',
' tid = f"Task_{s.step_no}"',
["tests/test_process.py::test_inserting_a_step_changes_no_other_elements_id"]),
"tc1": ("TC1: an overlap is not refused",
SRC / "domain" / "slices.py",
" if nxt.valid_from <= prev.valid_to:\n raise RuleBroken(",
" if False:\n raise RuleBroken(",
["tests/test_rules.py::test_tc1_refuses_a_gap_and_an_overlap",
"tests/test_rules.py::test_tc1_refuses_an_overlap_however_it_got_there"]),
"delimit": ("TC1: a change does not delimit the slice before it",
SRC / "domain" / "slices.py",
" cur.valid_to = valid_from - DAY\n",
" pass\n",
["tests/test_rules.py", "tests/test_scenarios.py::test_scenario[SC-HR-02]"]),
"sensitive_404": ("a manager reaches sensitive data",
SRC / "application" / "sensitive_data.py",
" if not S.has(caller, perm):\n raise Refused(404,",
" if False:\n raise Refused(404,",
["tests/test_doors.py::test_hr_13_a_manager_gets_404_on_every_sensitive_route_for_their_own_report_both_doors",
"tests/test_scenarios.py::test_scenario[PROBE-PERM-MANAGER]"]),
"sensitive_403": ("a manager is told 403 (which confirms the record exists) instead of 404",
SRC / "application" / "sensitive_data.py",
" raise Refused(404, f\"no sensitive data for person {person_id}\")\n\n\n@HR",
" raise Refused(403, f\"missing permission: {perm}\")\n\n\n@HR",
["tests/test_scenarios.py::test_scenario[PROBE-PERM-MANAGER]"]),
"version_freeze": ("an open run is re-expanded from the latest playbook version on every tick",
SRC / "application" / "engine.py",
" if ev is None or not ev.run_id or uow.runs.exists(ev.run_id):\n return None\n view = _view(uow, at)\n run = build_run(view, p, ev, at=at)\n uow.runs.save(run)",
" if ev is None or not ev.run_id:\n return None\n view = _view(uow, at)\n run = build_run(view, p, ev, at=at)\n old = uow.runs.get(ev.run_id)\n run.version = old.version if old else 0\n uow.runs.save(run)",
["tests/test_scenarios.py::test_scenario[SC-HR-25]"],
(' if e.get("run_id") and e["run_id"] not in existing:', ' if e.get("run_id"):')),
"one_run_per_hire": ("the run's id is minted at expansion, not derived: a replayed hire makes a second run",
SRC / "application" / "engine.py",
" if ev is None or not ev.run_id or uow.runs.exists(ev.run_id):\n return None",
" if ev is None or not ev.run_id:\n return None\n import uuid\n ev.run_id = ev.run_id + '-' + uuid.uuid4().hex[:6]",
["tests/test_scenarios.py::test_scenario[SC-HR-21]",
"tests/test_doors.py::test_hr_01_a_hire_through_mcp_is_the_same_hire"]),
"echo_sensitive": ("the public form shows a stored value back",
SRC / "application" / "intake.py",
" box = f'<input type=\"text\" name=\"{name}\" autocomplete=\"off\" value=\"\">' # never a stored value",
" box = f'<input type=\"text\" name=\"{name}\" value=\"{public.esc(str(_held(f)))}\">'",
["tests/test_doors.py::test_the_intake_form_through_the_public_door_never_echoes_a_value",
"tests/test_scenarios.py::test_scenario[SC-HR-22]"],
("def _page(form: dict, notice: str = \"\") -> public.Page:\n",
"def _held(f):\n from brick_hr.application.use_cases import repository\n with repository() as uow:\n"
" for pid in uow.persons.ids():\n s = uow.sensitive.get(pid)\n"
" for k, t in (s.kinds.items() if s else []):\n v = (t.last().data.get('value') or {})\n"
" if f['field'] in v:\n return v[f['field']]\n return ''\n\n\n"
"def _page(form: dict, notice: str = \"\") -> public.Page:\n")),
"approval": ("a destructive step fires without a granted approval",
SRC / "domain" / "run.py",
" if dep.kind == \"approval\" and dep.state == DONE and dep.approved is True:",
" if dep.kind == \"approval\":",
["tests/test_rules.py::test_hr_07_a_refused_approval_cancels_what_depends_on_it_and_a_granted_one_lets_it_act"]),
"bp_private": ("the business partner is sent a personal email",
SRC / "domain" / "services.py",
' return {"person_id": str(p.person_id), "legal_name": legal,',
' return {"personal_email": data.get("personal_email") or "", "person_id": str(p.person_id), "legal_name": legal,',
["tests/test_rules.py::test_hr_50_the_partner_card_carries_a_name_and_a_work_contact_only",
"tests/test_tier1.py::test_hr_50_the_partner_is_created_at_hire_with_a_name_and_nothing_private"]),
"tick_claim": ("a tick acts without claiming the step (an upsert-like claim that always wins)",
SRC / "infrastructure" / "repositories.py",
" self.conn.execute(insert(T.step_claim).values(",
" return None\n self.conn.execute(insert(T.step_claim).values(",
["tests/test_engine.py"]),
"config_token": ("a configuration change does not bump its version: the cached catalogue is served stale",
SRC / "infrastructure" / "repositories.py",
" done.update(todo)\n now = datetime.now(timezone.utc).replace(tzinfo=None)\n upsert(",
" done.update(todo)\n return\n now = datetime.now(timezone.utc).replace(tzinfo=None)\n upsert(",
["tests/test_performance.py::test_a_configuration_change_is_seen_by_the_next_request",
"tests/test_performance.py::test_a_playbook_and_an_org_change_are_seen_by_the_next_request"]),
"revoke_timing": ("offboarding.revoke_access_at is ignored: a dismissal set to immediate still waits for the last day",
SRC / "application" / "engine.py",
" timings[cfgm.REVOKES_ACCESS] = t",
" pass",
["tests/test_doors.py::test_hr_55_a_dismissal_set_to_immediate_revokes_the_login_once_the_access_removal_is_approved"]),
"monitor_business_date": ("the onboarding monitor counts days on the UTC date, not the business date",
SRC / "domain" / "monitoring.py",
" return work.aware(at).astimezone(work.zone(tz)).date()",
" return work.aware(at).date()",
["tests/test_rules.py::test_hr_29_a_late_evening_step_counts_on_its_business_date_not_the_utc_one"]),
"reason_condition": ("reason_is always holds: every leaver is offered outplacement, not only a redundancy",
SRC / "domain" / "run.py",
' r = (facts.get("reason") or "") in args',
" r = True",
["tests/test_doors.py::test_hr_42_a_resignation_skips_the_outplacement_offer"]),
"outcome_required": ("a step with outcomes completes with any answer (or none)",
SRC / "domain" / "run.py",
" if o not in self.outcomes:\n",
" if False:\n",
["tests/test_doors.py::test_hr_42_a_redundancy_offers_outplacement_and_records_the_answer"]),
"revoke_after_approval": ("the ERP sign-in is revoked without waiting for the access-removal approval",
SRC / "domain" / "packs" / "gb.py",
" depends_on=[2]), # after the access-removal approval",
" depends_on=[]),",
["tests/test_doors.py::test_hr_54_without_a_connector_revoking_the_erp_login_is_a_manual_task_with_the_exact_input"]),
"unique_race_at_commit": ("a claim lost at COMMIT on DataK3 answers 500 instead of the retry-safe 409",
SRC / "application" / "use_cases.py",
" if not _lost_a_unique_race(exc):\n raise\n",
" raise\n",
["tests/test_commit_conflict.py"]),
"contract_unsigned": ("a contract that expired unsigned is polled forever instead of becoming a task",
SRC / "application" / "engine.py",
'"EXPIRED_UNSIGNED": "left unsigned until it expired",',
"",
["tests/test_scenarios.py::test_scenario[SC-HR-30]"]),
}
def run(tests):
r = subprocess.run([PY, "-m", "pytest", "-q", "--tb=no", "-p", "no:cacheprovider", *tests], cwd=ROOT,
capture_output=True, text=True)
return r.returncode, (r.stdout.strip().splitlines() or [""])[-1]
def main(names):
results = []
for name in names or MUTATIONS:
title, path, old, new, tests, *extra = MUTATIONS[name]
original = path.read_text()
if old not in original:
results.append((name, "STALE", f"the code to mutate is not in {path.name}"))
continue
text = original.replace(old, new, 1)
if extra:
a, b = extra[0]
text = text.replace(a, b, 1)
try:
path.write_text(text)
code, last = run(tests)
finally:
path.write_text(original)
results.append((name, "CAUGHT" if code != 0 else "MISSED", f"{title}: {last}"))
for name, verdict, detail in results:
print(f"{verdict:7} {name:18} {detail}")
return 0 if all(v == "CAUGHT" for _, v, _ in results) else 1
if __name__ == "__main__":
raise SystemExit(main(sys.argv[1:]))