| """Persistence and security contract tests.""" |
| import json |
| import os |
| import tempfile |
|
|
| import pytest |
|
|
| BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) |
| if BACKEND_DIR not in __import__("sys").path: |
| __import__("sys").path.insert(0, BACKEND_DIR) |
|
|
| import persistence as persist |
|
|
|
|
| def test_atomic_save_roundtrip(): |
| with tempfile.TemporaryDirectory() as tmp: |
| path = os.path.join(tmp, "data.json") |
| payload = {"items": [1, 2, 3], "name": "test"} |
| persist.save_json(path, payload) |
| loaded = persist.load_json(path, {}) |
| assert loaded == payload |
|
|
|
|
| def test_load_missing_returns_default(): |
| with tempfile.TemporaryDirectory() as tmp: |
| path = os.path.join(tmp, "missing.json") |
| assert persist.load_json(path, []) == [] |
|
|
|
|
| def test_concurrent_writes_do_not_corrupt(): |
| with tempfile.TemporaryDirectory() as tmp: |
| path = os.path.join(tmp, "list.json") |
| for i in range(20): |
| data = persist.load_json(path, []) |
| data.append(i) |
| persist.save_json(path, data) |
| loaded = persist.load_json(path, []) |
| assert len(loaded) == 20 |
| assert loaded[0] == 0 |
| assert loaded[-1] == 19 |
|
|
