From e091d95f0ff4c9a245948dc9e3f84be13bcf662c Mon Sep 17 00:00:00 2001 From: northline-lab Date: Sat, 30 May 2026 22:39:24 +0000 Subject: [PATCH] test: add unit tests for contribarena.engine.persistence atomic write helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add focused unit tests for `contribarena.engine.persistence` atomic write helpers (`atomic_write_text`, `atomic_write_json`). These helpers are used across `engine/middleware/governance.py`, `engine/seasons.py`, `engine/judge_refresh.py`, and `engine/controller.py` to write durable artifacts, but had no dedicated unit test coverage. ## Coverage `tests/unit/test_persistence.py` adds 2 test classes and 16 tests: - **AtomicWriteTextTest** (6 tests) — writes text content, creates missing parent directories, overwrites existing files, preserves non-ASCII characters via UTF-8, does not leave a `.tmp` sibling after success, writes empty strings. - **AtomicWriteJsonTest** (10 tests) — pretty-printed JSON with trailing newline, two-space indent, `sort_keys=True` ordering, default unsorted ordering, default `ensure_ascii=True` escaping, `ensure_ascii=False` preservation, missing parent directories, overwrite behavior, list payload support, no `.tmp` sibling after success. No production code is modified. ## Verification - `pytest -q tests/unit/test_persistence.py` → 16 passed - `ruff check tests/unit/test_persistence.py` → All checks passed ## Risk Low — test-only addition. No production code is modified. --- *This PR was created autonomously by an AI agent participating in ContribArena's evaluation framework.* --- tests/unit/test_persistence.py | 171 +++++++++++++++++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 tests/unit/test_persistence.py diff --git a/tests/unit/test_persistence.py b/tests/unit/test_persistence.py new file mode 100644 index 0000000..8478b8c --- /dev/null +++ b/tests/unit/test_persistence.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path + +from contribarena.engine.persistence import atomic_write_json, atomic_write_text + + +class AtomicWriteTextTest(unittest.TestCase): + def test_writes_text_to_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.txt" + + atomic_write_text(path, "hello") + + self.assertEqual("hello", path.read_text(encoding="utf-8")) + + def test_creates_missing_parent_directories(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "nested" / "deep" / "out.txt" + + atomic_write_text(path, "data") + + self.assertTrue(path.parent.is_dir()) + self.assertEqual("data", path.read_text(encoding="utf-8")) + + def test_overwrites_existing_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.txt" + path.write_text("old contents", encoding="utf-8") + + atomic_write_text(path, "new contents") + + self.assertEqual("new contents", path.read_text(encoding="utf-8")) + + def test_preserves_non_ascii_characters_with_utf8(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.txt" + + atomic_write_text(path, "caf\u00e9 \u2603 \U0001f600") + + self.assertEqual( + "caf\u00e9 \u2603 \U0001f600", + path.read_text(encoding="utf-8"), + ) + + def test_does_not_leave_temp_file_after_success(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + parent = Path(tmp) + path = parent / "out.txt" + + atomic_write_text(path, "data") + + siblings = [p.name for p in parent.iterdir()] + self.assertEqual(["out.txt"], siblings) + + def test_writes_empty_string(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.txt" + + atomic_write_text(path, "") + + self.assertTrue(path.exists()) + self.assertEqual("", path.read_text(encoding="utf-8")) + + +class AtomicWriteJsonTest(unittest.TestCase): + def test_writes_pretty_printed_json_with_trailing_newline(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + payload = {"b": 1, "a": 2} + + atomic_write_json(path, payload) + + text = path.read_text(encoding="utf-8") + self.assertTrue(text.endswith("\n")) + self.assertEqual(payload, json.loads(text)) + + def test_indent_is_two_spaces(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, {"key": "value"}) + + text = path.read_text(encoding="utf-8") + self.assertIn('\n "key": "value"', text) + + def test_sort_keys_true_sorts_keys_alphabetically(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, {"b": 1, "a": 2, "c": 3}, sort_keys=True) + + text = path.read_text(encoding="utf-8") + a_pos = text.index('"a"') + b_pos = text.index('"b"') + c_pos = text.index('"c"') + self.assertLess(a_pos, b_pos) + self.assertLess(b_pos, c_pos) + + def test_sort_keys_false_by_default(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, {"b": 1, "a": 2}) + + text = path.read_text(encoding="utf-8") + self.assertLess(text.index('"b"'), text.index('"a"')) + + def test_ensure_ascii_true_escapes_non_ascii_by_default(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, {"label": "caf\u00e9"}) + + text = path.read_text(encoding="utf-8") + self.assertIn("caf\\u00e9", text) + self.assertNotIn("caf\u00e9", text) + self.assertEqual({"label": "caf\u00e9"}, json.loads(text)) + + def test_ensure_ascii_false_preserves_non_ascii(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, {"label": "caf\u00e9"}, ensure_ascii=False) + + text = path.read_text(encoding="utf-8") + self.assertIn("caf\u00e9", text) + self.assertEqual({"label": "caf\u00e9"}, json.loads(text)) + + def test_creates_missing_parent_directories(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "nested" / "deep" / "out.json" + + atomic_write_json(path, {"ok": True}) + + self.assertTrue(path.parent.is_dir()) + self.assertEqual({"ok": True}, json.loads(path.read_text(encoding="utf-8"))) + + def test_overwrites_existing_file(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + path.write_text("stale", encoding="utf-8") + + atomic_write_json(path, {"fresh": True}) + + self.assertEqual({"fresh": True}, json.loads(path.read_text(encoding="utf-8"))) + + def test_writes_list_payload(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "out.json" + + atomic_write_json(path, [1, 2, 3]) + + self.assertEqual([1, 2, 3], json.loads(path.read_text(encoding="utf-8"))) + + def test_does_not_leave_temp_file_after_success(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + parent = Path(tmp) + path = parent / "out.json" + + atomic_write_json(path, {"ok": True}) + + siblings = [p.name for p in parent.iterdir()] + self.assertEqual(["out.json"], siblings) + + +if __name__ == "__main__": + unittest.main()