421 lines
21 KiB
Python
421 lines
21 KiB
Python
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 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(test_inventory(current=['case_a', 'case_b']))['status'], 'pass')
|
|
|
|
def test_collected_inventory_requires_provenance(self):
|
|
value = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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(test_inventory())['status'], 'pass')
|
|
|
|
def test_unknown_inventory_fields_are_rejected(self):
|
|
value = test_inventory()
|
|
value['pretend_verified'] = True
|
|
self.assert_block(guard.check_tests, value)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|