"""MCP system tools + task/note issue wiring (service layer mocked).""" from unittest.mock import AsyncMock, MagicMock, patch import pytest def _fake_system(sid=1, name="Reader", project_id=5): s = MagicMock() s.to_dict.return_value = {"id": sid, "name": name, "project_id": project_id} s.project_id = project_id return s @pytest.mark.asyncio async def test_create_system_returns_dict(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.create_system = AsyncMock(return_value=_fake_system(name="Reader")) from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name="Reader", description="pdf reader") assert result["name"] == "Reader" @pytest.mark.asyncio async def test_create_system_no_access_raises(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.create_system = AsyncMock(return_value=None) from scribe.mcp.tools.systems import create_system with pytest.raises(ValueError): await create_system(project_id=5, name="Reader") @pytest.mark.asyncio async def test_get_system_splits_records_by_kind(): issue = MagicMock(); issue.to_dict.return_value = {"id": 10}; issue.task_kind = "issue"; issue.status = "todo" work = MagicMock(); work.to_dict.return_value = {"id": 11}; work.task_kind = "work"; work.status = "todo" note = MagicMock(); note.to_dict.return_value = {"id": 12}; note.task_kind = "work"; note.status = None with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.get_system = AsyncMock(return_value=_fake_system(sid=3)) svc.list_records_for_system = AsyncMock(return_value=[issue, work, note]) from scribe.mcp.tools.systems import get_system result = await get_system(system_id=3) assert [r["id"] for r in result["issues"]] == [10] assert [r["id"] for r in result["tasks"]] == [11] assert [r["id"] for r in result["notes"]] == [12] @pytest.mark.asyncio async def test_update_system_not_found_raises(): with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.update_system = AsyncMock(return_value=None) from scribe.mcp.tools.systems import update_system with pytest.raises(ValueError): await update_system(system_id=99, name="x") @pytest.mark.asyncio async def test_create_task_issue_sets_kind_provenance_and_systems(): note = MagicMock(); note.id = 50; note.to_dict.return_value = {"id": 50, "task_kind": "issue"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.systems_svc") as systems_svc: notes_svc.create_note = AsyncMock(return_value=note) systems_svc.set_record_systems = AsyncMock(return_value=[2, 3]) systems_svc.list_record_systems = AsyncMock(return_value=[]) from scribe.mcp.tools.tasks import create_task result = await create_task(title="bug", kind="issue", system_ids=[2, 3], arose_from_id=9) _, kwargs = notes_svc.create_note.call_args assert kwargs["task_kind"] == "issue" assert kwargs["arose_from_id"] == 9 systems_svc.set_record_systems.assert_awaited_once_with(1, 50, [2, 3]) assert result["id"] == 50 @pytest.mark.asyncio async def test_untagged_hint_names_the_projects_systems(): sys_a = MagicMock(); sys_a.id = 1; sys_a.name = "workers" sys_b = MagicMock(); sys_b.id = 2; sys_b.name = "scrape-pipeline" with patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[sys_a, sys_b]) from scribe.mcp.tools.systems import untagged_systems_hint hint = await untagged_systems_hint(1, 5) assert "workers" in hint and "scrape-pipeline" in hint assert "system_ids" in hint assert "create_system" in hint @pytest.mark.asyncio async def test_untagged_hint_zero_systems_prompts_first_create_and_fails_open(): from scribe.mcp.tools.systems import untagged_systems_hint with patch("scribe.mcp.tools.systems.systems_svc") as svc: # Zero Systems is the state nothing else nudges out of — the hint must # prompt the FIRST create_system, not go silent. svc.list_systems = AsyncMock(return_value=[]) hint = await untagged_systems_hint(1, 5) assert "no Systems yet" in hint and "create_system" in hint with patch("scribe.mcp.tools.systems.systems_svc") as svc: # A hint must never break a create — DB failure degrades to no hint. svc.list_systems = AsyncMock(side_effect=RuntimeError("db down")) assert await untagged_systems_hint(1, 5) is None @pytest.mark.asyncio async def test_create_system_same_normalized_name_is_duplicate_gated(): existing = _fake_system(sid=7, name="Scrape Pipeline") existing.id = 7 existing.name = "Scrape Pipeline" with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[existing]) svc.create_system = AsyncMock() from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name=" scrape pipeline ") assert result["duplicate"] is True assert result["existing_id"] == 7 svc.create_system.assert_not_awaited() @pytest.mark.asyncio async def test_create_system_distinct_name_passes_the_gate(): other = _fake_system(sid=7, name="Workers") other.name = "Workers" with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \ patch("scribe.mcp.tools.systems.systems_svc") as svc: svc.list_systems = AsyncMock(return_value=[other]) svc.create_system = AsyncMock(return_value=_fake_system(sid=8, name="Exporter")) from scribe.mcp.tools.systems import create_system result = await create_system(project_id=5, name="Exporter") assert result["name"] == "Exporter" @pytest.mark.asyncio async def test_create_task_untagged_in_project_carries_the_question(): note = MagicMock(); note.id = 60; note.to_dict.return_value = {"id": 60} sys_a = MagicMock(); sys_a.id = 4; sys_a.name = "plugin-hooks" with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.dedup_svc") as dedup_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.create_note = AsyncMock(return_value=note) dedup_svc.find_duplicate_note = AsyncMock(return_value=None) seam_svc.list_record_systems = AsyncMock(return_value=[]) seam_svc.list_systems = AsyncMock(return_value=[sys_a]) from scribe.mcp.tools.tasks import create_task result = await create_task(title="untagged", project_id=5) assert "plugin-hooks" in result["systems_hint"] @pytest.mark.asyncio async def test_create_task_tagged_shows_systems_and_projectless_gets_neither(): note = MagicMock(); note.id = 61 # Fresh dict per call — a shared return_value dict lets the first create's # mutation leak into the second create's response. note.to_dict.side_effect = lambda: {"id": 61} tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 4, "name": "plugin-hooks"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.dedup_svc") as dedup_svc, \ patch("scribe.mcp.tools.tasks.systems_svc") as systems_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.create_note = AsyncMock(return_value=note) dedup_svc.find_duplicate_note = AsyncMock(return_value=None) systems_svc.set_record_systems = AsyncMock() seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys]) from scribe.mcp.tools.tasks import create_task tagged = await create_task(title="tagged", project_id=5, system_ids=[4]) seam_svc.list_record_systems = AsyncMock(return_value=[]) orphan = await create_task(title="orphan", project_id=0) assert tagged["systems"] == [{"id": 4, "name": "plugin-hooks"}] assert "systems_hint" not in tagged assert "systems" not in orphan and "systems_hint" not in orphan @pytest.mark.asyncio async def test_get_task_shows_the_records_systems_on_read(): """The touching-a-System reflex needs the affiliation visible on READ — a session opening the task it is about to work must see its areas.""" note = MagicMock() note.id = 70; note.user_id = 1; note.project_id = 5 note.deleted_at = None; note.parent_id = None note.to_dict.return_value = {"id": 70, "task_kind": "work"} tagged_sys = MagicMock(); tagged_sys.to_dict.return_value = {"id": 2, "name": "exporter"} with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.tasks.access_svc") as access_svc, \ patch("scribe.mcp.tools.tasks.record_pulled"), \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: notes_svc.get_note_for_user = AsyncMock(return_value=(note, "owner")) access_svc.describe_provenance = AsyncMock(return_value={}) seam_svc.list_record_systems = AsyncMock(return_value=[tagged_sys]) from scribe.mcp.tools.tasks import get_task result = await get_task(task_id=70) assert result["systems"] == [{"id": 2, "name": "exporter"}] @pytest.mark.asyncio async def test_add_task_log_on_untagged_project_task_asks_the_question(): """Logging work IS working in some area — the strongest moment to ask.""" log = MagicMock(); log.to_dict.return_value = {"id": 9, "task_id": 70} task = MagicMock(); task.user_id = 1; task.project_id = 5 with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \ patch("scribe.mcp.tools.tasks.task_logs_svc") as logs_svc, \ patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \ patch("scribe.mcp.tools.systems.systems_svc") as seam_svc: logs_svc.create_log = AsyncMock(return_value=log) notes_svc.get_note_for_user = AsyncMock(return_value=(task, "owner")) seam_svc.list_record_systems = AsyncMock(return_value=[]) seam_svc.list_systems = AsyncMock(return_value=[]) from scribe.mcp.tools.tasks import add_task_log result = await add_task_log(task_id=70, content="progress") assert "no Systems yet" in result["systems_hint"]