diff --git a/tests/unit/test_complete_npm_lock_integrity.py b/tests/unit/test_complete_npm_lock_integrity.py index 1279b6f9..b5c061a9 100644 --- a/tests/unit/test_complete_npm_lock_integrity.py +++ b/tests/unit/test_complete_npm_lock_integrity.py @@ -3,11 +3,16 @@ from __future__ import annotations import json +import io import tempfile import unittest +from contextlib import redirect_stdout from pathlib import Path +from unittest.mock import patch -from scripts.complete_npm_lock_integrity import complete_lock +from scripts import complete_npm_lock_integrity as integrity_tool + +complete_lock = integrity_tool.complete_lock class TestCompleteLock(unittest.TestCase): @@ -61,6 +66,33 @@ class TestCompleteLock(unittest.TestCase): with self.assertRaisesRegex(ValueError, "conflicting integrity"): complete_lock(path) + def test_ignores_local_entries_and_leaves_complete_lock_unchanged(self): + with tempfile.TemporaryDirectory() as directory: + path = self._lock(directory, { + "": {"name": "root"}, + "node_modules/local": {"resolved": "file:../local"}, + "node_modules/complete": { + "resolved": "https://registry.example/complete.tgz", + "integrity": "sha512-complete", + }, + }) + before = path.read_text() + self.assertEqual(0, complete_lock(path)) + self.assertEqual(before, path.read_text()) + + def test_main_processes_every_requested_lock(self): + with tempfile.TemporaryDirectory() as directory: + first = self._lock(directory, {}) + second = Path(directory) / "second-lock.json" + second.write_text(first.read_text()) + stdout = io.StringIO() + with patch( + "sys.argv", + ["complete_npm_lock_integrity.py", str(first), str(second)], + ), redirect_stdout(stdout): + self.assertEqual(0, integrity_tool.main()) + self.assertEqual(2, stdout.getvalue().count("completed 0")) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_docker_util_image.py b/tests/unit/test_docker_util_image.py index 5e3e7d1d..bfbdaaed 100644 --- a/tests/unit/test_docker_util_image.py +++ b/tests/unit/test_docker_util_image.py @@ -223,6 +223,24 @@ class TestPinnedLocalImageRef(unittest.TestCase): with self.assertRaises(SystemExit): docker_mod.pinned_local_image_ref("base:latest") + def test_rejects_content_tag_that_resolves_to_another_image(self): + image = "sha256:" + "e" * 64 + with patch.object( + docker_mod, + "image_id", + side_effect=[image, "sha256:" + "f" * 64], + ), patch.object( + docker_mod, + "run_docker", + return_value=_ok(), + ), patch.object( + docker_mod, + "die", + side_effect=SystemExit("die"), + ): + with self.assertRaises(SystemExit): + docker_mod.pinned_local_image_ref("base:latest") + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_image_input_policy.py b/tests/unit/test_image_input_policy.py index feab748e..2402cd98 100644 --- a/tests/unit/test_image_input_policy.py +++ b/tests/unit/test_image_input_policy.py @@ -3,9 +3,12 @@ from __future__ import annotations import json +import io import tempfile import unittest +from contextlib import redirect_stderr, redirect_stdout from pathlib import Path +from unittest.mock import patch from scripts import check_image_inputs as policy @@ -55,6 +58,35 @@ class TestDockerfilePolicy(unittest.TestCase): problems = policy.check_dockerfile(Path("Dockerfile"), text) self.assertTrue(any("fixed Debian snapshot" in item for item in problems)) + def test_rejects_missing_dynamic_and_malformed_bases(self): + cases = ( + ("RUN true\n", "missing FROM"), + ("FROM ${BASE_IMAGE}\n", "dynamic base"), + ( + "ARG ORCHESTRATOR_BASE_IMAGE=base:latest\n" + "FROM ${ORCHESTRATOR_BASE_IMAGE}\n", + "mutable default", + ), + ("FROM node:22@sha256:abc\n", "64 lowercase hex"), + ) + for text, expected in cases: + with self.subTest(expected=expected): + problems = policy.check_dockerfile(Path("Dockerfile"), text) + self.assertTrue(any(expected in item for item in problems)) + + def test_requires_gateway_lock_and_verified_codex_markers(self): + base = "FROM node:22.23.1@sha256:" + "a" * 64 + "\n" + gateway = policy.check_dockerfile(Path("Dockerfile.gateway"), base) + self.assertTrue(any("require hashes" in item for item in gateway)) + self.assertTrue(any("lock is not consumed" in item for item in gateway)) + + codex = policy.check_dockerfile( + Path("bot_bottle/contrib/codex/Dockerfile"), + base, + ) + self.assertEqual(3, len(codex)) + self.assertTrue(all("verified pinned Codex install" in item for item in codex)) + class TestNpmLockPolicy(unittest.TestCase): def _write(self, root: Path, *, version: str, integrity: str = "") -> Path: @@ -97,11 +129,78 @@ class TestNpmLockPolicy(unittest.TestCase): self.assertTrue(any("not an exact version" in item for item in problems)) self.assertTrue(any("lacks sha512 integrity" in item for item in problems)) + def test_rejects_unreadable_stale_or_old_lock(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + missing = policy.check_npm_manifest(root, Path("missing/package.json")) + self.assertTrue(any("cannot read" in item for item in missing)) + + path = self._write(root, version="1.2.3", integrity="sha512-abc") + lock_path = root / path.with_name("package-lock.json") + lock = json.loads(lock_path.read_text()) + lock["lockfileVersion"] = 2 + lock["packages"][""]["dependencies"] = {} + lock_path.write_text(json.dumps(lock)) + problems = policy.check_npm_manifest(root, path) + self.assertTrue(any("root dependencies are stale" in item for item in problems)) + self.assertTrue(any("lockfileVersion 3" in item for item in problems)) + + +class TestPythonLockPolicy(unittest.TestCase): + def test_rejects_missing_or_empty_lock(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + missing = policy.check_python_lock(root) + self.assertTrue(any("cannot read lock" in item for item in missing)) + + (root / "requirements.gateway.lock").write_text("# empty\n") + problems = policy.check_python_lock(root) + self.assertTrue(any("contains no exact requirements" in item for item in problems)) + self.assertTrue(any("cannot read input" in item for item in problems)) + + def test_rejects_unhashed_and_stale_direct_requirements(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "requirements.gateway.lock").write_text( + "example== \\\n" + " # deliberately malformed and unhashed\n" + ) + (root / "requirements.gateway.in").write_text( + "example>=1\n" + "missing==2.0.0\n" + ) + problems = policy.check_python_lock(root) + self.assertTrue(any("example is not exact" in item for item in problems)) + self.assertTrue(any("example lacks a sha256 hash" in item for item in problems)) + self.assertTrue(any("direct dependency is not exact" in item for item in problems)) + self.assertTrue(any("missing direct dependency" in item for item in problems)) + class TestRepositoryPolicy(unittest.TestCase): def test_repository_image_inputs_pass(self): self.assertEqual([], policy.check_repo()) + def test_missing_repository_inputs_are_reported(self): + with tempfile.TemporaryDirectory() as directory, patch.object( + policy, + "DOCKERFILES", + (Path("missing.Dockerfile"),), + ), patch.object(policy, "NPM_MANIFESTS", ()): + problems = policy.check_repo(Path(directory)) + self.assertTrue(any("cannot read Dockerfile" in item for item in problems)) + + def test_main_reports_success_and_failure(self): + with patch.object(policy, "check_repo", return_value=[]), redirect_stdout(io.StringIO()): + self.assertEqual(0, policy.main()) + stderr = io.StringIO() + with patch.object( + policy, + "check_repo", + return_value=["bad base"], + ), redirect_stderr(stderr): + self.assertEqual(1, policy.main()) + self.assertIn("bad base", stderr.getvalue()) + if __name__ == "__main__": unittest.main()