70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from app.config import settings
|
|
from app.db import Database
|
|
from app.scheduler import JobRunner
|
|
|
|
|
|
def test_maintenance_removes_archives_after_two_days(tmp_path) -> None:
|
|
test_settings = replace(
|
|
settings,
|
|
data_dir=tmp_path,
|
|
database_path=tmp_path / "siteharbor.sqlite3",
|
|
work_dir=tmp_path / "work",
|
|
artifacts_dir=tmp_path / "artifacts",
|
|
reports_dir=tmp_path / "reports",
|
|
worker_enabled=False,
|
|
)
|
|
test_settings.ensure_directories()
|
|
database = Database(test_settings.database_path)
|
|
database.initialize()
|
|
job = database.create_job(
|
|
"https://example.com",
|
|
"https://example.com/",
|
|
{
|
|
"include_external_assets": True,
|
|
"max_pages": 20,
|
|
"max_depth": 3,
|
|
"max_bytes": 1_000_000,
|
|
"max_duration_seconds": 60,
|
|
"retention_days": 2,
|
|
},
|
|
)
|
|
assert database.claim_next("worker-a", lease_seconds=30)
|
|
|
|
archive_path = test_settings.artifacts_dir / "capture.zip"
|
|
report_path = test_settings.reports_dir / "capture.json"
|
|
archive_path.write_bytes(b"zip contents")
|
|
report_path.write_text("{}", encoding="utf-8")
|
|
assert database.mark_ready(
|
|
job["id"],
|
|
"worker-a",
|
|
artifact_path=str(archive_path),
|
|
report_path=str(report_path),
|
|
archive_size=archive_path.stat().st_size,
|
|
entry_point="example.com/index.html",
|
|
stats={},
|
|
warnings=[],
|
|
retention_days=2,
|
|
)
|
|
with database.connection() as connection:
|
|
connection.execute(
|
|
"""
|
|
UPDATE jobs
|
|
SET completed_at = '2000-01-01T00:00:00+00:00',
|
|
expires_at = '2100-01-01T00:00:00+00:00'
|
|
WHERE id = ?
|
|
""",
|
|
(job["id"],),
|
|
)
|
|
|
|
JobRunner(database, test_settings)._expire_artifacts_if_due()
|
|
|
|
expired = database.get_job(job["id"])
|
|
assert expired is not None
|
|
assert expired["state"] == "expired"
|
|
assert not archive_path.exists()
|
|
assert not report_path.exists()
|