291 lines
12 KiB
Python
291 lines
12 KiB
Python
"""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 (
|
|
"New", "Logged", "Triage", "Assigned", "Accepted", "Travelling", "On Site",
|
|
"In Progress", "Waiting Parts", "Escalated", "On-Field Verification",
|
|
"Wahab Review", "Reopened",
|
|
):
|
|
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())
|
|
|
|
|
|
# ── Transitions helper (status dropdown) ─────────────────────────────
|
|
async def test_transitions_helper_returns_valid_targets(client):
|
|
"""GET /api/tickets/{id}/transitions returns the valid next statuses."""
|
|
from app.services.ticket import VALID_TRANSITIONS
|
|
|
|
token = await _login(client)
|
|
ticket = await _create_ticket(client, token, description="transitions helper") # status: Logged
|
|
|
|
resp = await client.get(f"/api/tickets/{ticket['id']}/transitions")
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["current_status"] == "Logged"
|
|
assert set(data["transitions"]) == set(VALID_TRANSITIONS["Logged"])
|
|
|
|
|
|
async def test_transitions_helper_terminal_is_empty(client):
|
|
"""A cancelled ticket exposes no selectable next statuses."""
|
|
token = await _login(client)
|
|
ticket = await _create_ticket(client, token, description="terminal transitions")
|
|
resp = await client.patch(
|
|
f"/api/tickets/{ticket['id']}",
|
|
json={"status": "Cancelled"},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
data = (await client.get(f"/api/tickets/{ticket['id']}/transitions")).json()
|
|
assert data["current_status"] == "Cancelled"
|
|
assert data["transitions"] == []
|
|
|
|
|
|
# ── Assign auto-advance ───────────────────────────────────────────────
|
|
async def _first_tech_id(client, token: str) -> int:
|
|
users = (await client.get("/api/auth/users", headers={"Authorization": f"Bearer {token}"})).json()
|
|
return next(u["id"] for u in users if u["role"] == "Tech")
|
|
|
|
|
|
async def test_assign_auto_advances_to_assigned(client):
|
|
"""Assigning a pre-Assigned ticket advances it to Assigned with a timeline entry."""
|
|
token = await _login(client)
|
|
ticket = await _create_ticket(client, token, description="assign advance") # status: Logged
|
|
tech_id = await _first_tech_id(client, token)
|
|
|
|
resp = await client.patch(
|
|
f"/api/tickets/{ticket['id']}",
|
|
json={"assigned_to": tech_id},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
data = resp.json()
|
|
assert data["status"] == "Assigned"
|
|
assert data["assigned_to"] == tech_id
|
|
assert data["timeline"][-1]["to_status"] == "Assigned"
|
|
|
|
|
|
async def test_assign_does_not_regress_past_assigned(client):
|
|
"""Re-assigning a ticket already past Assigned leaves its status untouched."""
|
|
token = await _login(client)
|
|
ticket = await _create_ticket(client, token, description="no regression")
|
|
|
|
for step in ("Triage", "Assigned", "Accepted"):
|
|
resp = await client.patch(
|
|
f"/api/tickets/{ticket['id']}",
|
|
json={"status": step},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
users = (await client.get("/api/auth/users", headers={"Authorization": f"Bearer {token}"})).json()
|
|
techs = [u["id"] for u in users if u["role"] == "Tech"]
|
|
other_tech = techs[1] if len(techs) > 1 else techs[0]
|
|
|
|
resp = await client.patch(
|
|
f"/api/tickets/{ticket['id']}",
|
|
json={"assigned_to": other_tech},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json()["status"] == "Accepted"
|
|
|
|
|
|
# ── 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")
|
|
|
|
def suffix(tn: str) -> int:
|
|
return 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)
|