58 lines
1.5 KiB
Python
58 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
|
|
from dataloader.api.v1.router import cancel_job, get_status
|
|
from dataloader.api.v1.schemas import JobStatusResponse
|
|
|
|
|
|
class _SvcOK:
|
|
async def status(self, job_id: UUID) -> JobStatusResponse | None:
|
|
return JobStatusResponse(
|
|
job_id=job_id,
|
|
status="queued",
|
|
attempt=0,
|
|
started_at=None,
|
|
finished_at=None,
|
|
heartbeat_at=None,
|
|
error=None,
|
|
progress={},
|
|
)
|
|
|
|
async def cancel(self, job_id: UUID) -> JobStatusResponse | None:
|
|
return JobStatusResponse(
|
|
job_id=job_id,
|
|
status="canceled",
|
|
attempt=1,
|
|
started_at=datetime.now(timezone.utc),
|
|
finished_at=datetime.now(timezone.utc),
|
|
heartbeat_at=None,
|
|
error="by test",
|
|
progress={},
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_router_get_status_returns_response():
|
|
svc = _SvcOK()
|
|
jid = uuid4()
|
|
res = await get_status(jid, svc=svc)
|
|
assert isinstance(res, JobStatusResponse)
|
|
assert res.job_id == jid
|
|
assert res.status == "queued"
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_router_cancel_returns_response():
|
|
svc = _SvcOK()
|
|
jid = uuid4()
|
|
res = await cancel_job(jid, svc=svc)
|
|
assert isinstance(res, JobStatusResponse)
|
|
assert res.job_id == jid
|
|
assert res.status == "canceled"
|