"""Real-Postgres tests for "recently touched" in the enter_project handshake (#4045). What a mock cannot show: that a work-log moves its task up the list even though logging never bumps the task's updated_at, and that a milestone whose steps changed today reads as touched today even though its own row is old. """ from datetime import datetime, timedelta, timezone import pytest import pytest_asyncio from scribe.models import async_session from scribe.models.milestone import Milestone from scribe.models.note import Note from scribe.models.project import Project from scribe.models.task_log import TaskLog from scribe.services import milestones as milestones_svc from scribe.services import notes as notes_svc from tests.helpers import ensure_user pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")] NOW = datetime.now(timezone.utc) @pytest_asyncio.fixture async def world(): """Two open tasks and two milestones with deliberately staged timestamps.""" async with async_session() as s: owner = await ensure_user(s, "handshake_owner") project = Project(user_id=owner.id, title="Handshake target") s.add(project) await s.flush() # Milestone "old plan": its own row is 30 days old, one step changed now. old_plan = Milestone(user_id=owner.id, project_id=project.id, title="old plan", updated_at=NOW - timedelta(days=30)) # Milestone "recent edit": its own row changed 2 days ago, no steps. recent_edit = Milestone(user_id=owner.id, project_id=project.id, title="recent edit", updated_at=NOW - timedelta(days=2)) s.add_all([old_plan, recent_edit]) await s.flush() # "logged": last edited 10 days ago, but worked through a log just now. logged = Note(user_id=owner.id, project_id=project.id, title="logged", status="todo", task_kind="work", updated_at=NOW - timedelta(days=10)) # "edited": last edited 1 day ago, no logs. edited = Note(user_id=owner.id, project_id=project.id, title="edited", status="todo", task_kind="work", updated_at=NOW - timedelta(days=1)) step = Note(user_id=owner.id, project_id=project.id, milestone_id=old_plan.id, title="step", status="done", task_kind="work", updated_at=NOW) s.add_all([logged, edited, step]) await s.flush() s.add(TaskLog(task_id=logged.id, user_id=owner.id, content="worked on it", created_at=NOW)) ids = {"owner": owner.id, "pid": project.id, "logged": logged.id, "edited": edited.id, "old_plan": old_plan.id, "recent_edit": recent_edit.id} await s.commit() return ids async def test_a_work_log_counts_as_touching_its_task(world): by_edit, _ = await notes_svc.list_notes( world["owner"], is_task=True, project_id=world["pid"], status=["todo", "in_progress"], sort="updated_at", ) by_touch, _ = await notes_svc.list_notes( world["owner"], is_task=True, project_id=world["pid"], status=["todo", "in_progress"], sort="touched", ) assert [t.id for t in by_edit] == [world["edited"], world["logged"]] assert [t.id for t in by_touch] == [world["logged"], world["edited"]] async def test_a_step_changing_counts_as_touching_its_milestone(world): rows = await milestones_svc.get_project_milestone_summary(world["owner"], world["pid"]) brief, omitted = milestones_svc.brief_milestone_summary(rows, limit=1) assert omitted == 1 assert [m["id"] for m in brief] == [world["old_plan"]]