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)