Files
inkwell/tests/test_retention.py
bvandeusenandClaude Opus 5.5 a706644455
CI & Build / Python lint (push) Successful in 2s
CI & Build / Build now, or wait for Android? (push) Successful in 2s
Android / Build, or is the channel already serving this? (push) Successful in 3s
Desktop (Tauri) / Build, or is the channel already serving this? (push) Successful in 2s
CI & Build / TypeScript typecheck (push) Successful in 7s
CI & Build / Python tests (push) Successful in 15s
CI & Build / integration (push) Successful in 45s
CI & Build / Build & push image (push) Skipped
Desktop (Tauri) / Windows installer (cross-compiled) (push) Successful in 4m17s
Desktop (Tauri) / Tauri desktop (Linux) (push) Successful in 7m33s
Desktop (Tauri) / Update manifest (push) Successful in 7s
Android / Kotlin + Rust (APK) (push) Successful in 11m25s
rename: the server is Inkwell — package, env vars, image, compose, export marker
Step 2 of milestone 481. The operator chose a full rename (Scribe note 5071), so
this goes past the display strings into the identities:

- src/thoughtsync → src/inkwell; every import, the Dockerfile and both compose
  commands, alembic env, pyproject
- THOUGHTSYNC_* → INKWELL_* (database URL, secret key, log level, tag/port/bind)
- container data dir /var/thoughtsync → /var/inkwell
- image git.fabledsword.com/bvandeusen/inkwell; Postgres user/db default inkwell;
  CI's integration service follows
- the files the image serves are inkwell.*. fetch-clients.sh still fetches the
  thoughtsync-named release assets, because the lanes that publish them are
  renamed in steps 3 and 4
- exports are written with app "inkwell"

Two deliberate exceptions, both because data rides on them:

- compose volumes are now named explicitly and overridable (INKWELL_DB_VOLUME,
  INKWELL_DATA_VOLUME), so a deployment installed as ThoughtSync points at the
  volumes and DB identity it already has. .env.example says exactly what to set
- import still accepts app "thoughtsync", because exports written before the
  rename are backups. Tested both ways

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 14:18:49 -04:00

84 lines
3.1 KiB
Python

from datetime import datetime, timedelta, timezone
import pytest
from inkwell.retention import (
SWEEP_BATCH,
SWEEP_INTERVAL_SECONDS,
SWEEP_STARTUP_DELAY_SECONDS,
expired_before,
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():
# Reachable by typing a stray minus into the Settings field. The dangerous reading
# of -1 would be "expired a day in the FUTURE", which purges the entire trash on
# the next sweep; refusing to run is the only safe interpretation.
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
# The operator has to be able to 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