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>
79 lines
2.8 KiB
Python
79 lines
2.8 KiB
Python
"""Tests for the prune logic — both rolling (pin_kind IS NULL) and the
|
|
auto-pin bucket (pin_kind='auto').
|
|
"""
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from tests.helpers import make_mock_session
|
|
|
|
|
|
async def test_create_version_prune_sql_filters_to_unpinned():
|
|
"""The DELETE statement issued by create_version's prune step must
|
|
include `pin_kind IS NULL` in the inner SELECT so pinned versions
|
|
aren't counted toward MAX_VERSIONS and can't be pruned by it."""
|
|
mock_session = make_mock_session()
|
|
select_result = MagicMock()
|
|
# No prior version → skips the throttle/dedupe early-return paths and
|
|
# proceeds straight to insert + prune.
|
|
select_result.scalars.return_value.first.return_value = None
|
|
mock_session.add = MagicMock()
|
|
mock_session.commit = AsyncMock()
|
|
mock_session.refresh = AsyncMock()
|
|
|
|
captured_sql: list[str] = []
|
|
|
|
async def execute_capture(stmt, *args, **kwargs):
|
|
captured_sql.append(str(stmt))
|
|
if len(captured_sql) == 1:
|
|
return select_result
|
|
return MagicMock()
|
|
|
|
mock_session.execute = AsyncMock(side_effect=execute_capture)
|
|
|
|
with patch(
|
|
"grimoire.services.note_versions.async_session",
|
|
return_value=mock_session,
|
|
):
|
|
from grimoire.services.note_versions import create_version
|
|
await create_version(
|
|
user_id=1, note_id=42, body="content", title="t", tags=["a"],
|
|
)
|
|
|
|
assert any("pin_kind IS NULL" in s for s in captured_sql), (
|
|
f"Expected prune SQL to filter pin_kind IS NULL; got: {captured_sql!r}"
|
|
)
|
|
|
|
|
|
async def test_prune_auto_pins_filters_to_auto_kind():
|
|
"""prune_auto_pins must filter to pin_kind='auto' so manual pins and
|
|
rolling rows aren't touched, and must bind MAX_AUTO_PINS as the OFFSET."""
|
|
mock_session = make_mock_session()
|
|
mock_session.commit = AsyncMock()
|
|
|
|
captured_sql: list[str] = []
|
|
captured_params: list[dict] = []
|
|
|
|
async def execute_capture(stmt, *args, **kwargs):
|
|
captured_sql.append(str(stmt))
|
|
try:
|
|
captured_params.append(dict(stmt.compile().params))
|
|
except Exception:
|
|
captured_params.append({})
|
|
return MagicMock()
|
|
|
|
mock_session.execute = AsyncMock(side_effect=execute_capture)
|
|
|
|
with patch(
|
|
"grimoire.services.version_pinning.async_session",
|
|
return_value=mock_session,
|
|
):
|
|
from grimoire.services.version_pinning import (
|
|
prune_auto_pins, MAX_AUTO_PINS,
|
|
)
|
|
await prune_auto_pins(user_id=1, note_id=42)
|
|
|
|
assert any("pin_kind = 'auto'" in s for s in captured_sql), (
|
|
f"expected auto-bucket filter in SQL; got: {captured_sql!r}"
|
|
)
|
|
assert any(
|
|
p.get("max_auto_pins") == MAX_AUTO_PINS for p in captured_params
|
|
), f"bound params: {captured_params!r}"
|