from __future__ import annotations import json import os from pathlib import Path import stat import sys import tempfile import unittest from unittest import mock META_ROOT = Path(__file__).resolve().parents[1] ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments" RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release" for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT): if str(tools_root) not in sys.path: sys.path.insert(0, str(tools_root)) from govoplan_assessment.atomic_io import ( # noqa: E402 AtomicJsonWriteError, atomic_write_json, ) class CapabilityFitAtomicIOTests(unittest.TestCase): def test_writes_private_json_with_expected_content(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: target = Path(temporary_directory) / "evidence.json" payload = {"z": [1, 2], "message": "Grüße"} created_in: list[Path] = [] original_mkstemp = tempfile.mkstemp def observed_mkstemp(*args, **kwargs): created_in.append(Path(kwargs["dir"])) return original_mkstemp(*args, **kwargs) with ( mock.patch( "govoplan_assessment.atomic_io.tempfile.mkstemp", side_effect=observed_mkstemp, ), mock.patch( "govoplan_assessment.atomic_io.os.fsync", wraps=os.fsync, ) as fsync, ): atomic_write_json(target, payload, max_bytes=1024) self.assertEqual(payload, json.loads(target.read_text(encoding="utf-8"))) self.assertTrue(target.read_bytes().endswith(b"\n")) self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) self.assertEqual([target.parent], created_in) self.assertEqual(2, fsync.call_count) def test_replaces_existing_regular_file_and_secures_mode(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: target = Path(temporary_directory) / "report.json" target.write_text('{"old": true}\n', encoding="utf-8") target.chmod(0o644) old_handle = target.open("rb") self.addCleanup(old_handle.close) atomic_write_json(target, {"new": True}, max_bytes=1024) self.assertEqual(b'{"old": true}\n', old_handle.read()) self.assertEqual({"new": True}, json.loads(target.read_text("utf-8"))) self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) def test_size_bound_leaves_existing_target_unchanged(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) target = directory / "evidence.json" original = b'{"original": true}\n' target.write_bytes(original) with self.assertRaisesRegex(AtomicJsonWriteError, "size limit"): atomic_write_json(target, {"large": "x" * 100}, max_bytes=32) self.assertEqual(original, target.read_bytes()) self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) @unittest.skipUnless(hasattr(os, "symlink"), "symbolic links are unavailable") def test_rejects_symlink_without_modifying_link_or_referent(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) referent = directory / "outside.json" referent.write_bytes(b'{"keep": true}\n') target = directory / "evidence.json" target.symlink_to(referent.name) with self.assertRaisesRegex(AtomicJsonWriteError, "symbolic link"): atomic_write_json(target, {"replace": True}, max_bytes=1024) self.assertTrue(target.is_symlink()) self.assertEqual(b'{"keep": true}\n', referent.read_bytes()) self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) @unittest.skipUnless(hasattr(os, "symlink"), "symbolic links are unavailable") def test_rejects_symlinked_parent_component(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) real_parent = directory / "real-parent" real_parent.mkdir() linked_parent = directory / "linked-parent" linked_parent.symlink_to(real_parent, target_is_directory=True) target = linked_parent / "new" / "evidence.json" with self.assertRaisesRegex(AtomicJsonWriteError, "parent path"): atomic_write_json(target, {"unsafe": True}, max_bytes=1024) self.assertFalse((real_parent / "new").exists()) def test_requires_existing_parent_without_creating_directories(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) missing_parent = directory / "missing" / "nested" target = missing_parent / "evidence.json" with self.assertRaisesRegex(AtomicJsonWriteError, "already exist"): atomic_write_json(target, {"safe": True}, max_bytes=1024) self.assertFalse(missing_parent.exists()) def test_rejects_post_replace_mode_or_identity_change(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) target = directory / "evidence.json" original_replace = os.replace def insecure_replace(source, destination): original_replace(source, destination) Path(destination).chmod(0o644) with mock.patch( "govoplan_assessment.atomic_io.os.replace", side_effect=insecure_replace, ): with self.assertRaisesRegex( AtomicJsonWriteError, "permissions changed and were restored", ): atomic_write_json(target, {"safe": True}, max_bytes=1024) self.assertEqual(0o600, stat.S_IMODE(target.stat().st_mode)) self.assertEqual({"safe": True}, json.loads(target.read_text("utf-8"))) def test_does_not_unlink_unknown_post_replace_inode(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) target = directory / "evidence.json" original_replace = os.replace def replaced_again(source, destination): original_replace(source, destination) Path(destination).unlink() Path(destination).write_bytes(b"unknown replacement\n") with mock.patch( "govoplan_assessment.atomic_io.os.replace", side_effect=replaced_again, ): with self.assertRaisesRegex( AtomicJsonWriteError, "did not preserve", ): atomic_write_json(target, {"secret": True}, max_bytes=1024) self.assertEqual(b"unknown replacement\n", target.read_bytes()) def test_replace_failure_removes_private_temporary_file(self) -> None: with tempfile.TemporaryDirectory() as temporary_directory: directory = Path(temporary_directory) target = directory / "evidence.json" with mock.patch( "govoplan_assessment.atomic_io.os.replace", side_effect=OSError("simulated replacement failure"), ): with self.assertRaisesRegex( AtomicJsonWriteError, "could not be written atomically" ): atomic_write_json(target, {"safe": True}, max_bytes=1024) self.assertFalse(target.exists()) self.assertEqual([], list(directory.glob(".govoplan-json-*.tmp"))) if __name__ == "__main__": unittest.main()