Files
the-nexus/tests/test_mempalace_retain_closets.py
Alexander Whitestone e644b00dff
Some checks failed
CI / test (pull_request) Failing after 7s
CI / validate (pull_request) Failing after 3s
Review Approval Gate / verify-review (pull_request) Failing after 4s
feat(mempalace): retention enforcement + tunnel sync client (#1083, #1078)
**retain_closets.py** — 90-day closet aging enforcement for #1083.
Removes *.closet.json files older than --days (default 90) from the
fleet palace. Supports --dry-run for safe preview. Wired into the
weekly-audit workflow as a dry-run CI step; production cron guidance
added to workflow comments.

**tunnel_sync.py** — remote wizard wing pull client for #1078.
Connects to a peer's fleet_api.py HTTP endpoint, discovers wings via
/wings, and pulls core rooms via /search into local *.closet.json
files. Zero new dependencies (stdlib urllib only). Supports --dry-run.
This is the code side of the inter-wizard tunnel; infrastructure
(second wizard VPS + fleet_api.py running) still required.

**Tests:** 29 new tests, all passing. Total suite: 294 passing.

Refs #1075, #1078, #1083
2026-04-07 11:05:00 -04:00

140 lines
4.3 KiB
Python

"""
Tests for mempalace/retain_closets.py — 90-day closet retention enforcement.
Refs: #1083, #1075
"""
from __future__ import annotations
import json
import time
from pathlib import Path
import pytest
from mempalace.retain_closets import (
RetentionResult,
_file_age_days,
enforce_retention,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _write_closet(directory: Path, name: str, age_days: float) -> Path:
"""Create a *.closet.json file with a mtime set to *age_days* ago."""
p = directory / name
p.write_text(json.dumps({"drawers": [{"text": "summary", "closet": True}]}))
# Set mtime to simulate age
mtime = time.time() - age_days * 86400.0
import os
os.utime(p, (mtime, mtime))
return p
# ---------------------------------------------------------------------------
# _file_age_days
# ---------------------------------------------------------------------------
def test_file_age_days_recent(tmp_path):
p = tmp_path / "recent.closet.json"
p.write_text("{}")
age = _file_age_days(p)
assert 0 <= age < 1 # just created
def test_file_age_days_old(tmp_path):
p = _write_closet(tmp_path, "old.closet.json", age_days=100)
age = _file_age_days(p)
assert 99 < age < 101
# ---------------------------------------------------------------------------
# enforce_retention — dry_run
# ---------------------------------------------------------------------------
def test_dry_run_does_not_delete(tmp_path):
old = _write_closet(tmp_path, "old.closet.json", age_days=100)
_write_closet(tmp_path, "new.closet.json", age_days=10)
result = enforce_retention(tmp_path, retention_days=90, dry_run=True)
# File still exists after dry-run
assert old.exists()
assert result.removed == 1 # counted but not actually removed
assert result.kept == 1
assert result.ok
def test_dry_run_keeps_recent_files(tmp_path):
_write_closet(tmp_path, "recent.closet.json", age_days=5)
result = enforce_retention(tmp_path, retention_days=90, dry_run=True)
assert result.removed == 0
assert result.kept == 1
# ---------------------------------------------------------------------------
# enforce_retention — live mode
# ---------------------------------------------------------------------------
def test_live_removes_old_closets(tmp_path):
old = _write_closet(tmp_path, "old.closet.json", age_days=100)
new = _write_closet(tmp_path, "new.closet.json", age_days=10)
result = enforce_retention(tmp_path, retention_days=90, dry_run=False)
assert not old.exists()
assert new.exists()
assert result.removed == 1
assert result.kept == 1
assert result.ok
def test_live_keeps_files_within_window(tmp_path):
f = _write_closet(tmp_path, "edge.closet.json", age_days=89)
result = enforce_retention(tmp_path, retention_days=90, dry_run=False)
assert f.exists()
assert result.removed == 0
assert result.kept == 1
def test_empty_directory_is_ok(tmp_path):
result = enforce_retention(tmp_path, retention_days=90)
assert result.scanned == 0
assert result.removed == 0
assert result.ok
def test_subdirectory_closets_are_pruned(tmp_path):
"""enforce_retention should recurse into subdirs (wing directories)."""
sub = tmp_path / "bezalel"
sub.mkdir()
old = _write_closet(sub, "hermes.closet.json", age_days=120)
result = enforce_retention(tmp_path, retention_days=90, dry_run=False)
assert not old.exists()
assert result.removed == 1
def test_non_closet_files_ignored(tmp_path):
"""Non-closet files should not be counted or touched."""
(tmp_path / "readme.txt").write_text("hello")
(tmp_path / "data.drawer.json").write_text("{}")
result = enforce_retention(tmp_path, retention_days=90)
assert result.scanned == 0
# ---------------------------------------------------------------------------
# RetentionResult.ok
# ---------------------------------------------------------------------------
def test_retention_result_ok_with_no_errors():
r = RetentionResult(scanned=5, removed=2, kept=3)
assert r.ok is True
def test_retention_result_not_ok_with_errors():
r = RetentionResult(errors=["could not stat file"])
assert r.ok is False