141 lines
4.8 KiB
Python
141 lines
4.8 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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cannot_schedule_other_teams_tree(client: AsyncClient, auth_headers: dict, test_db):
|
|
"""User cannot create a schedule for a tree belonging to another team."""
|
|
import uuid as _uuid
|
|
from app.models.team import Team
|
|
from app.models.tree import Tree
|
|
|
|
# Create a tree belonging to a DIFFERENT team directly in DB
|
|
other_team = Team(name=f"Other Team {_uuid.uuid4()}")
|
|
test_db.add(other_team)
|
|
await test_db.flush()
|
|
|
|
other_tree = Tree(
|
|
name="Other Team Tree",
|
|
tree_type="maintenance",
|
|
team_id=other_team.id,
|
|
tree_structure={
|
|
"steps": [
|
|
{"id": "s1", "type": "procedure_step", "title": "Step",
|
|
"description": "Do it", "content_type": "action"},
|
|
{"id": "end", "type": "procedure_end", "title": "Done"},
|
|
]
|
|
},
|
|
status="published",
|
|
visibility="team",
|
|
)
|
|
test_db.add(other_tree)
|
|
await test_db.flush()
|
|
|
|
# Current user (from auth_headers) tries to schedule it
|
|
resp = await client.post(
|
|
"/api/v1/maintenance-schedules",
|
|
json={
|
|
"tree_id": str(other_tree.id),
|
|
"cron_expression": "0 9 1 * *",
|
|
"timezone": "UTC",
|
|
},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code in (403, 404) # either is acceptable
|