backend/tests/test_foreign_tracks_sweep_stats.py
Rama master · réplica de lectura del repositorio original.
"""Foreign-track sweep: cleanup_history rows, stats and the one-off backfill.
The owner asked (2026-09-26) whether the statistics page records how much the
sweep freed — it did not: every slice's result was only logged. These tests
pin down the history row per slice, the stats aggregate built from it, and
the backfill for strips that happened before the row existed.
"""
import json
import logging
import os
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
import pytest
from services.foreign_tracks import sweep as sw
ACTION = "scheduled_foreign_track_sweep"
INT4_MAX = 2_147_483_647
def _result(**overrides):
base = {
"phase": "strip",
"probed": 3,
"stripped_files": 2,
"tracks_removed": 5,
"bytes_freed": 1234,
"pending": 0,
"affected": 4,
"paused_reason": None,
}
base.update(overrides)
return base
def _history(action=ACTION):
from sqlalchemy import select
from db.models.cleanup import CleanupHistory
from extensions import db
return list(
db.session.execute(
select(CleanupHistory)
.where(CleanupHistory.action_type == action)
.order_by(CleanupHistory.id)
).scalars()
)
@pytest.fixture
def tick_env(app_ctx, monkeypatch, tmp_path):
"""Drive foreign_track_sweep_tick with a canned slice result."""
monkeypatch.setattr(
"config.get_settings",
lambda: SimpleNamespace(
foreign_track_sweep_enabled=True,
foreign_track_sweep_budget_s=60,
media_path=str(tmp_path),
),
)
monkeypatch.setattr(sw, "_find_rule", lambda repo: {"id": 7, "config_json": {}})
holder = {"result": _result()}
monkeypatch.setattr(sw, "run_slice", lambda *a, **k: dict(holder["result"]))
return holder
# ── 1. one history row per slice that stripped something ─────────────────────
def test_tick_writes_one_history_row_when_files_were_stripped(tick_env):
sw.foreign_track_sweep_tick()
rows = _history()
assert len(rows) == 1
row = rows[0]
assert row.rule_id == 7
assert row.files_processed == 2
assert row.files_deleted == 0
assert row.bytes_freed == 1234
assert json.loads(row.details_json) == {
"tracks_removed": 5,
"probed": 3,
"phase": "strip",
"backups_deleted": 0,
"verify_failed": 0,
}
def test_tick_writes_nothing_when_no_file_was_stripped(tick_env):
tick_env["result"] = _result(stripped_files=0, tracks_removed=0, bytes_freed=0)
sw.foreign_track_sweep_tick()
assert _history() == []
def test_failing_history_write_never_fails_the_tick(tick_env, monkeypatch, caplog):
from db.repositories.cleanup import CleanupRepository
def boom(self, *a, **k):
raise RuntimeError("db gone")
monkeypatch.setattr(CleanupRepository, "log_cleanup", boom)
caplog.set_level(logging.WARNING, logger="services.foreign_tracks.stats")
sw.foreign_track_sweep_tick() # must not raise
assert any("db gone" in r.getMessage() for r in caplog.records)
def test_slice_larger_than_int4_is_split_across_rows(tick_env):
"""cleanup_history.bytes_freed is INTEGER — int4 on Postgres."""
big = INT4_MAX * 2 + 10
tick_env["result"] = _result(bytes_freed=big)
sw.foreign_track_sweep_tick()
rows = _history()
assert len(rows) == 3
assert all(r.bytes_freed <= INT4_MAX for r in rows)
assert sum(r.bytes_freed for r in rows) == big
# Files and tracks are counted once, on the first row only.
assert [r.files_processed for r in rows] == [2, 0, 0]
assert sum(json.loads(r.details_json).get("tracks_removed", 0) for r in rows) == 5
# ── 2. stats aggregate + endpoint ─────────────────────────────────────────────
def _seed_scan(path, state, size=100, tracks=0, processed_at=None):
from db.models.foreign_tracks import ForeignTrackScan
from extensions import db
db.session.add(
ForeignTrackScan(
path=path,
size_bytes=size,
mtime=0.0,
state=state,
track_count=tracks,
processed_at=processed_at,
)
)
db.session.commit()
def _seed_history(bytes_freed, files, tracks, days_ago=0, details=None):
from db.models.cleanup import CleanupHistory
from extensions import db
db.session.add(
CleanupHistory(
rule_id=1,
action_type=ACTION,
files_processed=files,
files_deleted=0,
bytes_freed=bytes_freed,
details_json=json.dumps(details or {"tracks_removed": tracks}),
performed_at=datetime.now(UTC) - timedelta(days=days_ago),
)
)
db.session.commit()
def test_sweep_stats_aggregates_history_scan_counts_and_state(app_ctx):
from services.foreign_tracks.state import SweepState, save_state
from services.foreign_tracks.stats import get_sweep_stats
_seed_history(1000, files=2, tracks=4, days_ago=1)
_seed_history(500, files=1, tracks=3, days_ago=40)
_seed_scan("/m/a.mkv", "stripped")
_seed_scan("/m/b.mkv", "clean")
_seed_scan("/m/c.mkv", "clean")
_seed_scan("/m/d.mkv", "affected")
save_state(SweepState(phase="strip", paused_reason="disk floor reached"))
stats = get_sweep_stats()
assert stats["files_stripped"] == 3
assert stats["tracks_removed"] == 7
assert stats["bytes_freed_total"] == 1500
assert stats["bytes_freed_30d"] == 1000
assert stats["scan_counts"] == {
"pending": 0,
"clean": 2,
"affected": 1,
"stripped": 1,
"failed": 0,
}
assert stats["phase"] == "strip"
assert stats["paused_reason"] == "disk floor reached"
def test_sweep_stats_empty_database(app_ctx):
from services.foreign_tracks.stats import get_sweep_stats
stats = get_sweep_stats()
assert stats["files_stripped"] == 0
assert stats["tracks_removed"] == 0
assert stats["bytes_freed_total"] == 0
assert stats["bytes_freed_30d"] == 0
assert stats["phase"] == "idle"
assert stats["paused_reason"] is None
def test_sweep_stats_endpoint(client):
with client.application.app_context():
_seed_history(2048, files=1, tracks=2)
resp = client.get("/api/v1/statistics/foreign-tracks")
assert resp.status_code == 200
data = resp.get_json()
assert data["bytes_freed_total"] == 2048
assert data["files_stripped"] == 1
assert data["tracks_removed"] == 2
assert set(data["scan_counts"]) == {"pending", "clean", "affected", "stripped", "failed"}
assert "phase" in data and "paused_reason" in data
# ── 3. backfill for strips made before the history row existed ────────────────
def _file(tmp_path, name, size):
path = tmp_path / name
path.write_bytes(b"x" * size)
return str(path)
def test_backfill_sums_size_delta_of_stripped_rows(app_ctx, tmp_path):
from services.foreign_tracks.stats import backfill_sweep_history
old = datetime.now(UTC) - timedelta(days=3)
_seed_scan(_file(tmp_path, "a.mkv", 600), "stripped", size=1000, tracks=2, processed_at=old)
_seed_scan(_file(tmp_path, "b.mkv", 300), "stripped", size=500, tracks=3, processed_at=old)
# Grew after the strip (replaced file): clamped to 0, still counted.
_seed_scan(_file(tmp_path, "c.mkv", 900), "stripped", size=100, tracks=1, processed_at=old)
# Missing file: skipped entirely.
_seed_scan(str(tmp_path / "gone.mkv"), "stripped", size=999, tracks=9, processed_at=old)
# Not stripped: ignored.
_seed_scan(_file(tmp_path, "d.mkv", 10), "clean", size=5000, processed_at=old)
summary = backfill_sweep_history(dry_run=False)
assert summary["written"] is True
assert summary["bytes_freed"] == 400 + 200
rows = _history()
assert len(rows) == 1
assert rows[0].bytes_freed == 600
assert rows[0].files_processed == 3
details = json.loads(rows[0].details_json)
assert details == {"backfill": True, "files": 3, "tracks_removed": 6}
def test_backfill_dry_run_writes_nothing(app_ctx, tmp_path):
from services.foreign_tracks.stats import backfill_sweep_history
old = datetime.now(UTC) - timedelta(days=3)
_seed_scan(_file(tmp_path, "a.mkv", 600), "stripped", size=1000, tracks=2, processed_at=old)
summary = backfill_sweep_history(dry_run=True)
assert summary["written"] is False
assert summary["bytes_freed"] == 400
assert _history() == []
def test_backfill_is_idempotent(app_ctx, tmp_path):
from services.foreign_tracks.stats import backfill_sweep_history
old = datetime.now(UTC) - timedelta(days=3)
_seed_scan(_file(tmp_path, "a.mkv", 600), "stripped", size=1000, tracks=2, processed_at=old)
backfill_sweep_history(dry_run=False)
second = backfill_sweep_history(dry_run=False)
assert second["written"] is False
assert second["reason"] == "already backfilled"
assert len(_history()) == 1
def test_backfill_only_counts_strips_older_than_the_first_history_row(app_ctx, tmp_path):
from services.foreign_tracks.stats import backfill_sweep_history
_seed_history(50, files=1, tracks=1, days_ago=2)
before = datetime.now(UTC) - timedelta(days=5)
after = datetime.now(UTC) - timedelta(hours=1)
_seed_scan(_file(tmp_path, "a.mkv", 600), "stripped", size=1000, tracks=2, processed_at=before)
_seed_scan(_file(tmp_path, "b.mkv", 100), "stripped", size=900, tracks=7, processed_at=after)
summary = backfill_sweep_history(dry_run=False)
assert summary["files"] == 1
assert summary["bytes_freed"] == 400
assert summary["tracks_removed"] == 2
def test_backfill_larger_than_int4_splits_but_stays_one_backfill(app_ctx, tmp_path, monkeypatch):
from services.foreign_tracks import stats as st
old = datetime.now(UTC) - timedelta(days=3)
path = _file(tmp_path, "a.mkv", 10)
_seed_scan(path, "stripped", size=INT4_MAX + 10 + 10, tracks=4, processed_at=old)
st.backfill_sweep_history(dry_run=False)
again = st.backfill_sweep_history(dry_run=False)
rows = _history()
assert len(rows) == 2
assert sum(r.bytes_freed for r in rows) == INT4_MAX + 10
assert all(json.loads(r.details_json).get("backfill") is True for r in rows)
assert again["written"] is False
assert os.path.exists(path)