97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.domain import InvalidTransitionError, complete_task, start_task
|
|
from app.storage import StoredTask
|
|
|
|
|
|
def make_task(
|
|
*,
|
|
title: str = "Task",
|
|
started_at: datetime | None = None,
|
|
done_at: datetime | None = None,
|
|
) -> StoredTask:
|
|
return StoredTask(
|
|
id=uuid4(),
|
|
title=title,
|
|
created_at=datetime.now(UTC),
|
|
started_at=started_at,
|
|
done_at=done_at,
|
|
)
|
|
|
|
|
|
def test_start_task_from_backlog_sets_started_at():
|
|
task = make_task(started_at=None, done_at=None)
|
|
before_call = datetime.now(UTC)
|
|
|
|
result = start_task(task)
|
|
|
|
after_call = datetime.now(UTC)
|
|
|
|
assert result.id == task.id
|
|
assert result.title == task.title
|
|
assert result.created_at == task.created_at
|
|
assert result.done_at is None
|
|
assert result.started_at is not None
|
|
assert before_call <= result.started_at <= after_call
|
|
|
|
|
|
def test_start_task_is_idempotent_for_already_started_task():
|
|
started_at = datetime.now(UTC) - timedelta(minutes=5)
|
|
task = make_task(started_at=started_at, done_at=None)
|
|
|
|
result = start_task(task)
|
|
|
|
assert result == task
|
|
assert result.started_at == started_at
|
|
|
|
|
|
def test_start_task_raises_for_completed_task():
|
|
started_at = datetime.now(UTC) - timedelta(minutes=10)
|
|
done_at = datetime.now(UTC) - timedelta(minutes=1)
|
|
task = make_task(started_at=started_at, done_at=done_at)
|
|
|
|
with pytest.raises(InvalidTransitionError, match="invalid_transaction"):
|
|
start_task(task)
|
|
|
|
|
|
def test_complete_task_from_in_progress_sets_done_at():
|
|
started_at = datetime.now(UTC) - timedelta(minutes=10)
|
|
task = make_task(started_at=started_at, done_at=None)
|
|
before_call = datetime.now(UTC)
|
|
|
|
result = complete_task(task)
|
|
|
|
after_call = datetime.now(UTC)
|
|
|
|
assert result.id == task.id
|
|
assert result.title == task.title
|
|
assert result.created_at == task.created_at
|
|
assert result.started_at == started_at
|
|
assert result.done_at is not None
|
|
assert result.started_at <= result.done_at
|
|
assert before_call <= result.done_at <= after_call
|
|
|
|
|
|
def test_complete_task_is_idempotent_for_already_completed_task():
|
|
started_at = datetime.now(UTC) - timedelta(minutes=10)
|
|
done_at = datetime.now(UTC) - timedelta(minutes=1)
|
|
task = make_task(started_at=started_at, done_at=done_at)
|
|
|
|
result = complete_task(task)
|
|
|
|
assert result == task
|
|
assert result.done_at == done_at
|
|
|
|
|
|
def test_complete_task_raises_for_backlog_task():
|
|
task = make_task(started_at=None, done_at=None)
|
|
|
|
with pytest.raises(InvalidTransitionError, match="invalid_transaction"):
|
|
complete_task(task)
|
|
|