FabledGrimoire/tests/test_version_pinning_scan.py

"""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 grimoire.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 grimoire.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 grimoire.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 grimoire.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 grimoire.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 grimoire.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 grimoire.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 grimoire.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