inkwell/tests/test_retention.py

from datetime import datetime, timedelta, timezone

import pytest

from inkwell.common import expired_before
from inkwell.retention import (
    SWEEP_BATCH,
    SWEEP_INTERVAL_SECONDS,
    SWEEP_STARTUP_DELAY_SECONDS,
    sweep_expired_trash,
)
from inkwell.settings import REGISTRY, get_public_config, validate_updates

NOW = datetime(2026, 7, 26, 12, 0, tzinfo=timezone.utc)


def test_expired_before_is_the_window_ago():
    assert expired_before(NOW, 30) == NOW - timedelta(days=30)
    assert expired_before(NOW, 1) == NOW - timedelta(days=1)


def test_zero_means_keep_forever():
    # The opt-out. A user who wants Trash to be an indefinite archive gets one, and
    # `None` is what stops the sweep before it builds a query at all.
    assert expired_before(NOW, 0) is None


def test_a_negative_window_also_means_never():
    # A stray minus in Settings. Read as "expired a day in the future", -1 would purge
    # the whole trash, so it means "never".
    assert expired_before(NOW, -1) is None
    assert expired_before(NOW, -3650) is None


async def test_sweep_is_a_noop_when_retention_is_off():
    # Passing None as the session proves it: retention off must return before it so
    # much as touches the database.
    assert await sweep_expired_trash(None, 0) == 0
    assert await sweep_expired_trash(None, -1) == 0


def test_retention_setting_is_registered_with_a_30_day_default():
    defn = next((d for d in REGISTRY if d.key == "trash_retention_days"), None)
    assert defn is not None, "the setting must appear in the admin Settings UI"
    assert defn.type == "int"
    assert defn.default == 30
    # An admin can tell what it does without reading the code.
    assert "0" in defn.description, "the keep-forever escape hatch must be documented"


def test_retention_setting_accepts_an_int_and_rejects_nonsense():
    clean, err = validate_updates({"trash_retention_days": "7"})
    assert err is None
    assert clean == {"trash_retention_days": 7}
    _, err = validate_updates({"trash_retention_days": "soon"})
    assert err is not None


async def test_public_config_publishes_the_window():
    # Clients need it to say how long a note has left in Trash, and a native client
    # reads it before it holds any credential — so it rides the unauthenticated
    # config. A stub session stands in for the DB: no row set => registry default.
    class _NoRows:
        async def get(self, *_args):
            return None

    cfg = await get_public_config(_NoRows())
    assert cfg["trash_retention_days"] == 30


@pytest.mark.parametrize(
    "value", [SWEEP_INTERVAL_SECONDS, SWEEP_STARTUP_DELAY_SECONDS, SWEEP_BATCH]
)
def test_sweeper_pacing_constants_are_positive(value):
    # A zero interval would turn the background loop into a busy spin against the DB.
    assert value > 0


def test_sweep_interval_is_well_under_a_day():
    # Retention is measured in days, but the sweep still has to run often enough that
    # "30 days" doesn't quietly become 31.
    assert SWEEP_INTERVAL_SECONDS <= 12 * 60 * 60