191 lines
7.9 KiB
Python
191 lines
7.9 KiB
Python
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()
|