bvandeusenandClaude Opus 5.5
cf3de5bae1
CI & Build / Plugin hooks (push) Successful in 18s
CI & Build / Python lint (push) Successful in 3s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / integration (push) Successful in 1m23s
CI & Build / Python tests (push) Successful in 2m0s
CI & Build / Build & push image (push) Successful in 36s
moment_actions.resolve(tool, input) names every moment a call reaches and the action that reached it. One call can reach several: kubectl apply is a run, a deliver and a reach outside the workspace. Command tools match by how each segment of the line starts, with a word boundary; other tools by field=value arguments. The MCP server prefix and case are ignored. 56 shipped defaults cover the harness tools, Scribe tools and common command shapes. moment_mappings (migration 0116) holds what an install adds and the defaults it switches off. A removal is a stored row, so an upgrade does not switch the default back on. Per the operator ruling, corrections happen in the session: map_action and unmap_action (write tools) return now_reaches so the fix can be confirmed in the same reply. list_moments now shows each moment's actions on this install. REST mirrors both doors, recorded as human. Backup v21 carries the mappings. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
"""Real-Postgres tests for an install's moment mappings (milestone 458 step 2).
|
|
|
|
What these pin is what the in-session correction promises the operator: a
|
|
mapping made is in force on the next call, a default switched off stays off,
|
|
switching it back on leaves no residue, one user's corrections are not
|
|
another's, and asking to remove what nothing maps is refused rather than
|
|
recorded. Every one of those is a claim about rows, so none of it is mocked.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import delete, select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.moment_mapping import MomentMapping
|
|
from scribe.services import moment_actions as ma
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
OWNER_USERNAME = "moment_mapping_owner"
|
|
STRANGER_USERNAME = "moment_mapping_stranger"
|
|
|
|
SHIP = {"command": "make ship"}
|
|
CURL = {"command": "curl localhost:8000"}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def users():
|
|
async with async_session() as s:
|
|
owner = await ensure_user(s, OWNER_USERNAME)
|
|
stranger = await ensure_user(s, STRANGER_USERNAME)
|
|
await s.commit()
|
|
ids = (owner.id, stranger.id)
|
|
# At SETUP: the lane shares one database, so a previous run's rows
|
|
# are cleared before this one reads anything.
|
|
await s.execute(delete(MomentMapping).where(MomentMapping.user_id.in_(ids)))
|
|
await s.commit()
|
|
return ids
|
|
|
|
|
|
async def _reached(uid, tool, tool_input):
|
|
return [h["moment"] for h in await ma.moments_for(uid, tool, tool_input)]
|
|
|
|
|
|
async def _rows(uid):
|
|
async with async_session() as s:
|
|
return (await s.execute(
|
|
select(MomentMapping).where(MomentMapping.user_id == uid)
|
|
)).scalars().all()
|
|
|
|
|
|
async def test_a_mapping_is_in_force_on_the_next_call(users):
|
|
uid, _ = users
|
|
assert "work.deliver" not in await _reached(uid, "Bash", SHIP)
|
|
|
|
out = await ma.map_action(uid, "Bash", "make ship", "work.deliver",
|
|
reason="this install ships with make")
|
|
assert out["change"] == "mapped"
|
|
assert "work.deliver" in [h["moment"] for h in out["now_reaches"]]
|
|
assert "work.deliver" in await _reached(uid, "Bash", SHIP)
|
|
|
|
|
|
async def test_mapping_twice_is_one_row(users):
|
|
uid, _ = users
|
|
await ma.map_action(uid, "Bash", "make ship", "work.deliver")
|
|
again = await ma.map_action(uid, "Bash", "make ship", "work.deliver", reason="why")
|
|
assert again["change"].startswith("already mapped")
|
|
rows = await _rows(uid)
|
|
assert len(rows) == 1 and rows[0].reason == "why"
|
|
|
|
|
|
async def test_one_users_corrections_are_not_anothers(users):
|
|
uid, other = users
|
|
await ma.map_action(uid, "Bash", "make ship", "work.deliver")
|
|
assert "work.deliver" not in await _reached(other, "Bash", SHIP)
|
|
|
|
|
|
async def test_unmapping_an_installs_mapping_deletes_it(users):
|
|
uid, _ = users
|
|
await ma.map_action(uid, "Bash", "make ship", "work.deliver")
|
|
out = await ma.unmap_action(uid, "Bash", "make ship", "work.deliver")
|
|
assert out["change"] == "removed this install's mapping"
|
|
assert await _rows(uid) == []
|
|
assert "work.deliver" not in await _reached(uid, "Bash", SHIP)
|
|
|
|
|
|
async def test_a_default_switched_off_stays_off_and_switches_back_cleanly(users):
|
|
uid, _ = users
|
|
assert "env.reach" in await _reached(uid, "Bash", CURL)
|
|
|
|
off = await ma.unmap_action(uid, "Bash", "curl", "env.reach",
|
|
reason="curl here only hits the local dev server")
|
|
assert off["change"] == "switched off the shipped default"
|
|
assert await _reached(uid, "Bash", CURL) == ["work.run"]
|
|
[row] = await _rows(uid)
|
|
assert row.effect == ma.REMOVE
|
|
|
|
by_moment = await ma.actions_by_moment(uid)
|
|
assert [r["match"] for r in by_moment["removed_defaults"]] == ["curl"]
|
|
assert {"tool": "Bash", "match": "curl", "via": ma.DEFAULT} not in by_moment["actions"]["env.reach"]
|
|
|
|
on = await ma.map_action(uid, "Bash", "curl", "env.reach")
|
|
assert on["change"] == "restored the shipped default"
|
|
assert await _rows(uid) == []
|
|
assert "env.reach" in await _reached(uid, "Bash", CURL)
|
|
|
|
|
|
async def test_mapping_a_default_already_in_force_writes_nothing(users):
|
|
uid, _ = users
|
|
out = await ma.map_action(uid, "update_task", "status=done", "work.finish")
|
|
assert out["change"].startswith("already a shipped default")
|
|
assert await _rows(uid) == []
|
|
|
|
|
|
async def test_removing_what_nothing_maps_is_refused(users):
|
|
uid, _ = users
|
|
with pytest.raises(ValueError, match="nothing to remove"):
|
|
await ma.unmap_action(uid, "Bash", "make ship", "work.deliver")
|
|
assert await _rows(uid) == []
|