Codex
63 lines
2.6 KiB
Python
63 lines
2.6 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from src.services.visual_watch.capture_retention_service import VisualCaptureRetentionService
|
|
|
|
|
|
class VisualCaptureRetentionServiceTests(unittest.TestCase):
|
|
def test_cleanup_deletes_old_html_and_json_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
old_html = root / "watch-old.html"
|
|
old_json = root / "watch-old.json"
|
|
new_html = root / "watch-new.html"
|
|
old_html.write_text("old", encoding="utf-8")
|
|
old_json.write_text("old-json", encoding="utf-8")
|
|
new_html.write_text("new", encoding="utf-8")
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(hours=200)).timestamp()
|
|
new_ts = (datetime.now(timezone.utc) - timedelta(hours=2)).timestamp()
|
|
os.utime(old_html, (old_ts, old_ts))
|
|
os.utime(old_json, (old_ts, old_ts))
|
|
os.utime(new_html, (new_ts, new_ts))
|
|
|
|
service = VisualCaptureRetentionService(output_root=root)
|
|
result = service.cleanup_captures(retention_hours=168, max_delete_files=10, dry_run=False)
|
|
|
|
self.assertEqual(result["status"], "completed")
|
|
self.assertEqual(result["eligible_file_count"], 2)
|
|
self.assertEqual(result["deleted_file_count"], 2)
|
|
self.assertFalse(old_html.exists())
|
|
self.assertFalse(old_json.exists())
|
|
self.assertTrue(new_html.exists())
|
|
|
|
def test_cleanup_dry_run_keeps_files(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
root = Path(tmpdir)
|
|
old_html = root / "watch-old.html"
|
|
old_html.write_text("old", encoding="utf-8")
|
|
old_ts = (datetime.now(timezone.utc) - timedelta(hours=200)).timestamp()
|
|
os.utime(old_html, (old_ts, old_ts))
|
|
|
|
service = VisualCaptureRetentionService(output_root=root)
|
|
result = service.cleanup_captures(retention_hours=168, dry_run=True)
|
|
|
|
self.assertEqual(result["deleted_file_count"], 1)
|
|
self.assertTrue(old_html.exists())
|
|
|
|
def test_cleanup_returns_idle_when_root_missing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
missing = Path(tmpdir) / "missing"
|
|
service = VisualCaptureRetentionService(output_root=missing)
|
|
|
|
result = service.cleanup_captures()
|
|
|
|
self.assertEqual(result["status"], "idle")
|
|
self.assertEqual(result["reason"], "capture_root_missing")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|