103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
from app.storage import DATA_FORMAT_VERSION, StoredTask
|
|
from app.storage.repository import JsonFileTaskRepository
|
|
|
|
|
|
def make_task(title: str = "Task") -> StoredTask:
|
|
return StoredTask(
|
|
id=uuid4(),
|
|
title=title,
|
|
created_at=datetime.now(UTC),
|
|
started_at=None,
|
|
done_at=None,
|
|
)
|
|
|
|
|
|
def test_corrupted_json_triggers_backup_and_reset(tmp_path):
|
|
file_path = tmp_path / "tasks.json"
|
|
file_path.write_text('{"version": 1, "tasks": [', encoding="utf-8")
|
|
|
|
repo = JsonFileTaskRepository(file_path)
|
|
|
|
assert repo.list_tasks() == []
|
|
|
|
backups = list(tmp_path.glob("tasks.json.corrupted*"))
|
|
assert len(backups) == 1
|
|
assert backups[0].read_text(encoding="utf-8") == '{"version": 1, "tasks": ['
|
|
|
|
restored = json.loads(file_path.read_text(encoding="utf-8"))
|
|
assert restored["version"] == DATA_FORMAT_VERSION
|
|
assert restored["tasks"] == []
|
|
|
|
|
|
def test_invalid_schema_triggers_backup_and_reset(tmp_path):
|
|
file_path = tmp_path / "tasks.json"
|
|
file_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"version": 1,
|
|
"tasks": [
|
|
{
|
|
"id": str(uuid4()),
|
|
"created_at": datetime.now(UTC).isoformat(),
|
|
},
|
|
],
|
|
},
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
repo = JsonFileTaskRepository(file_path)
|
|
|
|
assert repo.list_tasks() == []
|
|
backups = list(tmp_path.glob("tasks.json.corrupted*"))
|
|
assert len(backups) == 1
|
|
|
|
|
|
def test_repository_remains_writable_after_recovery(tmp_path):
|
|
file_path = tmp_path / "tasks.json"
|
|
file_path.write_text("not valid json", encoding="utf-8")
|
|
repo = JsonFileTaskRepository(file_path)
|
|
|
|
task = make_task("Recovered")
|
|
repo.create_task(task)
|
|
|
|
assert repo.list_tasks() == [task]
|
|
|
|
|
|
def test_unsupported_version_creates_backup_and_resets_state(tmp_path):
|
|
file_path = tmp_path / "tasks.json"
|
|
file_path.write_text(
|
|
json.dumps({"version": 999, "tasks": []}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
repo = JsonFileTaskRepository(file_path)
|
|
|
|
assert repo.list_tasks() == []
|
|
backups = list(tmp_path.glob("tasks.json.corrupted*"))
|
|
assert len(backups) == 1
|
|
|
|
restored = json.loads(file_path.read_text(encoding="utf-8"))
|
|
assert restored["version"] == DATA_FORMAT_VERSION
|
|
assert restored["tasks"] == []
|
|
|
|
|
|
def test_restart_after_corrupted_persisted_state_does_not_crash(tmp_path):
|
|
file_path = tmp_path / "tasks.json"
|
|
repo_1 = JsonFileTaskRepository(file_path)
|
|
repo_1.create_task(make_task("Initial"))
|
|
|
|
file_path.write_text("{broken", encoding="utf-8")
|
|
|
|
repo_2 = JsonFileTaskRepository(file_path)
|
|
|
|
assert repo_2.list_tasks() == []
|
|
assert len(list(tmp_path.glob("tasks.json.corrupted*"))) == 1
|
|
|