tests/test_cleanup_worker.py

from __future__ import annotations

import asyncio
import os
import time
from uuid import uuid4

from app.services.cleanup import CleanupService
from app.workers.cleanup_worker import CleanupWorker


async def test_cleanup_removes_expired_workspace(settings) -> None:
    service = CleanupService(settings)
    request_id = str(uuid4())
    workspace = await service.create_workspace(request_id)
    await service.complete(request_id)
    old = time.time() - 120
    os.utime(workspace.root, (old, old))
    removed = await service.cleanup_expired()
    assert removed == 1
    assert not workspace.root.exists()


async def test_cleanup_keeps_active_workspace(settings) -> None:
    service = CleanupService(settings)
    workspace = await service.create_workspace(str(uuid4()))
    old = time.time() - 120
    os.utime(workspace.root, (old, old))
    assert await service.cleanup_expired() == 0
    assert workspace.root.exists()


async def test_cleanup_worker_runs_and_stops() -> None:
    class FakeCleanup:
        def __init__(self) -> None:
            self.called = asyncio.Event()

        async def cleanup_expired(self) -> int:
            self.called.set()
            return 0

    service = FakeCleanup()
    worker = CleanupWorker(service, interval_seconds=60)  # type: ignore[arg-type]
    await worker.start()
    await asyncio.wait_for(service.called.wait(), timeout=1)
    await worker.stop()