import copy import hashlib import json import os import subprocess import tempfile import unittest from pathlib import Path from harness.lib import workflow_guard as guard def inventory_digest(ids, statuses): value = {'ids': sorted(ids), 'statuses': {key: statuses[key] for key in sorted(ids)}} return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest() def make_test_inventory(ids=('case_a',), current=None): current = list(ids if current is None else current) ids = list(ids) old_status = {key: {'skip': False, 'expected_failure': False} for key in ids} new_status = {key: {'skip': False, 'expected_failure': False} for key in current} def collection(names, statuses): return {'source_revision': 'a' * 40, 'command_id': 'collect-tests', 'argv': ['python3', '-B', '-m', 'unittest', 'discover'], 'inventory_sha256': inventory_digest(names, statuses), 'evidence': 'fixture-collection-record'} return {'baseline_ids': ids, 'current_ids': current, 'baseline_status': old_status, 'current_status': new_status, 'baseline_collection': collection(ids, old_status), 'current_collection': collection(current, new_status), 'authorized_changes': [], 'changed_tests': [], 'findings': []} def contract_inventory(): return {'baseline': [{'semantic_key': 'ready-fields', 'baseline_revision': 'a' * 40, 'path': 'src/broker.py', 'meaning': 'Public ready fields are handle/socket.'}], 'current': [{'semantic_key': 'ready-fields', 'meaning': 'Public ready fields are handle/socket.', 'observation': 'observed', 'evidence': 'fixture-response-check'}], 'mappings': [{'semantic_key': 'ready-fields', 'current_keys': ['ready-fields'], 'status': 'preserved', 'authorization': '', 'verification': 'fixture-static-contract'}], 'stages': ['runtime', 'ready', 'admission'], 'precedence': [['runtime', 'ready'], ['ready', 'admission']]} def finding_record(): def execution(code, kind): return {'test_id': 'case_a', 'source_revision': 'a' * 40, 'scope_sha256': 'b' * 64, 'command_id': 'named-regression', 'argv': ['python3', '-B', '-m', 'unittest', 'case_a'], 'exit_code': code, 'expected_exit_code': code, 'failure_kind': kind, 'evidence': 'fixture-execution-record'} return {'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion', 'red': execution(1, 'assertion'), 'green': execution(0, 'pass')} class WorkflowGuardTests(unittest.TestCase): def repository(self, directory): root = Path(directory) / 'repo' root.mkdir(mode=0o700) subprocess.run(['git', 'init', '-q', str(root)], check=True) subprocess.run(['git', '-C', str(root), '-c', 'user.name=Fixture', '-c', 'user.email=fixture@example.invalid', 'commit', '--allow-empty', '-qm', 'baseline'], check=True) (root / 'src').mkdir(mode=0o700) (root / 'src/a.py').write_text('one') return root def assert_block(self, function, value): try: result = function(value) except guard.GuardError: return self.assertEqual(result['status'], 'block') def test_snapshot_compare_and_change_detection(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) out = Path(td) / 'snapshot.json' guard.snapshot(root, ['src/a.py'], out) self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'pass') (root / 'src/a.py').write_text('two') self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block') def test_rejects_traversal_and_malformed_baseline(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) with self.assertRaises(guard.GuardError): guard.snapshot(root, ['../outside'], Path(td) / 'x.json') bad = Path(td) / 'bad.json' bad.write_text('{}') with self.assertRaises(guard.GuardError): guard.compare_source(root, bad, ['src/a.py']) def test_test_inventory_requires_identity_and_review_evidence(self): self.assert_block(guard.check_tests, {'baseline_ids': ['a'], 'current_ids': ['b']}) def test_contract_inventory_requires_full_coverage(self): self.assert_block(guard.check_contracts, {'baseline': [{'semantic_key': 'x'}], 'current': [], 'mappings': []}) def test_snapshot_rejects_literal_root_alias_and_symlink_root(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) alias = Path(td) / 'alias' alias.symlink_to(root, target_is_directory=True) for value in (str(root) + '/./', alias): with self.subTest(root=str(value)), self.assertRaises(guard.GuardError): guard.snapshot(value, ['src/a.py'], Path(td) / 'snapshot.json') def test_snapshot_rejects_symlink_output_parent(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) actual = Path(td) / 'private' actual.mkdir(mode=0o700) alias = Path(td) / 'alias' alias.symlink_to(actual, target_is_directory=True) with self.assertRaises(guard.GuardError): guard.snapshot(root, ['src/a.py'], alias / 'snapshot.json') self.assertFalse((actual / 'snapshot.json').exists()) def test_snapshot_exact_mode_survives_restrictive_umask(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) out = Path(td) / 'snapshot.json' before = os.umask(0o777) try: guard.snapshot(root, ['src/a.py'], out) finally: os.umask(before) self.assertEqual(out.stat().st_mode & 0o777, 0o600) def test_snapshot_rejects_boolean_schema_baseline(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) out = Path(td) / 'snapshot.json' guard.snapshot(root, ['src/a.py'], out) value = json.loads(out.read_text()) value['schema'] = True out.write_text(json.dumps(value)) with self.assertRaises(guard.GuardError): guard.compare_source(root, out, ['src/a.py']) def test_snapshot_rejects_repository_subdirectory_root(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) with self.assertRaises(guard.GuardError): guard.snapshot(root / 'src', ['a.py'], Path(td) / 'snapshot.json') def test_snapshot_detects_same_bytes_inode_replacement(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) out = Path(td) / 'snapshot.json' guard.snapshot(root, ['src/a.py'], out) replacement = root / 'src/new.py' replacement.write_text('one') replacement.replace(root / 'src/a.py') self.assertEqual(guard.compare_source(root, out, ['src/a.py'])['status'], 'block') def test_new_regression_tests_do_not_require_deletion_authorization(self): self.assertEqual(guard.check_tests(make_test_inventory(current=['case_a', 'case_b']))['status'], 'pass') def test_collected_inventory_requires_provenance(self): value = make_test_inventory() del value['baseline_collection'] self.assert_block(guard.check_tests, value) def test_collected_inventory_digest_must_match_ids_and_statuses(self): value = make_test_inventory() value['baseline_collection']['inventory_sha256'] = '0' * 64 self.assert_block(guard.check_tests, value) def test_new_skip_requires_explicit_approval(self): value = make_test_inventory() value['current_status']['case_a']['skip'] = True value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status']) self.assert_block(guard.check_tests, value) def test_new_expected_failure_requires_explicit_approval(self): value = make_test_inventory() value['current_status']['case_a']['expected_failure'] = True value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status']) self.assert_block(guard.check_tests, value) def test_rename_requires_real_replacement_and_semantic_evidence(self): value = make_test_inventory(current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['ghost'], 'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assert_block(guard.check_tests, value) def test_approved_rename_preserves_mapped_test(self): value = make_test_inventory(current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': ['case_b'], 'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assertEqual(guard.check_tests(value)['status'], 'pass') def test_modified_test_requires_semantic_review(self): value = make_test_inventory() value['changed_tests'] = [{'test_id': 'case_a', 'baseline_sha256': 'a' * 64, 'current_sha256': 'b' * 64, 'review_evidence': ''}] self.assert_block(guard.check_tests, value) def test_finding_truthy_strings_are_not_red_green_records(self): value = make_test_inventory() value['findings'] = [{'id': 'finding-a', 'test_id': 'case_a', 'classification': 'assertion', 'red': 'claimed-red', 'green': 'claimed-green'}] self.assert_block(guard.check_tests, value) def test_red_success_exit_cannot_demonstrate_failure(self): value = make_test_inventory() finding = finding_record() finding['red']['exit_code'] = finding['red']['expected_exit_code'] = 0 value['findings'] = [finding] self.assert_block(guard.check_tests, value) def test_green_failure_exit_cannot_demonstrate_resolution(self): value = make_test_inventory() finding = finding_record() finding['green']['exit_code'] = finding['green']['expected_exit_code'] = 1 value['findings'] = [finding] self.assert_block(guard.check_tests, value) def test_environment_failure_is_not_valid_red(self): value = make_test_inventory() finding = finding_record() finding['red']['failure_kind'] = 'environment' value['findings'] = [finding] self.assert_block(guard.check_tests, value) def test_finding_must_name_present_regression_test(self): value = make_test_inventory() finding = finding_record() finding['test_id'] = 'ghost' value['findings'] = [finding] self.assert_block(guard.check_tests, value) def test_named_red_green_records_with_provenance_pass(self): value = make_test_inventory() value['findings'] = [finding_record()] self.assertEqual(guard.check_tests(value)['status'], 'pass') def test_missing_and_ambiguous_contracts_cannot_be_approved_away(self): for status in ('missing', 'ambiguous'): with self.subTest(status=status): value = contract_inventory() value['mappings'][0].update(status=status, authorization='fixture-approval') self.assert_block(guard.check_contracts, value) def test_preserved_contract_requires_existing_current_mapping(self): value = contract_inventory() value['current'] = [] self.assert_block(guard.check_contracts, value) def test_unknown_observation_is_not_actual_public_response_evidence(self): value = contract_inventory() value['current'][0].update(observation='unknown', evidence='') self.assert_block(guard.check_contracts, value) def test_preserved_contract_requires_verification(self): value = contract_inventory() value['mappings'][0]['verification'] = '' self.assert_block(guard.check_contracts, value) def test_precedence_cycle_blocks_startup_order(self): value = contract_inventory() value['precedence'].append(['admission', 'runtime']) self.assert_block(guard.check_contracts, value) def test_complete_observed_contract_and_acyclic_order_pass(self): self.assertEqual(guard.check_contracts(contract_inventory())['status'], 'pass') def test_duplicate_current_contract_is_not_coverage(self): value = contract_inventory() value['current'].append(copy.deepcopy(value['current'][0])) self.assert_block(guard.check_contracts, value) def test_invalid_stage_types_fail_closed_without_unhandled_typeerror(self): value = contract_inventory() value['stages'] = [['unhashable']] self.assert_block(guard.check_contracts, value) def guard_cli(self, *args): command = ['python3', '-B', str(Path(guard.__file__).resolve()), *map(str, args)] try: return subprocess.run(command, capture_output=True, text=True, timeout=3) except subprocess.TimeoutExpired: self.fail('guard did not fail closed within 3 seconds') def test_snapshot_fifo_protected_file_fails_fast(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) os.mkfifo(root / 'src/pipe') paths = Path(td) / 'paths.json' paths.write_text('["src/pipe"]') result = self.guard_cli('snapshot', '--repo-root', root, '--paths-json', paths, '--output', Path(td) / 'snapshot.json') self.assertEqual(result.returncode, 2) self.assertEqual(json.loads(result.stdout)['status'], 'block') def test_compare_source_fifo_protected_file_fails_fast(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) out = Path(td) / 'snapshot.json' guard.snapshot(root, ['src/a.py'], out) (root / 'src/a.py').unlink() os.mkfifo(root / 'src/a.py') paths = Path(td) / 'paths.json' paths.write_text('["src/a.py"]') result = self.guard_cli('compare-source', '--repo-root', root, '--paths-json', paths, '--before', out) self.assertEqual(result.returncode, 1) # Valid baseline; observed source-type drift. self.assertEqual(json.loads(result.stdout)['status'], 'block') def test_snapshot_rejects_root_ending_dotdot(self): with tempfile.TemporaryDirectory() as td: root = self.repository(td) with self.assertRaises(guard.GuardError): guard.snapshot(str(root / 'src') + '/..', ['src/a.py'], Path(td) / 'snapshot.json') def test_snapshot_rejects_ancestor_symlink_root(self): with tempfile.TemporaryDirectory() as td: self.repository(td) alias = Path(td) / 'alias' alias.symlink_to(td, target_is_directory=True) with self.assertRaises(guard.GuardError): guard.snapshot(alias / 'repo', ['src/a.py'], Path(td) / 'snapshot.json') def test_json_fifo_input_fails_fast(self): with tempfile.TemporaryDirectory() as td: pipe = Path(td) / 'pipe.json' os.mkfifo(pipe) result = self.guard_cli('check-tests', '--input', pipe) self.assertEqual(result.returncode, 2) self.assertEqual(json.loads(result.stdout)['status'], 'block') def test_deeply_nested_json_blocks_without_traceback(self): with tempfile.TemporaryDirectory() as td: bad = Path(td) / 'bad.json' bad.write_text('[' * 2000 + ']' * 2000) result = self.guard_cli('check-tests', '--input', bad) self.assertEqual(result.returncode, 2) self.assertNotIn('Traceback', result.stderr) self.assertEqual(json.loads(result.stdout)['status'], 'block') def test_rename_empty_replacement_cannot_stand_in_for_deletion(self): value = make_test_inventory(current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'rename', 'replacement_ids': [], 'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assert_block(guard.check_tests, value) def test_unclassified_empty_replacement_cannot_stand_in_for_deletion(self): value = make_test_inventory(current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'replacement_ids': [], 'reason': 'rename', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assert_block(guard.check_tests, value) def test_explicit_approved_deletion_is_allowed(self): value = make_test_inventory(ids=['case_a', 'case_b'], current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'delete', 'replacement_ids': [], 'reason': 'approved obsolete case deletion', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assertEqual(guard.check_tests(value)['status'], 'pass') def test_status_approval_cannot_authorize_removal(self): value = make_test_inventory(current=['case_b']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_b'], 'reason': 'approved status update', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assert_block(guard.check_tests, value) def test_explicit_status_approval_maps_same_current_id(self): value = make_test_inventory() value['current_status']['case_a']['skip'] = True value['current_collection']['inventory_sha256'] = inventory_digest(value['current_ids'], value['current_status']) value['authorized_changes'] = [{'old_id': 'case_a', 'kind': 'status', 'replacement_ids': ['case_a'], 'reason': 'approved skip', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assertEqual(guard.check_tests(value)['status'], 'pass') def test_unhashable_authorization_id_fails_closed_api(self): value = make_test_inventory() value['authorized_changes'] = [{'old_id': ['case_a'], 'replacement_ids': [], 'reason': 'fixture', 'approval': 'fixture-approval', 'evidence': 'fixture-review'}] self.assert_block(guard.check_tests, value) def test_approved_contract_substitution_can_map_new_current_key(self): value = contract_inventory() value['current'][0]['semantic_key'] = 'published-ready-fields' value['mappings'][0].update(current_keys=['published-ready-fields'], status='intentionally_changed', authorization='fixture-approved-substitution') self.assertEqual(guard.check_contracts(value)['status'], 'pass') def test_approved_contract_merge_retains_each_old_semantic_mapping(self): value = contract_inventory() value['baseline'].append({'semantic_key': 'state-fields', 'baseline_revision': 'a' * 40, 'path': 'src/broker.py', 'meaning': 'Public state is held.'}) value['current'][0].update(semantic_key='combined-fields', meaning='Public fields are handle/socket and state held.') value['mappings'] = [{'semantic_key': key, 'current_keys': ['combined-fields'], 'status': 'intentionally_changed', 'authorization': 'fixture-approved-merge', 'verification': 'fixture-checked-each-old-meaning'} for key in ('ready-fields', 'state-fields')] self.assertEqual(guard.check_contracts(value)['status'], 'pass') def test_schema_docstring_lists_nested_fields_and_exit_rules(self): for field in ('baseline_status', 'current_status', 'skip', 'expected_failure', 'authorized_changes', 'old_id', 'kind', 'replacement_ids', 'changed_tests', 'baseline_sha256', 'current_sha256', 'review_evidence', 'findings', 'scope_sha256', 'expected_exit_code', 'failure_kind', 'baseline_revision', 'semantic_key', 'current_keys', 'observation', 'authorization', 'verification', 'precedence', "sort_keys=True", "separators=(',', ':')", 'RED', 'GREEN', 'delete', 'rename', 'status'): with self.subTest(field=field): self.assertIn(field, guard.__doc__ or '') def test_declared_evidence_shape_does_not_authenticate_records(self): # Deliberately synthetic fixture claims: the main agent must check actual evidence separately. self.assertEqual(guard.check_tests(make_test_inventory())['status'], 'pass') def test_unknown_inventory_fields_are_rejected(self): value = make_test_inventory() value['pretend_verified'] = True self.assert_block(guard.check_tests, value) if __name__ == '__main__': unittest.main()