91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
# tests/test_job.py
|
|
"""Tests for job queue (src/job.py)."""
|
|
|
|
import pytest
|
|
from src.job import JobQueue, Job, JobStatus
|
|
|
|
|
|
def test_job_initial_status():
|
|
"""New job starts in PENDING status."""
|
|
job = Job(
|
|
job_id="test-1",
|
|
video_path="/tmp/test.mp4",
|
|
model_configs=[],
|
|
output_dir="/tmp/output",
|
|
)
|
|
assert job.status == JobStatus.PENDING
|
|
|
|
|
|
def test_queue_add_job():
|
|
"""Adding a job returns the job with PENDING status."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
job = q.add_job(video_path="/tmp/test.mp4", model_configs=[])
|
|
assert job.job_id.startswith("job-")
|
|
assert job.status in (JobStatus.PENDING, JobStatus.RUNNING, JobStatus.COMPLETED)
|
|
|
|
|
|
def test_queue_get_job():
|
|
"""get_job returns the job by ID."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
job = q.add_job(video_path="/tmp/test.mp4", model_configs=[])
|
|
fetched = q.get_job(job.job_id)
|
|
assert fetched is not None
|
|
assert fetched.job_id == job.job_id
|
|
|
|
|
|
def test_queue_get_nonexistent():
|
|
"""get_job returns None for unknown ID."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
assert q.get_job("nope") is None
|
|
|
|
|
|
def test_queue_list_jobs():
|
|
"""list_jobs returns all jobs."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
q.add_job(video_path="/tmp/a.mp4", model_configs=[])
|
|
q.add_job(video_path="/tmp/b.mp4", model_configs=[])
|
|
jobs = q.list_jobs()
|
|
assert len(jobs) >= 2
|
|
|
|
|
|
def test_queue_cancel_pending():
|
|
"""Canceling a pending job sets status to CANCELLED."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
# Add job without starting (simulate by adding then immediately canceling)
|
|
# Since add_job starts a thread, we test cancel on a job we control
|
|
job = q.add_job(video_path="/nonexistent.mp4", model_configs=[])
|
|
# Wait briefly for thread to start
|
|
import time
|
|
time.sleep(0.1)
|
|
assert q.cancel_job(job.job_id) in (True, False) # may have already started
|
|
|
|
|
|
def test_job_empty_config_completes():
|
|
"""add_job with [] model_configs completes with COMPLETED + empty results."""
|
|
import time
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
job = q.add_job(video_path="/tmp/test.mp4", model_configs=[])
|
|
deadline = time.time() + 5.0
|
|
while time.time() < deadline:
|
|
fetched = q.get_job(job.job_id)
|
|
if fetched is not None and fetched.status == JobStatus.COMPLETED:
|
|
break
|
|
time.sleep(0.05)
|
|
fetched = q.get_job(job.job_id)
|
|
assert fetched is not None
|
|
assert fetched.status == JobStatus.COMPLETED
|
|
assert fetched.results == []
|
|
|
|
|
|
def test_queue_status_counts():
|
|
"""status_counts returns correct tally."""
|
|
q = JobQueue(output_dir="/tmp/output")
|
|
j1 = q.add_job(video_path="/nonexistent1.mp4", model_configs=[])
|
|
j2 = q.add_job(video_path="/nonexistent2.mp4", model_configs=[])
|
|
import time
|
|
time.sleep(0.5) # let them fail quickly
|
|
counts = q.status_counts()
|
|
assert isinstance(counts, dict)
|
|
# At least some count should be populated
|
|
assert sum(counts.values()) >= 2
|