"""Which actions reach which moment (milestone 458 step 2) — the pure half. Matching and resolution take no database: the defaults are code and an install's mappings arrive as rows, so every case below hands `resolve` the rows it needs. The writes that store those rows are exercised against real Postgres in `test_integration_moment_mappings.py`. """ from types import SimpleNamespace from unittest.mock import AsyncMock, patch import pytest from scribe.services import moment_actions as ma from scribe.services import moments def _row(tool, match, moment, effect=ma.ADD): return SimpleNamespace(tool=tool, match=match, moment=moment, effect=effect) def _reached(tool, tool_input=None, mappings=()): return [h["moment"] for h in ma.resolve(tool, tool_input, mappings)] # ── the defaults ───────────────────────────────────────────────────────── def test_every_default_reaches_a_real_moment(): """A default naming a moment that is not in the catalog would never be mounted on — the typo require_moment refuses at the door.""" bad = [a for a in ma.DEFAULT_ACTIONS if not moments.is_moment(a.moment)] assert not bad, bad def test_no_default_is_listed_twice(): keys = [(ma.tool_key(a.tool), a.match, a.moment) for a in ma.DEFAULT_ACTIONS] assert len(keys) == len(set(keys)) def test_every_default_is_a_mapping_the_door_would_accept(): """The defaults pass the same validation an install's mapping does, so a default and an install's copy of it can never disagree about its shape.""" for a in ma.DEFAULT_ACTIONS: assert ma._clean(a.tool, a.match, a.moment) == (a.tool, a.match, a.moment) # ── matching ───────────────────────────────────────────────────────────── @pytest.mark.parametrize("command", [ "git push origin dev", "git push", "cd app && git push", "make lint; git push --tags", "GIT_TRACE=1 git push", " git push ", ]) def test_a_command_prefix_matches_any_segment_of_the_line(command): assert "work.deliver" in _reached("Bash", {"command": command}) @pytest.mark.parametrize("command", ["git pushd", "echo git push", "git status"]) def test_a_command_prefix_needs_a_word_boundary_and_the_head_of_a_segment(command): assert "work.deliver" not in _reached("Bash", {"command": command}) def test_one_action_reaches_every_moment_it_is(): """`kubectl apply` is a run, a deliver and a reach outside the workspace.""" assert _reached("Bash", {"command": "kubectl apply -f deploy.yaml"}) == [ "work.run", "work.deliver", "env.reach", ] def test_the_most_specific_action_is_the_one_named(): hits = ma.resolve("Bash", {"command": "git push"}) deliver = next(h for h in hits if h["moment"] == "work.deliver") assert deliver["match"] == "git push" run = next(h for h in hits if h["moment"] == "work.run") assert run["match"] == "" def test_an_ordinary_command_is_only_a_run(): assert _reached("Bash", {"command": "ls -la"}) == ["work.run"] def test_an_unknown_tool_reaches_nothing(): assert _reached("SomeOtherTool", {"x": 1}) == [] @pytest.mark.parametrize("name", [ "mcp__plugin_scribe_scribe__update_task", "mcp__scribe__update_task", "update_task", "Update_Task", ]) def test_the_server_prefix_and_case_do_not_matter(name): assert _reached(name, {"task_id": 1, "status": "done"}) == ["work.finish"] @pytest.mark.parametrize("status,expected", [ ("done", ["work.finish"]), ("cancelled", ["work.finish"]), ("in_progress", ["work.start"]), ("todo", []), ("", []), ]) def test_arguments_select_the_moment(status, expected): assert _reached("update_task", {"task_id": 1, "status": status}) == expected def test_a_whole_tool_mapping_ignores_its_arguments(): assert _reached("Edit", {"file_path": "x", "old_string": "a"}) == ["work.change"] def test_loading_a_procedure_reaches_its_own_moment(): assert _reached("Skill", {"skill": "release-notes"}) == ["skill.release-notes"] def test_a_bundled_skill_also_reaches_the_moment_it_is_for(): """Loading the planning procedure IS planning: a rule mounted on work.plan arrives on the load, not only on the plan it goes on to open.""" assert _reached("Skill", {"skill": "scribe:writing-plans"}) == [ "work.plan", "skill.scribe:writing-plans", ] assert _reached("Skill", {"skill": "scribe:reporting-back"})[0] == "reply.report" def test_only_the_products_own_skill_reaches_a_bundled_default(): """Another plugin's `verification` skill is not the product's procedure; the plugin's name is what makes the declaration ours.""" assert _reached("Skill", {"skill": "other:verification"}) == ["skill.other:verification"] def test_a_procedure_name_that_is_not_a_moment_reaches_nothing(): assert _reached("Skill", {"skill": "two words"}) == [] assert _reached("Skill", {}) == [] # ── a stored process's own moments ─────────────────────────────────────── def test_a_process_declares_moments_its_load_reaches(): hits = ma.resolve("Skill", {"skill": "scribe-proc-release"}, [], ["work.deliver"]) assert [(h["moment"], h["via"]) for h in hits] == [ ("work.deliver", ma.PROCESS), ("skill.scribe-proc-release", ma.DEFAULT), ] assert hits[0]["match"] == "skill=scribe-proc-release" def test_a_declared_moment_another_action_reaches_is_named_once(): rows = [_row("Skill", "skill=scribe-proc-release", "work.deliver")] hits = ma.resolve("Skill", {"skill": "scribe-proc-release"}, rows, ["work.deliver"]) assert [h["moment"] for h in hits].count("work.deliver") == 1 def test_declared_moments_only_reach_a_skill_load(): hits = ma.resolve("Bash", {"command": "ls"}, [], ["work.deliver"]) assert [h["moment"] for h in hits] == ["work.run"] @pytest.mark.parametrize("data, expected", [ (None, []), ({}, []), ({"moments": ["work.deliver", "Work.Deliver", "reply.report"]}, ["work.deliver", "reply.report"]), ({"moments": "work.verify"}, ["work.verify"]), ({"moments": ["work.finished", "skill.two words", "work.debug"]}, ["work.debug"]), ]) def test_a_process_field_is_read_defensively(data, expected): """A moment the catalog no longer has stops firing; it does not break the load.""" assert ma.declared_moments(data) == expected async def test_only_a_process_skill_asks_for_its_process(): asked = AsyncMock(return_value=["work.deliver"]) with patch.object(ma, "list_mappings", AsyncMock(return_value=[])), \ patch.object(ma, "process_moments", asked): reached = await ma.moments_for(1, "Skill", {"skill": "scribe-proc-release"}) await ma.moments_for(1, "Skill", {"skill": "scribe:verification"}) await ma.moments_for(1, "Bash", {"command": "git push"}) assert "work.deliver" in [h["moment"] for h in reached] asked.assert_awaited_once_with(1, "scribe-proc-release") async def _process_moments(manifest, note=None, *, raises=False): from scribe.services import notes as notes_svc from scribe.services import plugin_context built = AsyncMock(side_effect=RuntimeError("down")) if raises else AsyncMock(return_value=manifest) loader = AsyncMock(return_value=(note, None) if note is not None else None) with patch.object(plugin_context, "build_process_manifest", built), \ patch.object(notes_svc, "get_note_for_user", loader): return await ma.process_moments(1, "scribe-proc-release"), loader async def test_a_process_slug_resolves_through_the_sync_manifest(): """The same manifest the sync script wrote the skill from, so the slug names the same process here as it did on disk.""" note = SimpleNamespace(data={"moments": ["work.deliver"]}) found, loader = await _process_moments( {"processes": [{"id": 7, "slug": "other"}, {"id": 9, "slug": "release"}]}, note) assert found == ["work.deliver"] loader.assert_awaited_once_with(1, 9) async def test_an_unknown_or_unreadable_process_reaches_nothing_more(): found, loader = await _process_moments({"processes": [{"id": 7, "slug": "other"}]}) assert found == [] loader.assert_not_awaited() found, _ = await _process_moments({}, raises=True) assert found == [] # ── an install's own ───────────────────────────────────────────────────── def test_an_install_mapping_adds_a_moment(): rows = [_row("Bash", "make ship", "work.deliver")] assert "work.deliver" in _reached("Bash", {"command": "make ship"}, rows) assert "work.deliver" not in _reached("Bash", {"command": "make ship"}) def test_removing_a_default_removes_only_that_default(): """`curl` stops reaching env.reach here; it is still a run, and `ssh` still reaches out — a removal overrides nothing it did not name.""" rows = [_row("Bash", "curl", "env.reach", ma.REMOVE)] assert _reached("Bash", {"command": "curl localhost:8000"}, rows) == ["work.run"] assert "env.reach" in _reached("Bash", {"command": "ssh box"}, rows) def test_a_removal_matches_the_default_whatever_the_tool_is_called(): rows = [_row("mcp__x__update_task", "status=done", "work.finish", ma.REMOVE)] assert _reached("update_task", {"status": "done"}, rows) == [] def test_an_install_mapping_is_named_as_the_installs(): rows = [_row("Bash", "make ship", "work.deliver")] hit = next(h for h in ma.resolve("Bash", {"command": "make ship"}, rows) if h["moment"] == "work.deliver") assert (hit["via"], hit["match"]) == (ma.INSTALL, "make ship") # ── validation ─────────────────────────────────────────────────────────── def test_a_prefix_on_a_tool_that_runs_no_command_is_refused(): """It would be stored and never match — a mapping that looks made and is not, which is the misfire this table exists to fix.""" with pytest.raises(ValueError, match="field=value"): ma._clean("update_task", "done", "work.finish") def test_an_unknown_moment_is_refused(): with pytest.raises(ValueError, match="unknown moment"): ma._clean("Bash", "make ship", "work.ship") def test_a_mapping_is_normalised_before_it_is_stored(): assert ma._clean("mcp__x__update_task", " status=done ", " Work.Finish") == ( "update_task", "status=done", "work.finish", ) assert ma._clean("Bash", "make ship", "work.deliver")[1] == "make ship" def test_a_tool_is_required(): with pytest.raises(ValueError, match="tool is required"): ma._clean(" ", "", "work.run") # ── failing open ───────────────────────────────────────────────────────── async def test_an_unreadable_mapping_table_costs_the_corrections_not_the_moments(): with patch.object(ma, "list_mappings", AsyncMock(side_effect=RuntimeError("db down"))): hits = await ma.moments_for(7, "Bash", {"command": "git push"}) assert [h["moment"] for h in hits] == ["work.run", "work.deliver"]