bvandeusenandClaude Opus 5.5
38b94bd57e
- common.detach: the hold-a-reference-until-done task start that mailer and unfurl_queue each wrote. - common.expired_before: the retention window, now shared by trash and the audit log (it moves out of retention.py, which imports audit). - responses.too_many: the 429 with Retry-After from the credential throttle and the client-download throttle. - share_sync.revoke_lost: revoke whoever could see a note before and no longer can, after a share or a group goes. - serialize.serialize_person: a member as the directory, a share and a group listing show them. - groups_api._get_group: the path-id lookup four group routes wrote. - settings.apply_session_ttl: the session lifetime set at boot and on save. Kept: the attachment-id claim check (one query; each caller answers an id it already holds differently), the strict UUID-list parses in reorder and set_note_labels (distinct error messages), and the checklist-items loops (one line each). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
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():
|
|
# 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
|