File size: 2,496 Bytes
0e76632 | 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 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | """Tests for /integration endpoints (Phase 3)."""
import pytest
@pytest.mark.asyncio
async def test_routing_rules(client, auth_headers):
resp = await client.get("/api/v1/integration/routing-rules", headers=auth_headers)
assert resp.status_code == 200
data = resp.json()
assert "rules" in data
assert isinstance(data["rules"], list)
assert len(data["rules"]) > 0 # Default rules exist
@pytest.mark.asyncio
async def test_routing_rules_unauthenticated(client):
resp = await client.get("/api/v1/integration/routing-rules")
# /integration/routing-rules has no auth dependency — public
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_update_routing_rules_requires_admin(client, auth_headers):
resp = await client.put("/api/v1/integration/routing-rules", headers=auth_headers,
json={"rules": []})
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_update_routing_rules_as_admin(client, admin_headers):
new_rules = [
{"task_type": "test", "target_model": "test:latest", "priority": 1}
]
resp = await client.put("/api/v1/integration/routing-rules", headers=admin_headers,
json={"rules": new_rules})
assert resp.status_code == 200
assert len(resp.json()["rules"]) == 1
# Verify
resp2 = await client.get("/api/v1/integration/routing-rules")
assert len(resp2.json()["rules"]) == 1
@pytest.mark.asyncio
async def test_workers_list(client, auth_headers):
resp = await client.get("/api/v1/integration/workers", headers=auth_headers)
assert resp.status_code == 200
data = resp.json()
assert "workers" in data
assert len(data["workers"]) >= 1
@pytest.mark.asyncio
async def test_worker_detail(client, auth_headers):
resp = await client.get("/api/v1/integration/workers/node-local", headers=auth_headers)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_worker_detail_not_found(client, auth_headers):
resp = await client.get("/api/v1/integration/workers/nonexistent", headers=auth_headers)
assert resp.status_code == 404
@pytest.mark.asyncio
async def test_queue_status(client, auth_headers):
resp = await client.get("/api/v1/integration/queue", headers=auth_headers)
assert resp.status_code == 200
data = resp.json()
assert "pending" in data
assert "processing" in data
|