278 lines
9.1 KiB
Python
278 lines
9.1 KiB
Python
import json
|
|
from datetime import date, datetime
|
|
from io import BytesIO
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.auth.services import create_user
|
|
from app.tasks.models import Subtask, Task
|
|
from app.tasks.services import (
|
|
calculate_forgiving_streak,
|
|
clear_completed_tasks,
|
|
complete_task,
|
|
create_task,
|
|
get_task,
|
|
list_tasks,
|
|
undo_clear_completed,
|
|
)
|
|
from app.tasks.transfer import (
|
|
TransferValidationError,
|
|
export_backup,
|
|
import_backup,
|
|
parse_markdown,
|
|
)
|
|
|
|
|
|
def login(client, email: str) -> None:
|
|
client.post(
|
|
"/auth/login",
|
|
data={"email": email, "password": "secure password"},
|
|
)
|
|
|
|
|
|
def make_task(user_id: int, title: str, *, status: str = "not_started") -> Task:
|
|
return create_task(
|
|
user_id,
|
|
title=title,
|
|
description=None,
|
|
status=status,
|
|
priority="normal",
|
|
due_date=None,
|
|
)
|
|
|
|
|
|
def test_markdown_parser_supports_lists_checkboxes_and_plain_text():
|
|
content = b"""# Weekend
|
|
- [ ] Buy groceries
|
|
- [x] Book appointment
|
|
1. Water the plants
|
|
```
|
|
- Ignore example code
|
|
```
|
|
Call a friend
|
|
"""
|
|
|
|
tasks = parse_markdown(content)
|
|
|
|
assert [task.title for task in tasks] == [
|
|
"Buy groceries",
|
|
"Book appointment",
|
|
"Water the plants",
|
|
"Call a friend",
|
|
]
|
|
assert [task.status for task in tasks] == [
|
|
"not_started",
|
|
"done",
|
|
"not_started",
|
|
"not_started",
|
|
]
|
|
|
|
|
|
def test_markdown_parser_rejects_invalid_or_oversized_input():
|
|
with pytest.raises(TransferValidationError, match="UTF-8"):
|
|
parse_markdown(b"\xff")
|
|
with pytest.raises(TransferValidationError, match="200-character"):
|
|
parse_markdown(("- " + "a" * 201).encode())
|
|
with pytest.raises(TransferValidationError, match="at most 200"):
|
|
parse_markdown("\n".join(f"- Task {i}" for i in range(201)).encode())
|
|
|
|
|
|
def test_markdown_upload_imports_atomically(client, db):
|
|
user = create_user("text_user", "text@example.com", "secure password")
|
|
login(client, user.email)
|
|
|
|
response = client.post(
|
|
"/tasks/import/markdown",
|
|
data={
|
|
"markdown_file": (BytesIO(b"- First\n- [x] Second"), "tasks.md"),
|
|
"confirm": "y",
|
|
},
|
|
content_type="multipart/form-data",
|
|
follow_redirects=True,
|
|
)
|
|
tasks = list(db.session.scalars(select(Task).order_by(Task.id)))
|
|
|
|
assert b"Imported 2 tasks" in response.data
|
|
assert [task.user_id for task in tasks] == [user.id, user.id]
|
|
assert [task.status for task in tasks] == ["not_started", "done"]
|
|
|
|
|
|
def test_markdown_upload_requires_confirmation(client, db):
|
|
user = create_user("confirm_user", "confirm@example.com", "secure password")
|
|
login(client, user.email)
|
|
|
|
response = client.post(
|
|
"/tasks/import/markdown",
|
|
data={"markdown_file": (BytesIO(b"- Not imported"), "tasks.md")},
|
|
content_type="multipart/form-data",
|
|
follow_redirects=True,
|
|
)
|
|
|
|
assert b"This field is required" in response.data
|
|
assert db.session.scalar(select(Task)) is None
|
|
|
|
|
|
def test_json_export_is_versioned_complete_and_private(client, db):
|
|
user = create_user("backup_user", "backup@example.com", "secure password")
|
|
user.theme = "dark"
|
|
task = make_task(user.id, "Portable task")
|
|
task.description = "Useful notes"
|
|
task.timer_seconds = 600
|
|
task.context = "Quiet desk"
|
|
task.chunk_sessions = [{"minutes": 10, "completed": True}]
|
|
task.subtasks.append(Subtask(title="Portable step", completed=True))
|
|
db.session.commit()
|
|
login(client, user.email)
|
|
|
|
response = client.get("/tasks/export/json")
|
|
document = json.loads(response.data)
|
|
|
|
assert response.status_code == 200
|
|
assert response.mimetype == "application/json"
|
|
assert "attachment" in response.headers["Content-Disposition"]
|
|
assert document["schema_version"] == 1
|
|
assert document["settings"]["theme"] == "dark"
|
|
assert document["tasks"][0]["title"] == "Portable task"
|
|
assert document["tasks"][0]["subtasks"][0]["title"] == "Portable step"
|
|
serialized = response.data.decode()
|
|
assert user.email not in serialized
|
|
assert "password" not in serialized
|
|
assert "user_id" not in serialized
|
|
|
|
|
|
def test_json_backup_round_trip_restores_nested_data_and_settings(db):
|
|
source = create_user("source", "source@example.com", "secure password")
|
|
source.completion_chime = False
|
|
task = make_task(source.id, "Round-trip task")
|
|
task.chunk_sessions = [{"minutes": 5, "completed": True}]
|
|
task.subtasks.append(Subtask(title="Nested step", completed=True))
|
|
db.session.commit()
|
|
content = export_backup(source)
|
|
|
|
target = create_user("target", "target@example.com", "secure password")
|
|
count = import_backup(target, content, replace_existing=False)
|
|
restored = db.session.scalar(select(Task).where(Task.user_id == target.id))
|
|
|
|
assert count == 1
|
|
assert restored.title == "Round-trip task"
|
|
assert restored.chunk_sessions == [{"minutes": 5, "completed": True}]
|
|
assert restored.subtasks[0].title == "Nested step"
|
|
assert target.completion_chime is False
|
|
|
|
|
|
def test_json_upload_route_assigns_tasks_to_current_user(client, db):
|
|
user = create_user("route_user", "route@example.com", "secure password")
|
|
login(client, user.email)
|
|
document = {
|
|
"schema_version": 1,
|
|
"settings": {"theme": "light"},
|
|
"tasks": [
|
|
{
|
|
"title": "Uploaded backup task",
|
|
"status": "not_started",
|
|
"priority": "important",
|
|
}
|
|
],
|
|
}
|
|
|
|
response = client.post(
|
|
"/tasks/import/json",
|
|
data={
|
|
"backup_file": (
|
|
BytesIO(json.dumps(document).encode()),
|
|
"steady.json",
|
|
),
|
|
"confirm": "y",
|
|
},
|
|
content_type="multipart/form-data",
|
|
follow_redirects=True,
|
|
)
|
|
restored = db.session.scalar(select(Task))
|
|
|
|
assert b"Merged 1 tasks from backup" in response.data
|
|
assert restored.user_id == user.id
|
|
assert restored.priority == "important"
|
|
assert user.theme == "light"
|
|
|
|
|
|
def test_invalid_replace_backup_leaves_existing_data_unchanged(db):
|
|
user = create_user("atomic_user", "atomic@example.com", "secure password")
|
|
existing = make_task(user.id, "Must survive")
|
|
invalid = json.dumps(
|
|
{
|
|
"schema_version": 1,
|
|
"settings": {"theme": "unknown"},
|
|
"tasks": [{"title": "Would replace"}],
|
|
}
|
|
).encode()
|
|
|
|
with pytest.raises(TransferValidationError, match="unknown theme"):
|
|
import_backup(user, invalid, replace_existing=True)
|
|
|
|
assert db.session.get(Task, existing.id) is not None
|
|
assert user.theme == "auto"
|
|
|
|
|
|
def test_replace_backup_affects_only_current_user(db):
|
|
source = create_user("replace_source", "replace-source@example.com", "secure password")
|
|
make_task(source.id, "Replacement task")
|
|
backup = export_backup(source)
|
|
|
|
target = create_user("replace_target", "replace-target@example.com", "secure password")
|
|
old_target = make_task(target.id, "Old target task")
|
|
other = create_user("other_user", "other@example.com", "secure password")
|
|
other_task = make_task(other.id, "Other user's task")
|
|
|
|
import_backup(target, backup, replace_existing=True)
|
|
|
|
assert db.session.get(Task, old_target.id) is None
|
|
assert db.session.get(Task, other_task.id) is not None
|
|
assert [task.title for task in list_tasks(target.id)] == ["Replacement task"]
|
|
|
|
|
|
def test_clear_and_undo_are_scoped_and_preserve_streak(db):
|
|
user = create_user("clear_user", "clear@example.com", "secure password")
|
|
completed = make_task(user.id, "Completed work")
|
|
complete_task(completed, completed_at=datetime(2026, 8, 8, 12, 0))
|
|
active = make_task(user.id, "Active work")
|
|
other = create_user("clear_other", "clear-other@example.com", "secure password")
|
|
other_completed = make_task(other.id, "Other completion", status="done")
|
|
|
|
batch_id, count = clear_completed_tasks(user.id)
|
|
|
|
assert count == 1
|
|
assert [task.id for task in list_tasks(user.id)] == [active.id]
|
|
with pytest.raises(LookupError):
|
|
get_task(completed.id, user.id)
|
|
assert calculate_forgiving_streak(
|
|
user.id, today=date(2026, 8, 8)
|
|
).days == 1
|
|
assert db.session.get(Task, other_completed.id).cleared_at is None
|
|
|
|
assert undo_clear_completed(other.id, batch_id) == 0
|
|
assert undo_clear_completed(user.id, batch_id) == 1
|
|
assert db.session.get(Task, completed.id).cleared_at is None
|
|
|
|
|
|
def test_clear_route_surfaces_working_undo_action(client, db):
|
|
user = create_user("undo_user", "undo@example.com", "secure password")
|
|
completed = make_task(user.id, "Undoable completion", status="done")
|
|
login(client, user.email)
|
|
|
|
clear_response = client.post(
|
|
"/tasks/completed/clear",
|
|
follow_redirects=True,
|
|
)
|
|
db.session.refresh(completed)
|
|
|
|
assert b"Undo clear" in clear_response.data
|
|
assert completed.cleared_at is not None
|
|
|
|
undo_response = client.post(
|
|
f"/tasks/completed/undo/{completed.clear_batch_id}",
|
|
follow_redirects=True,
|
|
)
|
|
db.session.refresh(completed)
|
|
assert b"Restored 1 completed tasks" in undo_response.data
|
|
assert completed.cleared_at is None
|