"""The engine repositories behind the v2 UI: read and write paths. What these tests defend: * ``last_diff`` and ``merged_sha`` are patched under the version lock and read back unchanged (the task page shows both) * the organization turn feed never returns another organization's turns, even when both organizations worked in the same window (turns carry no organization column, so the tenant filter goes through the task) * stage counts are per organization: one organization's board does not count another organization's tasks """ from __future__ import annotations from datetime import datetime, timedelta import pytest_asyncio from app.domain.base import new_id, utcnow from app.domain.workforce.engine import EngineTurn from app.domain.workforce.models import RoleKey from app.domain.workforce.pipeline import Mode, Stage, TaskPipelineState from app.infrastructure.db.models import OrganizationRow, RunRow, TaskRow, UserRow from app.infrastructure.repositories.engine import ( EngineTurnRepository, PipelineStateRepository, ) @pytest_asyncio.fixture async def two_orgs(session_factory): """Two organizations, each with its own owner: the tenant boundary under test.""" org_ids: list[str] = [] async with session_factory() as session: for label in ("الف", "ب"): user = UserRow(email=f"repo-ui-{new_id()[:8]}@example.com", password_hash="x") session.add(user) await session.flush() org = OrganizationRow( name=f"سازمان {label}", slug=f"repo-ui-{new_id()[:8]}", owner_user_id=user.id ) session.add(org) await session.flush() org_ids.append(org.id) await session.commit() return session_factory, org_ids async def _add_task(session_factory, organization_id: str, title: str) -> str: async with session_factory() as session: task = TaskRow(organization_id=organization_id, title=title) session.add(task) await session.commit() return task.id async def _add_run(session_factory, task_id: str) -> str: async with session_factory() as session: run = RunRow(task_id=task_id) session.add(run) await session.commit() return run.id async def _add_state( session_factory, organization_id: str, task_id: str, stage: Stage ) -> TaskPipelineState: async with session_factory() as session: state = await PipelineStateRepository(session).create( TaskPipelineState(task_id=task_id, organization_id=organization_id, stage=stage) ) await session.commit() return state async def _add_turn( session_factory, task_id: str, run_id: str, excerpt: str, started_at: datetime ) -> None: async with session_factory() as session: await EngineTurnRepository(session).create( EngineTurn( task_id=task_id, run_id=run_id, role=RoleKey.DEVELOPER, mode=Mode.WORK, stage=Stage.PENDING_IMPLEMENTATION, reply_excerpt=excerpt, started_at=started_at, ) ) await session.commit() async def test_last_diff_and_merged_sha_are_patched_and_read_back(two_orgs): session_factory, (org_a, _org_b) = two_orgs task_id = await _add_task(session_factory, org_a, "کار دیف") state = await _add_state(session_factory, org_a, task_id, Stage.PENDING_IMPLEMENTATION) assert state.last_diff == "" assert state.merged_sha == "" diff_text = "diff --git a/app.py b/app.py\n+print('سلام')\n" sha = "ab12cd34" * 5 async with session_factory() as session: patched = await PipelineStateRepository(session).patch( task_id=task_id, expected_version=state.version, last_diff=diff_text, merged_sha=sha, ) await session.commit() assert patched.last_diff == diff_text assert patched.merged_sha == sha assert patched.version == state.version + 1 async with session_factory() as session: read_back = await PipelineStateRepository(session).get_or_raise(task_id) assert read_back.last_diff == diff_text assert read_back.merged_sha == sha async def test_the_org_turn_feed_never_returns_another_organizations_turns(two_orgs): session_factory, (org_a, org_b) = two_orgs base = utcnow() task_a = await _add_task(session_factory, org_a, "کار الف") task_b = await _add_task(session_factory, org_b, "کار ب") run_a = await _add_run(session_factory, task_a) run_b = await _add_run(session_factory, task_b) # Interleaved in time, and org B's turn is the newest of all. await _add_turn(session_factory, task_a, run_a, "الف-یک", base + timedelta(seconds=1)) await _add_turn(session_factory, task_b, run_b, "ب-دو", base + timedelta(seconds=2)) await _add_turn(session_factory, task_a, run_a, "الف-سه", base + timedelta(seconds=3)) await _add_turn(session_factory, task_b, run_b, "ب-ده", base + timedelta(seconds=10)) async with session_factory() as session: repo = EngineTurnRepository(session) feed_a = await repo.list_for_org(org_a) feed_b = await repo.list_for_org(org_b) newest_a = await repo.list_for_org(org_a, limit=1) assert [t.reply_excerpt for t in feed_a] == ["الف-سه", "الف-یک"] assert all(t.task_id == task_a for t in feed_a) assert [t.reply_excerpt for t in feed_b] == ["ب-ده", "ب-دو"] assert all(t.task_id == task_b for t in feed_b) assert [t.reply_excerpt for t in newest_a] == ["الف-سه"] async def test_stage_counts_are_per_organization(two_orgs): session_factory, (org_a, org_b) = two_orgs for title in ("الف یک", "الف دو"): task_id = await _add_task(session_factory, org_a, title) await _add_state(session_factory, org_a, task_id, Stage.PENDING_IMPLEMENTATION) review_id = await _add_task(session_factory, org_a, "الف سه") await _add_state(session_factory, org_a, review_id, Stage.PENDING_MANAGER_REVIEW) b_task = await _add_task(session_factory, org_b, "ب یک") await _add_state(session_factory, org_b, b_task, Stage.PENDING_IMPLEMENTATION) async with session_factory() as session: repo = PipelineStateRepository(session) counts_a = await repo.stage_counts(org_a) counts_b = await repo.stage_counts(org_b) at_implementation = await repo.count_at_by_org(Stage.PENDING_IMPLEMENTATION) at_review = await repo.count_at_by_org(Stage.PENDING_MANAGER_REVIEW) assert counts_a == { str(Stage.PENDING_IMPLEMENTATION): 2, str(Stage.PENDING_MANAGER_REVIEW): 1, } assert counts_b == {str(Stage.PENDING_IMPLEMENTATION): 1} assert at_implementation == {org_a: 2, org_b: 1} assert at_review == {org_a: 1}