bvandeusenandClaude Opus 5.5
0f5ef0be63
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 12s
CI & Build / TypeScript typecheck (push) Successful in 56s
CI & Build / integration (push) Successful in 1m4s
CI & Build / Python tests (push) Successful in 1m55s
CI & Build / Build & push image (push) Successful in 44s
Step 2 of the Scribe -> Grimoire rename. - src/scribe -> src/grimoire, and every import, patch target, module path, hypercorn entry point and `python -m grimoire.restore` with it; pyproject, uv.lock and the frontend package are named grimoire. - Image git.fabledsword.com/bvandeusen/fabledgrimoire; CI's Postgres service runs as grimoire / grimoire_test. - Compose files default the DB user, database and (prod) volume to grimoire, each overridable from .env, so an install created as Scribe keeps its data by setting POSTGRES_USER, POSTGRES_DB and PGDATA_VOLUME. - Backups are written as grimoire-<ts>.tar with format grimoire-backup-tar. Archives named scribe-* with the scribe-backup-tar manifest still restore and are still counted at the destination (folder and S3), with tests. - Exports, the MCP auth realm, the ASGI scope key and the persistent logger carry the new name. Kept on purpose: the secret-sealing key label stays "scribe.secrets.v1", because every stored secret was derived under it. Plugin identity (plugin name, hooks, env vars, .scribe marker, skills) is step 3. 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.
|
|
|
|
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 grimoire.models import async_session
|
|
from grimoire.models.moment_mapping import MomentMapping
|
|
from grimoire.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) == []
|