bvandeusenandClaude Opus 5.5
914f31359c
CI & Build / Python lint (push) Successful in 5s
CI & Build / Plugin hooks (push) Successful in 17s
CI & Build / TypeScript typecheck (push) Successful in 57s
CI & Build / integration (push) Successful in 1m19s
CI & Build / Python tests (push) Successful in 2m2s
CI & Build / Build & push image (push) Successful in 15s
Docstrings, section headers and assertion messages drop the record number, milestone or step that introduced the behaviour, and incident accounts are restated as the present-tense failure shape. A dead pointer to a design doc that is not in the repo is removed. Renamed to the property they check: test_the_old_shape_would_be_caught -> test_a_rank_then_scope_query_is_caught test_a_ledger_repeat_no_longer_produces_that_case_at_all -> test_a_ledger_repeat_is_counted_not_withheld test_the_header_no_longer_promises_once_per_session -> test_the_header_explains_the_seen_marker test_the_portal_case_no_longer_stamps_one_canon_onto_a_whole_file -> test_one_canon_is_not_stamped_onto_a_whole_file test_the_repos_own_prose_no_longer_defines_with -> test_the_repos_own_prose_does_not_define_with test_the_original_regression_create_note_with_content -> test_the_real_server_refuses_content_on_create_note Fixture and example ids stay: "note 5" in a supersession chain, "rule 156" in an illustrative question, `#2275` as the hex-versus-citation case, `#3244` as the example of a bare number, fixed dates in date-filter tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
268 lines
11 KiB
Python
268 lines
11 KiB
Python
"""Which actions reach which moment — 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"]
|