resolutionflow/backend/tests/test_maintenance_schedules.py

"""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_create_schedule_for_non_maintenance_tree(
    client: AsyncClient, auth_headers: dict, test_tree: dict
):
    """Schedules are restricted to maintenance flows."""
    resp = await client.post(
        "/api/v1/maintenance-schedules",
        json={
            "tree_id": test_tree["id"],
            "cron_expression": "0 0 1 * *",
            "timezone": "UTC",
        },
        headers=auth_headers,
    )
    assert resp.status_code == 400


@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()

    from uuid import UUID as _UUID
    other_tree = Tree(
        name="Other Team Tree",
        tree_type="maintenance",
        team_id=other_team.id,
        account_id=_UUID("00000000-0000-0000-0000-000000000001"),
        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