[verified] Add canonical harness test runner and enforce role collection
This commit is contained in:
@@ -0,0 +1,345 @@
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest import mock
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
REPO = Path(__file__).resolve().parents[3]
|
||||
RUNNER = REPO / "harness" / "tests" / "run.py"
|
||||
|
||||
|
||||
class CanonicalRunnerTests(unittest.TestCase):
|
||||
SCRATCH = Path("/opt/data/cache/scratch/canonical-tests-main/fix/test-tmp")
|
||||
|
||||
def make_suite(self, *, roles=None, extra="", nested=False):
|
||||
self.SCRATCH.mkdir(parents=True, exist_ok=True)
|
||||
temp = tempfile.TemporaryDirectory(dir=self.SCRATCH)
|
||||
self.addCleanup(temp.cleanup)
|
||||
root = Path(temp.name)
|
||||
tests = root / "harness" / "tests"
|
||||
roles = roles or {"common": "test_common", "spec": "test_spec", "impl": "test_impl", "review": "test_review"}
|
||||
(tests / "__init__.py").parent.mkdir(parents=True)
|
||||
(tests / "__init__.py").write_text("")
|
||||
(tests / "run.py").write_bytes(RUNNER.read_bytes())
|
||||
for role, stem in roles.items():
|
||||
d = tests / role
|
||||
d.mkdir()
|
||||
(d / "__init__.py").write_text("")
|
||||
(d / f"{stem}.py").write_text(
|
||||
"import unittest\n"
|
||||
"class Example(unittest.TestCase):\n"
|
||||
" def test_one(self):\n"
|
||||
" self.assertTrue(True)\n"
|
||||
)
|
||||
if nested:
|
||||
d = tests / "common" / "nested"
|
||||
d.mkdir()
|
||||
(d / "__init__.py").write_text("")
|
||||
(d / "test_nested.py").write_text(
|
||||
"import unittest\nclass Nested(unittest.TestCase):\n"
|
||||
" def test_nested(self): self.assertTrue(True)\n"
|
||||
)
|
||||
if extra:
|
||||
(tests / "test_outside.py").write_text(extra)
|
||||
return root
|
||||
|
||||
def execute(self, root, *args):
|
||||
env = dict(os.environ)
|
||||
env["PYTHONPATH"] = str(root) + os.pathsep + env.get("PYTHONPATH", "")
|
||||
return subprocess.run([sys.executable, "-B", str(root / "harness/tests/run.py"), *args], cwd="/", env=env,
|
||||
capture_output=True, text=True, timeout=20)
|
||||
|
||||
def test_real_canonical_collect_only_succeeds_with_every_role(self):
|
||||
result = subprocess.run([sys.executable, "-B", str(RUNNER), "--collect-only"], cwd="/",
|
||||
capture_output=True, text=True, timeout=30)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
data = json.loads(result.stdout)
|
||||
self.assertEqual(set(data["role_counts"]), {"common", "spec", "impl", "review"})
|
||||
self.assertTrue(all(data["role_counts"].values()))
|
||||
self.assertEqual(len(data["ids"]), sum(data["role_counts"].values()))
|
||||
self.assertTrue(data["ids"])
|
||||
|
||||
def test_module_load_tests_is_rejected(self):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text(
|
||||
"import unittest\ndef load_tests(loader, tests, pattern): return tests\n"
|
||||
"class X(unittest.TestCase):\n def test_one(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("load_tests", result.stderr)
|
||||
|
||||
def test_zero_file_and_mixed_free_function_fail(self):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text("import unittest\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("zero", result.stderr.lower())
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text('''import unittest
|
||||
class X(unittest.TestCase):
|
||||
def test_one(self): pass
|
||||
def test_free(): pass
|
||||
''')
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("free-function", result.stderr.lower())
|
||||
|
||||
def test_direct_case_does_not_short_circuit_cycle_or_unknown_base(self):
|
||||
for source, word in (
|
||||
('''import unittest
|
||||
class A(B, unittest.TestCase):
|
||||
def test_a(self): pass
|
||||
class B(A):
|
||||
def test_b(self): pass
|
||||
''', "cyclic"),
|
||||
('''import unittest
|
||||
class X(unittest.TestCase, Missing):
|
||||
def test_x(self): pass
|
||||
''', "unknown"),
|
||||
):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text(source)
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn(word, result.stderr.lower())
|
||||
|
||||
def test_package_load_tests_is_rejected(self):
|
||||
root = self.make_suite(nested=True)
|
||||
(root / "harness/tests/common/nested/__init__.py").write_text('''def load_tests(loader, tests, pattern):
|
||||
return tests
|
||||
''')
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("load_tests", result.stderr)
|
||||
|
||||
def test_alias_local_inheritance_and_runtime_helpers_are_valid(self):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text('''import unittest as ut
|
||||
class Mixin:
|
||||
def test_mix(self): pass
|
||||
class Base(ut.TestCase, Mixin):
|
||||
def test_base(self):
|
||||
class Helper:
|
||||
def test_not_discovered(self): pass
|
||||
self.assertTrue(Helper)
|
||||
class Child(Base):
|
||||
def test_child(self): pass
|
||||
''')
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_conditional_discovery_declarations_fail_closed(self):
|
||||
for source in (
|
||||
'''import unittest
|
||||
if True:
|
||||
class X(unittest.TestCase):
|
||||
def test_x(self): pass
|
||||
''',
|
||||
'''import unittest
|
||||
class X(unittest.TestCase):
|
||||
if True:
|
||||
def test_x(self): pass
|
||||
''',
|
||||
):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text(source)
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("conditional", result.stderr.lower())
|
||||
|
||||
def test_custom_loader_duplicate_and_omission_are_rejected(self):
|
||||
root = self.make_suite()
|
||||
marker = root / "admission-marker"
|
||||
script = root / "probe.py"
|
||||
script.write_text("""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, sys.argv[1])
|
||||
import harness.tests.run as run
|
||||
root = Path(sys.argv[1])
|
||||
files = run.source_files()
|
||||
expected, _, _ = run.inventory(files)
|
||||
baseline, suite = run.collect(files)
|
||||
assert not any(isinstance(t, run.unittest.loader._FailedTest) for t in run.flatten(suite))
|
||||
assert baseline == expected
|
||||
items = list(run.flatten(suite))
|
||||
class Fake(run.unittest.FunctionTestCase):
|
||||
def __init__(self, ident):
|
||||
super().__init__(lambda: Path(sys.argv[2]).write_text('ran'))
|
||||
self.ident = ident
|
||||
def id(self):
|
||||
return self.ident
|
||||
for mode in ('omission', 'addition', 'duplicate'):
|
||||
class Loader:
|
||||
def discover(self, *args, **kwargs):
|
||||
selected = list(items)
|
||||
if mode == 'omission': selected = selected[:-1]
|
||||
elif mode == 'addition': selected.append(Fake('fake.added'))
|
||||
else: selected = [Fake('fake.duplicate'), Fake('fake.duplicate')]
|
||||
return run.unittest.TestSuite(selected)
|
||||
run.unittest.defaultTestLoader = Loader()
|
||||
try:
|
||||
if mode == 'duplicate': run.collect(files)
|
||||
else: run.validate()
|
||||
except run.CanonicalError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError(mode + ' was admitted')
|
||||
assert not Path(sys.argv[2]).exists()
|
||||
""")
|
||||
result = subprocess.run([sys.executable, "-B", str(script), str(root), str(marker)], cwd="/",
|
||||
capture_output=True, text=True, timeout=20)
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_collect_only_reports_exact_roles_ids_and_hashes(self):
|
||||
root = self.make_suite()
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
data = json.loads(result.stdout)
|
||||
self.assertTrue(data["collection_only"])
|
||||
self.assertEqual(data["role_counts"], {"common": 1, "spec": 1, "impl": 1, "review": 1})
|
||||
self.assertEqual(len(data["ids"]), 4)
|
||||
self.assertEqual(set(data["source_sha256"]), {"harness/tests/common/test_common.py", "harness/tests/spec/test_spec.py", "harness/tests/impl/test_impl.py", "harness/tests/review/test_review.py"})
|
||||
|
||||
def test_missing_role_marker_duplicate_and_unexpected_file_fail_before_run(self):
|
||||
for mode in ("missing", "marker", "duplicate", "unexpected"):
|
||||
with self.subTest(mode=mode):
|
||||
root = self.make_suite(extra=("import unittest\nclass Example(unittest.TestCase):\n def test_one(self): pass\n" if mode == "unexpected" else ""))
|
||||
if mode == "missing":
|
||||
(root / "harness/tests/review/test_review.py").unlink()
|
||||
elif mode == "marker":
|
||||
(root / "harness/tests/spec/__init__.py").unlink()
|
||||
elif mode == "duplicate":
|
||||
(root / "harness/tests/common/test_common.py").write_text(
|
||||
"import unittest\nclass Example(unittest.TestCase):\n def test_one(self): pass\nclass Example(unittest.TestCase):\n def test_two(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("canonical", result.stderr.lower())
|
||||
|
||||
def test_loader_omission_addition_zero_file_and_unsupported_function_fail(self):
|
||||
cases = {
|
||||
"omission": lambda root: (root / "harness/tests/common/test_common.py").write_text("import unittest\ndef load_tests(loader, tests, pattern): return unittest.TestSuite()\nclass X(unittest.TestCase):\n def test_one(self): pass\n"),
|
||||
"zero": lambda root: (root / "harness/tests/review/test_review.py").write_text("import unittest\n"),
|
||||
"free": lambda root: (root / "harness/tests/impl/test_impl.py").write_text("def test_free(): pass\n"),
|
||||
}
|
||||
for name, mutate in cases.items():
|
||||
with self.subTest(name=name):
|
||||
root = self.make_suite(); mutate(root)
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/spec/test_spec.py").write_text("import unittest\nclass X(unittest.TestCase):\n def test_one(self): pass\n def test_two(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
|
||||
def test_import_failure_source_drift_nested_omission_and_runtime_failure(self):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/impl/test_impl.py").write_text("raise RuntimeError('broken import')\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("canonical", result.stderr.lower())
|
||||
root = self.make_suite(nested=True)
|
||||
(root / "harness/tests/common/nested/__init__.py").unlink()
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/review/test_review.py").write_text(
|
||||
"import unittest\nfrom pathlib import Path\np = Path(__file__)\np.write_text(p.read_text() + '# drift\\n')\n"
|
||||
"class X(unittest.TestCase):\n def test_one(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_common.py").write_text(
|
||||
"import unittest\nclass X(unittest.TestCase):\n def test_one(self): self.fail('boom')\n")
|
||||
result = self.execute(root)
|
||||
self.assertNotEqual(result.returncode, 0)
|
||||
self.assertIn("FAILED", result.stderr + result.stdout)
|
||||
|
||||
def test_package_manifest_drift_and_extra_zero_file_fail_closed(self):
|
||||
cases = (
|
||||
("initializer", "from pathlib import Path\np = Path(__file__).parent / '__init__.py'\np.write_text(p.read_text() + '# drift\\n')\n"),
|
||||
("add-source", "from pathlib import Path\nPath(__file__).with_name('test_added.py').write_text('import unittest\\nclass Added(unittest.TestCase):\\n def test_added(self): pass\\n')\n"),
|
||||
("remove-source", "from pathlib import Path\n(Path(__file__).parents[1] / 'review' / 'test_review.py').unlink()\n"),
|
||||
("extra-zero", "from pathlib import Path\nPath(__file__).with_name('test_empty.py').write_text('import unittest\\n')\n"),
|
||||
)
|
||||
for name, mutation in cases:
|
||||
with self.subTest(name=name):
|
||||
root = self.make_suite()
|
||||
target = root / "harness/tests/common/test_common.py"
|
||||
target.write_text("import unittest\n" + mutation +
|
||||
"class X(unittest.TestCase):\n def test_one(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertIn("source/package manifest drift", result.stderr.lower())
|
||||
|
||||
def test_symlinked_source_and_package_paths_fail_before_import(self):
|
||||
for kind in ("source", "marker", "role", "nested"):
|
||||
with self.subTest(kind=kind):
|
||||
root = self.make_suite()
|
||||
marker = root / "escaped-import"
|
||||
outside = root / "outside"
|
||||
outside.mkdir()
|
||||
payload = ("from pathlib import Path\n"
|
||||
+ "Path(" + repr(str(marker)) + ").write_text('imported')\n"
|
||||
+ "import unittest\nclass Escape(unittest.TestCase):\n def test_escape(self): pass\n")
|
||||
if kind == "source":
|
||||
external = outside / "escape.py"
|
||||
external.write_text(payload)
|
||||
(root / "harness/tests/common/test_escape.py").symlink_to(external)
|
||||
elif kind == "marker":
|
||||
external = outside / "init.py"
|
||||
external.write_text(payload)
|
||||
path = root / "harness/tests/common/__init__.py"
|
||||
path.unlink()
|
||||
path.symlink_to(external)
|
||||
else:
|
||||
(outside / "__init__.py").write_text(payload)
|
||||
(outside / "test_escape.py").write_text(payload)
|
||||
if kind == "role":
|
||||
path = root / "harness/tests/review"
|
||||
(path / "__init__.py").unlink()
|
||||
(path / "test_review.py").unlink()
|
||||
path.rmdir()
|
||||
else:
|
||||
path = root / "harness/tests/common/nested"
|
||||
path.symlink_to(outside, target_is_directory=True)
|
||||
result = self.execute(root)
|
||||
self.assertEqual(result.returncode, 2, result.stderr)
|
||||
self.assertIn("symlink", result.stderr.lower())
|
||||
self.assertFalse(marker.exists(), "escaped code was imported before rejection")
|
||||
|
||||
def test_static_extra_zero_file_fails_with_valid_sibling(self):
|
||||
root = self.make_suite()
|
||||
(root / "harness/tests/common/test_empty.py").write_text("# empty candidate\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertIn("zero inventoried tests", result.stderr)
|
||||
|
||||
def test_removed_role_marker_is_manifest_failure(self):
|
||||
root = self.make_suite()
|
||||
source = root / "harness/tests/common/test_common.py"
|
||||
source.write_text("import unittest\nfrom pathlib import Path\n"
|
||||
"(Path(__file__).parents[1] / 'review' / '__init__.py').unlink()\n"
|
||||
"class X(unittest.TestCase):\n def test_one(self): pass\n")
|
||||
result = self.execute(root, "--collect-only")
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertIn("missing package marker", result.stderr)
|
||||
|
||||
def test_normal_synthetic_success_and_assertion_failure_exit_statuses(self):
|
||||
root = self.make_suite()
|
||||
success = self.execute(root)
|
||||
self.assertEqual(success.returncode, 0, success.stderr)
|
||||
(root / "harness/tests/common/test_common.py").write_text(
|
||||
"import unittest\nclass X(unittest.TestCase):\n def test_one(self): self.fail('named assertion')\n")
|
||||
failure = self.execute(root)
|
||||
self.assertEqual(failure.returncode, 1)
|
||||
self.assertIn("named assertion", failure.stderr + failure.stdout)
|
||||
(root / "harness/tests/common/test_common.py").write_text(
|
||||
"import unittest\nclass X(unittest.TestCase):\n def test_one(self): raise RuntimeError('named error')\n")
|
||||
error = self.execute(root)
|
||||
self.assertEqual(error.returncode, 1)
|
||||
self.assertIn("named error", error.stderr + error.stdout)
|
||||
@@ -15,7 +15,7 @@ def inventory_digest(ids, statuses):
|
||||
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
||||
|
||||
|
||||
def test_inventory(ids=('case_a',), current=None):
|
||||
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}
|
||||
@@ -157,84 +157,84 @@ class WorkflowGuardTests(unittest.TestCase):
|
||||
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')
|
||||
self.assertEqual(guard.check_tests(make_test_inventory(current=['case_a', 'case_b']))['status'], 'pass')
|
||||
|
||||
def test_collected_inventory_requires_provenance(self):
|
||||
value = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory(current=['case_b'])
|
||||
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 = test_inventory(current=['case_b'])
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
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 = test_inventory()
|
||||
value = make_test_inventory()
|
||||
value['findings'] = [finding_record()]
|
||||
self.assertEqual(guard.check_tests(value)['status'], 'pass')
|
||||
|
||||
@@ -342,31 +342,31 @@ class WorkflowGuardTests(unittest.TestCase):
|
||||
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 = 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 = test_inventory(current=['case_b'])
|
||||
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 = test_inventory(ids=['case_a', 'case_b'], current=['case_b'])
|
||||
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 = test_inventory(current=['case_b'])
|
||||
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 = test_inventory()
|
||||
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'],
|
||||
@@ -374,7 +374,7 @@ class WorkflowGuardTests(unittest.TestCase):
|
||||
self.assertEqual(guard.check_tests(value)['status'], 'pass')
|
||||
|
||||
def test_unhashable_authorization_id_fails_closed_api(self):
|
||||
value = test_inventory()
|
||||
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)
|
||||
@@ -408,10 +408,10 @@ class WorkflowGuardTests(unittest.TestCase):
|
||||
|
||||
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')
|
||||
self.assertEqual(guard.check_tests(make_test_inventory())['status'], 'pass')
|
||||
|
||||
def test_unknown_inventory_fields_are_rejected(self):
|
||||
value = test_inventory()
|
||||
value = make_test_inventory()
|
||||
value['pretend_verified'] = True
|
||||
self.assert_block(guard.check_tests, value)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user