Spaces:
Running
Running
File size: 1,413 Bytes
7cc81cb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 | 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()
|