99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
"""Tests for maintenance schedule CRUD."""
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
async def _create_maintenance_tree(client, headers):
|
|
resp = await client.post(
|
|
"/api/v1/trees",
|
|
json={
|
|
"name": "Scheduled Patch",
|
|
"tree_type": "maintenance",
|
|
"tree_structure": {
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step",
|
|
"description": "Do it", "content_type": "action"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
},
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()["id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_schedule(client: AsyncClient, auth_headers: dict):
|
|
tree_id = await _create_maintenance_tree(client, auth_headers)
|
|
resp = await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={
|
|
"tree_id": tree_id,
|
|
"cron_expression": "0 9 15 * *",
|
|
"timezone": "America/New_York",
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
data = resp.json()
|
|
assert data["cron_expression"] == "0 9 15 * *"
|
|
assert data["timezone"] == "America/New_York"
|
|
assert data["is_active"] is True
|
|
assert data["next_run_at"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_schedule_rejected(client: AsyncClient, auth_headers: dict):
|
|
"""Cannot create two schedules for the same tree."""
|
|
tree_id = await _create_maintenance_tree(client, auth_headers)
|
|
await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={"tree_id": tree_id, "cron_expression": "0 0 1 * *", "timezone": "UTC"},
|
|
headers=auth_headers,
|
|
)
|
|
resp = await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={"tree_id": tree_id, "cron_expression": "0 6 1 * *", "timezone": "UTC"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schedule_for_tree(client: AsyncClient, auth_headers: dict):
|
|
tree_id = await _create_maintenance_tree(client, auth_headers)
|
|
await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={"tree_id": tree_id, "cron_expression": "0 0 1 * *", "timezone": "UTC"},
|
|
headers=auth_headers,
|
|
)
|
|
resp = await client.get(f"/api/v1/maintenance-schedules/tree/{tree_id}", headers=auth_headers)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["cron_expression"] == "0 0 1 * *"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_schedule(client: AsyncClient, auth_headers: dict):
|
|
tree_id = await _create_maintenance_tree(client, auth_headers)
|
|
create = await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={"tree_id": tree_id, "cron_expression": "0 6 * * 1", "timezone": "UTC"},
|
|
headers=auth_headers,
|
|
)
|
|
sched_id = create.json()["id"]
|
|
resp = await client.patch(
|
|
f"/api/v1/maintenance-schedules/{sched_id}",
|
|
json={"is_active": False},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["is_active"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schedule_not_found(client: AsyncClient, auth_headers: dict):
|
|
tree_id = await _create_maintenance_tree(client, auth_headers)
|
|
resp = await client.get(f"/api/v1/maintenance-schedules/tree/{tree_id}", headers=auth_headers)
|
|
assert resp.status_code == 404
|