Files
FabledScribe/tests/test_integration_placement.py
T
bvandeusenandClaude Opus 5 46d9134b10
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 9s
CI & Build / TypeScript typecheck (push) Successful in 53s
CI & Build / integration (push) Successful in 57s
CI & Build / Python tests (push) Successful in 1m30s
CI & Build / Build & push image (push) Successful in 27s
feat(409): a task write returns where the task sits (#4010)
Step 1 of milestone 409 "Response shapes". An agent reporting finished work
is asked to say which milestone it belongs to, which step of how many, and
what is next. Without those facts to hand it reconstructs them, and a
reconstruction reads exactly like the truth when it is wrong.

create_task and update_task (MCP) and the REST create/update task routes now
return a placement block: project; and for a task in a milestone, the
milestone, position {step, of}, progress {completed, total, pct} and next
(the next open step, falling back to the earliest open one before it).

- Step order is creation order, not get_milestone listing order, which
  reshuffles on every update.
- Siblings are read through readable_notes_clause; position and progress
  are computed over that same readable set, so a collaborator is never shown
  a step title they cannot open, and a note share alone reveals no plan.
- Fail-open and omitted when empty, like every in-band decoration.
- _no_embedding moves into conftest as one opt-in fixture for both
  integration modules that need it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-14 08:43:04 -04:00

97 lines
4.2 KiB
Python

"""Real-Postgres tests for task placement (milestone 409 step 1).
What a mock cannot show: that step order is creation order rather than update
order, that position and progress are computed over the steps the CALLER may
read, and that a collaborator holding one shared task is not shown the titles
of steps they cannot open.
"""
import pytest
import pytest_asyncio
from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.project import Project
from scribe.models.share import NoteShare, ProjectShare
from scribe.services import notes as notes_svc
from scribe.services.placement import task_placement
from scribe.services.record_batch import BatchItem, BatchMilestone, create_batch
from tests.helpers import ensure_user
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine", "_no_embedding")]
@pytest_asyncio.fixture
async def plan():
"""Owner, a collaborator, a project and a four-step milestone."""
async with async_session() as s:
owner = await ensure_user(s, "placement_owner")
other = await ensure_user(s, "placement_collaborator")
project = Project(user_id=owner.id, title="Placement target")
s.add(project)
await s.flush()
ids = {"owner": owner.id, "other": other.id, "pid": project.id}
await s.commit()
ms, steps = await create_batch(ids["owner"], [
BatchItem(title="Step 1", status="done"),
BatchItem(title="Step 2", status="in_progress"),
BatchItem(title="Step 3"),
BatchItem(title="Step 4"),
], project_id=ids["pid"], milestone=BatchMilestone(title="The plan", body="b"))
ids.update({"mid": ms.id, "steps": steps})
return ids
async def test_the_owner_sees_the_whole_placement(plan):
step2 = plan["steps"][1]
out = await task_placement(plan["owner"], step2)
assert out["project"] == {"id": plan["pid"], "title": "Placement target"}
assert out["milestone"] == {"id": plan["mid"], "title": "The plan", "status": "active"}
assert out["position"] == {"step": 2, "of": 4}
assert out["progress"] == {"completed": 1, "total": 4, "pct": 25.0}
assert out["next"] == {"id": plan["steps"][2].id, "title": "Step 3", "status": "todo"}
async def test_step_order_is_creation_order_not_last_update(plan):
"""Touching step 1 must not move it to the end of the list."""
first = plan["steps"][0]
updated = await notes_svc.update_note(plan["owner"], first.id, body="edited later")
out = await task_placement(plan["owner"], updated)
assert out["position"] == {"step": 1, "of": 4}
async def test_the_last_step_points_back_at_the_earliest_open_one(plan):
last = await notes_svc.update_note(plan["owner"], plan["steps"][3].id, status="done")
out = await task_placement(plan["owner"], last)
assert out["next"]["title"] == "Step 2"
assert out["progress"]["completed"] == 2
async def test_a_task_outside_any_milestone_gets_its_project_alone(plan):
loose = await notes_svc.create_note(plan["owner"], title="Loose task", status="todo",
project_id=plan["pid"])
out = await task_placement(plan["owner"], loose)
assert out == {"project": {"id": plan["pid"], "title": "Placement target"}}
async def test_a_collaborator_holding_one_shared_task_sees_no_plan(plan):
"""A note share opens one task. It does not open the project, so neither the
milestone nor its other steps' titles may leak through the placement."""
step3 = plan["steps"][2]
async with async_session() as s:
s.add(NoteShare(note_id=step3.id, shared_with_user_id=plan["other"],
permission="editor", invited_by=plan["owner"]))
await s.commit()
assert await task_placement(plan["other"], step3) is None
async def test_a_project_collaborator_sees_the_plan(plan):
async with async_session() as s:
s.add(ProjectShare(project_id=plan["pid"], shared_with_user_id=plan["other"],
permission="viewer", invited_by=plan["owner"]))
await s.commit()
step3 = await s.get(Note, plan["steps"][2].id)
out = await task_placement(plan["other"], step3)
assert out["milestone"]["title"] == "The plan"
assert out["position"] == {"step": 3, "of": 4}
assert out["next"]["title"] == "Step 4"