110 lines
3.0 KiB
Python
110 lines
3.0 KiB
Python
"""Integration tests for Flask web app."""
|
|
|
|
import io
|
|
|
|
import pytest
|
|
from app import app
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
app.config["TESTING"] = True
|
|
with app.test_client() as client:
|
|
yield client
|
|
|
|
|
|
def test_index_page(client):
|
|
"""GET / returns 200."""
|
|
resp = client.get("/")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_jobs_page(client):
|
|
"""GET /jobs returns 200."""
|
|
resp = client.get("/jobs")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
def test_api_models(client):
|
|
"""GET /api/models returns JSON list."""
|
|
resp = client.get("/api/models")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert isinstance(data, list)
|
|
|
|
|
|
def test_api_jobs(client):
|
|
"""GET /api/jobs returns JSON list."""
|
|
resp = client.get("/api/jobs")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert isinstance(data, list)
|
|
|
|
|
|
def test_upload_no_video(client):
|
|
"""POST /upload without video returns 400."""
|
|
resp = client.post("/upload")
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_upload_invalid_extension_rejected(client):
|
|
"""POST /upload with a non-video file returns 400."""
|
|
resp = client.post(
|
|
"/upload",
|
|
data={"video": (io.BytesIO(b"not a video"), "notes.txt")},
|
|
content_type="multipart/form-data",
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
def test_status_nonexistent(client):
|
|
"""GET /status/nonexistent returns 404."""
|
|
resp = client.get("/status/nonexistent")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_api_job_detail_nonexistent(client):
|
|
"""GET /api/jobs/nonexistent returns 404."""
|
|
resp = client.get("/api/jobs/nonexistent")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
def test_cancel_nonexistent_job(client):
|
|
"""POST /api/jobs/nonexistent/cancel returns 404."""
|
|
resp = client.post("/api/jobs/nonexistent/cancel")
|
|
assert resp.status_code == 404
|
|
data = resp.get_json()
|
|
assert data["error"] == "not found"
|
|
|
|
|
|
def test_cancel_completed_job(client):
|
|
"""POST /api/jobs/<id>/cancel returns 400 for completed job."""
|
|
from src.job import JobQueue, Job, JobStatus
|
|
from app import job_queue
|
|
job = job_queue.add_job(video_path="/tmp/test.mp4", model_configs=[])
|
|
import time
|
|
time.sleep(0.3)
|
|
# Force status to COMPLETED
|
|
with job_queue._lock:
|
|
job.status = JobStatus.COMPLETED
|
|
resp = client.post(f"/api/jobs/{job.job_id}/cancel")
|
|
assert resp.status_code == 400
|
|
data = resp.get_json()
|
|
assert "cannot cancel" in data["error"]
|
|
|
|
|
|
def test_cancel_pending_or_running_job(client):
|
|
"""POST /api/jobs/<id>/cancel returns 200 for cancellable job."""
|
|
from src.job import JobQueue, JobStatus
|
|
from app import job_queue
|
|
job = job_queue.add_job(video_path="/tmp/test.mp4", model_configs=[])
|
|
import time
|
|
time.sleep(0.1)
|
|
resp = client.post(f"/api/jobs/{job.job_id}/cancel")
|
|
assert resp.status_code == 200
|
|
data = resp.get_json()
|
|
assert data["status"] == "cancelled"
|
|
# Verify job is now CANCELLED
|
|
fetched = job_queue.get_job(job.job_id)
|
|
assert fetched.status == JobStatus.CANCELLED
|