feat(versions): auto-pin scan promotes stable versions
_promote_stable_versions_for_note is the pure-function core: walks versions chronologically and pins any with a >= AUTO_PIN_STABILITY_DAYS (2-day) gap to the next version (or to now, for the latest). Auto- generated label describes the stability window. _scan_one_note loads versions for one note, runs the promotion, commits mutations to the attached rows, then calls prune_auto_pins to cap the auto bucket. scan_user_for_auto_pins fans out across the user's notes; scan_all_users_for_auto_pins is the top-level entrypoint for the cron. Per-note and per-user errors are caught and logged.
This commit is contained in:
@@ -0,0 +1,161 @@
|
||||
"""Tests for the auto-pin scan algorithm.
|
||||
|
||||
Tests the per-note promotion logic directly via a pure helper so the
|
||||
algorithm can be driven without standing up a real DB.
|
||||
"""
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from types import SimpleNamespace
|
||||
|
||||
|
||||
def _v(version_id: int, days_ago: int, pin_kind=None):
|
||||
"""Build a SimpleNamespace mimicking a NoteVersion row."""
|
||||
return SimpleNamespace(
|
||||
id=version_id,
|
||||
created_at=datetime.now(timezone.utc) - timedelta(days=days_ago),
|
||||
pin_kind=pin_kind,
|
||||
pin_label=None,
|
||||
)
|
||||
|
||||
|
||||
def test_promote_stable_versions_pins_versions_with_2_day_gap():
|
||||
"""Versions followed by another version >= 2 days later get promoted."""
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
# 3 versions: v1 (10 days ago), v2 (5 days ago), v3 (1 day ago)
|
||||
# Gaps: v1→v2 = 5d, v2→v3 = 4d. Both ≥ 2d → both pinned.
|
||||
# v3 (latest) has gap to now = 1d → NOT pinned (< 2 days).
|
||||
versions = [_v(1, 10), _v(2, 5), _v(3, 1)]
|
||||
pinned = _promote_stable_versions_for_note(versions)
|
||||
assert {v.id for v in pinned} == {1, 2}
|
||||
assert versions[0].pin_kind == "auto"
|
||||
assert versions[1].pin_kind == "auto"
|
||||
assert versions[2].pin_kind is None
|
||||
|
||||
|
||||
def test_promote_stable_versions_pins_latest_if_old_enough():
|
||||
"""If the latest version is >= 2 days old with no successor, pin it."""
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
versions = [_v(1, 10), _v(2, 3)] # latest is 3 days old
|
||||
pinned = _promote_stable_versions_for_note(versions)
|
||||
assert {v.id for v in pinned} == {1, 2}
|
||||
|
||||
|
||||
def test_promote_stable_versions_skips_already_pinned():
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
versions = [_v(1, 10, pin_kind="manual"), _v(2, 5)]
|
||||
pinned = _promote_stable_versions_for_note(versions)
|
||||
# v1 already manual — not re-pinned. v2 has 5-day gap to v1, then
|
||||
# latest-with-no-successor gap of 5d to now → pinned auto.
|
||||
assert {v.id for v in pinned} == {2}
|
||||
assert versions[0].pin_kind == "manual"
|
||||
assert versions[1].pin_kind == "auto"
|
||||
|
||||
|
||||
def test_promote_stable_versions_skips_active_editing():
|
||||
"""Versions a few hours apart with no >2-day gaps → no pins."""
|
||||
base = datetime.now(timezone.utc)
|
||||
v1 = SimpleNamespace(
|
||||
id=1, created_at=base - timedelta(hours=10),
|
||||
pin_kind=None, pin_label=None,
|
||||
)
|
||||
v2 = SimpleNamespace(
|
||||
id=2, created_at=base - timedelta(hours=2),
|
||||
pin_kind=None, pin_label=None,
|
||||
)
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
pinned = _promote_stable_versions_for_note([v1, v2])
|
||||
assert pinned == []
|
||||
|
||||
|
||||
def test_promote_stable_versions_writes_descriptive_label():
|
||||
"""The auto-generated label references stability."""
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
versions = [_v(1, 10), _v(2, 5), _v(3, 1)]
|
||||
_promote_stable_versions_for_note(versions)
|
||||
assert versions[0].pin_label is not None
|
||||
assert "stable" in versions[0].pin_label.lower()
|
||||
|
||||
|
||||
def test_promote_stable_versions_naive_datetime_is_treated_as_utc():
|
||||
"""created_at may come in as a naive datetime from some code paths;
|
||||
the algorithm coerces it to UTC rather than crashing on subtraction."""
|
||||
naive_old = datetime.utcnow() - timedelta(days=10)
|
||||
naive_recent = datetime.utcnow() - timedelta(days=1)
|
||||
versions = [
|
||||
SimpleNamespace(id=1, created_at=naive_old, pin_kind=None, pin_label=None),
|
||||
SimpleNamespace(id=2, created_at=naive_recent, pin_kind=None, pin_label=None),
|
||||
]
|
||||
from fabledassistant.services.version_pinning import (
|
||||
_promote_stable_versions_for_note,
|
||||
)
|
||||
pinned = _promote_stable_versions_for_note(versions)
|
||||
# v1 → v2 gap is 9 days → v1 pinned. v2 → now is 1 day → not pinned.
|
||||
assert {v.id for v in pinned} == {1}
|
||||
|
||||
|
||||
async def test_scan_user_for_auto_pins_iterates_all_notes(monkeypatch):
|
||||
"""scan_user_for_auto_pins iterates every note for the user and calls
|
||||
the per-note flow on each, summing returned counts."""
|
||||
from fabledassistant.services import version_pinning
|
||||
|
||||
note_ids = [10, 20, 30]
|
||||
seen: list[int] = []
|
||||
|
||||
async def fake_list_note_ids(uid):
|
||||
return note_ids
|
||||
|
||||
async def fake_one_note(uid, nid):
|
||||
seen.append(nid)
|
||||
# Pretend each note pinned one version.
|
||||
return 1
|
||||
|
||||
monkeypatch.setattr(
|
||||
version_pinning,
|
||||
"_list_user_note_ids_with_versions",
|
||||
fake_list_note_ids,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
version_pinning, "_scan_one_note", fake_one_note,
|
||||
)
|
||||
|
||||
total = await version_pinning.scan_user_for_auto_pins(user_id=1)
|
||||
|
||||
assert seen == note_ids
|
||||
assert total == 3
|
||||
|
||||
|
||||
async def test_scan_user_for_auto_pins_swallows_per_note_errors(monkeypatch):
|
||||
"""One bad note doesn't stop the scan; the error is logged and the
|
||||
others continue."""
|
||||
from fabledassistant.services import version_pinning
|
||||
|
||||
async def fake_list_note_ids(uid):
|
||||
return [10, 20, 30]
|
||||
|
||||
async def fake_one_note(uid, nid):
|
||||
if nid == 20:
|
||||
raise RuntimeError("simulated DB error")
|
||||
return 1
|
||||
|
||||
monkeypatch.setattr(
|
||||
version_pinning,
|
||||
"_list_user_note_ids_with_versions",
|
||||
fake_list_note_ids,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
version_pinning, "_scan_one_note", fake_one_note,
|
||||
)
|
||||
|
||||
total = await version_pinning.scan_user_for_auto_pins(user_id=1)
|
||||
|
||||
# 10 and 30 each contributed 1; 20 raised and contributed 0.
|
||||
assert total == 2
|
||||
Reference in New Issue
Block a user