"""Tests for the Wahab demo-prep hardening batch. Anchors: * ``Cancelled`` is a terminal status reachable from every active state, is excluded from SLA breach reporting, and shows up in the status pickers. * Customer ``phone`` collected on the new-issue form is persisted (it used to be silently dropped). * ``GET /api/tickets/{id}`` returns nested ``unit``/``category`` objects so the detail page stops rendering "—" for Unit/Property/Category. * ``DELETE /api/tickets/{id}`` is admin-only and removes ticket children. * ``GET /api/auth/users`` powers the assign-technician dropdown. * Ticket numbering uses max+1, so deleting tickets never re-issues a number. """ from __future__ import annotations import pytest pytestmark = pytest.mark.asyncio async def _login(client, email="wahab@denya.com", password="denya123") -> str: resp = await client.post( "/api/auth/login", json={"email": email, "password": password}, ) assert resp.status_code == 200, resp.text return resp.json()["access_token"] async def _create_ticket(client, token: str, **overrides) -> dict: payload = { "unit_id": 2, "category_id": 3, "priority": "medium", "reporter": "Demo Prep Test", "reported_via": "walk-in", "description": "hardening batch test ticket", **overrides, } resp = await client.post( "/api/tickets", json=payload, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 201, resp.text return resp.json() # ── Cancelled status ────────────────────────────────────────────────── async def test_cancelled_is_terminal(client): """A cancelled ticket has no valid next status.""" from app.services.ticket import VALID_TRANSITIONS assert "Cancelled" in VALID_TRANSITIONS assert VALID_TRANSITIONS["Cancelled"] == [] async def test_cancelled_reachable_from_active_states(client): """Logged/Triage/In Progress → Cancelled are all valid transitions.""" from app.services.ticket import VALID_TRANSITIONS for state in ("Logged", "Triage", "Assigned", "Accepted", "In Progress", "Escalated"): assert "Cancelled" in VALID_TRANSITIONS[state], f"{state} should allow Cancelled" async def test_cancel_ticket_via_api(client): """PATCH status=Cancelled works end-to-end and lands in the timeline.""" token = await _login(client) ticket = await _create_ticket(client, token, description="cancel me") resp = await client.patch( f"/api/tickets/{ticket['id']}", json={"status": "Cancelled", "note": "demo sample"}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200, resp.text data = resp.json() assert data["status"] == "Cancelled" assert data["timeline"][-1]["to_status"] == "Cancelled" async def test_cancelled_ticket_not_sla_breached(client): """A cancelled ticket must not report SLA breach (like Closed/Completed).""" token = await _login(client) ticket = await _create_ticket(client, token, priority="urgent", description="cancel sla test") # Force the SLA deadline into the past by patching priority (recomputes from now), # then cancel; the breach flags must be False either way. resp = await client.patch( f"/api/tickets/{ticket['id']}", json={"status": "Cancelled"}, headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 200 sla = resp.json()["sla_status"] assert sla["resolution_breached"] is False assert sla["response_breached"] is False # ── Phone persistence ───────────────────────────────────────────────── async def test_phone_persisted_on_create(client): """Customer phone sent at creation is stored and returned.""" token = await _login(client) ticket = await _create_ticket(client, token, phone="+233123456789", description="phone test") assert ticket["phone"] == "+233123456789" detail = await client.get(f"/api/tickets/{ticket['id']}") assert detail.json()["phone"] == "+233123456789" # ── Nested unit/category in detail ──────────────────────────────────── async def test_ticket_detail_returns_unit_and_category(client): """TicketOut includes nested unit/category objects (detail page fix).""" token = await _login(client) ticket = await _create_ticket(client, token, unit_id=2, category_id=3) resp = await client.get(f"/api/tickets/{ticket['id']}") assert resp.status_code == 200 data = resp.json() assert data["unit"] is not None assert data["unit"]["apartment_code"] assert data["category"] is not None assert data["category"]["name"] # ── Admin-only DELETE ───────────────────────────────────────────────── async def test_delete_ticket_requires_admin(client): """A non-admin (CS Rep) gets 403 on DELETE.""" token = await _login(client, email="bella@denya.com") # CS Rep ticket = await _create_ticket(client, token, description="delete perm test") resp = await client.delete( f"/api/tickets/{ticket['id']}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 403 async def test_delete_ticket_removes_children(client): """Admin DELETE removes the ticket, timeline, and photos.""" token = await _login(client) # Admin/Wahab ticket = await _create_ticket(client, token, description="delete me") tid = ticket["id"] resp = await client.delete( f"/api/tickets/{tid}", headers={"Authorization": f"Bearer {token}"}, ) assert resp.status_code == 204 gone = await client.get(f"/api/tickets/{tid}") assert gone.status_code == 404 from sqlalchemy import func, select from app.core.database import async_session_factory from app.models.ticket import TicketTimeline async with async_session_factory() as session: count = (await session.execute( select(func.count(TicketTimeline.id)).where(TicketTimeline.ticket_id == tid) )).scalar() assert count == 0 # ── Users endpoint for the assign picker ────────────────────────────── async def test_users_endpoint_requires_auth(client): resp = await client.get("/api/auth/users") assert resp.status_code == 401 async def test_users_endpoint_lists_technicians(client): token = await _login(client) resp = await client.get("/api/auth/users", headers={"Authorization": f"Bearer {token}"}) assert resp.status_code == 200 users = resp.json() assert any(u["role"] == "Tech" for u in users) # Fields the assign dropdown needs assert {"id", "full_name", "role"} <= set(users[0].keys()) # ── Ticket numbering survives deletions ─────────────────────────────── async def test_ticket_number_uses_max_plus_one(client, seed_tickets): """After deleting a middle ticket, numbering must skip over surviving numbers.""" from sqlalchemy import delete as sa_delete from app.core.database import async_session_factory from app.models.ticket import Ticket token = await _login(client) first = await _create_ticket(client, token, description="numbering a") # …001 second = await _create_ticket(client, token, description="numbering b") # …002 third = await _create_ticket(client, token, description="numbering c") # …003 async with async_session_factory() as session: # Delete the middle ticket; count+1 numbering would now re-issue …003 # (colliding with the surviving third ticket). await session.execute(sa_delete(Ticket).where(Ticket.id == second["id"])) await session.commit() fourth = await _create_ticket(client, token, description="numbering d") suffix = lambda tn: int(tn.rsplit("-", 1)[1]) survivors = {suffix(first["ticket_number"]), suffix(third["ticket_number"])} assert suffix(fourth["ticket_number"]) not in survivors assert suffix(fourth["ticket_number"]) > max(survivors)