94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
"""Settings commits retain valid files on failure and concurrent writes."""
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import ctypes
|
|
import errno
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
from python_app.file_io import atomic_write_json, rename_no_replace
|
|
from python_app.orchestration.gui_session_state import GuiSessionStateStore
|
|
|
|
|
|
class AtomicJsonTest(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.path = Path(directory.name) / "settings.json"
|
|
atomic_write_json(self.path, {"old": True})
|
|
|
|
def test_failed_sync_or_commit_preserves_previous_file(self) -> None:
|
|
for operation in ('fsync', 'replace'):
|
|
with self.subTest(operation=operation), \
|
|
mock.patch(f'python_app.file_io.os.{operation}', side_effect=OSError('disk failure')), \
|
|
self.assertRaises(OSError):
|
|
atomic_write_json(self.path, {'new': True}, durable=True)
|
|
self.assertEqual(json.loads(self.path.read_text()), {'old': True})
|
|
self.assertEqual(list(self.path.parent.glob('*.tmp')), [])
|
|
|
|
def test_nonfinite_json_cannot_overwrite_settings(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
atomic_write_json(self.path, {'bad': float('nan')})
|
|
self.assertEqual(json.loads(self.path.read_text()), {'old': True})
|
|
|
|
def test_replacement_preserves_existing_permissions(self) -> None:
|
|
self.path.chmod(0o664)
|
|
atomic_write_json(self.path, {"new": True}, durable=True)
|
|
self.assertEqual(stat.S_IMODE(self.path.stat().st_mode), 0o664)
|
|
|
|
def test_new_file_uses_process_umask(self) -> None:
|
|
path = self.path.with_name("new.json")
|
|
previous = os.umask(0o027)
|
|
try:
|
|
atomic_write_json(path, {"new": True})
|
|
finally:
|
|
os.umask(previous)
|
|
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o640)
|
|
|
|
def test_publish_cannot_replace_existing_file_or_empty_directory(self) -> None:
|
|
source = self.path.with_name('source')
|
|
source.write_text('new', encoding='utf-8')
|
|
before = self.path.read_bytes()
|
|
with self.assertRaises(FileExistsError):
|
|
rename_no_replace(source, self.path)
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
self.assertEqual(source.read_text(), 'new')
|
|
source.unlink()
|
|
source.mkdir()
|
|
destination = self.path.with_name('destination')
|
|
destination.mkdir()
|
|
with self.assertRaises(FileExistsError):
|
|
rename_no_replace(source, destination)
|
|
self.assertTrue(source.is_dir())
|
|
self.assertTrue(destination.is_dir())
|
|
|
|
def test_publish_propagates_filesystem_permission_error_without_touching_source(self) -> None:
|
|
destination = self.path.with_name('denied.json')
|
|
|
|
def deny(*_args):
|
|
ctypes.set_errno(errno.EACCES)
|
|
return -1
|
|
|
|
before = self.path.read_bytes()
|
|
with mock.patch('python_app.file_io._renameat2', side_effect=deny):
|
|
with self.assertRaises(PermissionError):
|
|
rename_no_replace(self.path, destination)
|
|
self.assertEqual(self.path.read_bytes(), before)
|
|
self.assertFalse(destination.exists())
|
|
|
|
def test_concurrent_writers_publish_complete_json_and_leave_no_temp_files(self) -> None:
|
|
payloads = [{'index': i, 'values': [i] * 1000} for i in range(32)]
|
|
with ThreadPoolExecutor(max_workers=8) as executor:
|
|
list(executor.map(lambda payload: atomic_write_json(self.path, payload), payloads))
|
|
self.assertIn(json.loads(self.path.read_text()), payloads)
|
|
self.assertEqual(list(self.path.parent.glob('*.tmp')), [])
|
|
|
|
def test_invalid_utf8_session_cache_uses_defaults(self) -> None:
|
|
self.path.write_bytes(b'\xff')
|
|
self.assertEqual(GuiSessionStateStore(self.path).load().last_profile_path, '')
|