64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
"""Tests for job state transitions & usage tracking."""
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from asf_domain import (
|
|
Job,
|
|
JobCreate,
|
|
JobUpdate,
|
|
JobStatus,
|
|
JobPriority,
|
|
JobType,
|
|
JobUsage,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 9, 7, 12, 0, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_job_create_defaults() -> None:
|
|
jc = JobCreate(
|
|
type=JobType.AI_INTERACTIVE,
|
|
payload={"prompt": "summarize REQ-1", "artifact_id": str(uuid4())},
|
|
)
|
|
assert jc.priority == JobPriority.NORMAL
|
|
assert jc.model is None
|
|
|
|
|
|
def test_job_lifecycle_updates() -> None:
|
|
tenant_id = uuid4()
|
|
j = Job(
|
|
tenant_id=tenant_id,
|
|
type=JobType.AI_BACKGROUND,
|
|
payload={"artifact_id": str(uuid4())},
|
|
created_at=NOW,
|
|
)
|
|
assert j.status == JobStatus.QUEUED
|
|
|
|
# worker claims it
|
|
update1 = JobUpdate(status=JobStatus.RUNNING, worker="worker-asf-ai-7", started_at=NOW)
|
|
data = j.model_copy(update=update1.model_dump(exclude_unset=True))
|
|
assert data.status == JobStatus.RUNNING
|
|
assert data.worker == "worker-asf-ai-7"
|
|
|
|
# worker completes with usage
|
|
usage = JobUsage(
|
|
input_tokens=1500, output_tokens=400, gpu_seconds=2.3, estimated_cost=0.012
|
|
)
|
|
update2 = JobUpdate(status=JobStatus.COMPLETED, completed_at=NOW, usage=usage)
|
|
final = data.model_copy(update=update2.model_dump(exclude_unset=True))
|
|
assert final.status == JobStatus.COMPLETED
|
|
final_usage = JobUsage.model_validate(final.usage)
|
|
assert final_usage.output_tokens == 400
|
|
assert final_usage.estimated_cost == pytest.approx(0.012)
|
|
|
|
|
|
def test_job_update_validates_progress_bounds() -> None:
|
|
from pydantic import ValidationError
|
|
|
|
with pytest.raises(ValidationError):
|
|
JobUpdate(progress_percent=150)
|