288 lines
12 KiB
Python
288 lines
12 KiB
Python
"""Tests for the technician-performance dashboard and the ``Tech #N`` fix.
|
|
|
|
Anchors two Wahab-facing defects/requests:
|
|
|
|
1. The FM dashboard built its "Technician Workload" card from
|
|
``Tech #<user id>`` instead of the technician's real name. The data sources
|
|
the card consumes must expose the technician's real name via
|
|
``assigned_technician_name``, never an id placeholder.
|
|
2. ``GET /api/tickets/tech-performance`` reports per-technician workload and
|
|
outcomes by real name: totals, per-status buckets, completion rate and
|
|
``created_at -> closed_at`` resolution times. Everything derives from
|
|
existing ticket columns — no schema change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.core.database import async_session_factory
|
|
from app.models.ticket import Ticket
|
|
from app.models.user import User
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def _login(client, email: str = "wahab@denya.com", password: str = "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"]
|
|
|
|
|
|
def _auth(token: str) -> dict[str, str]:
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
async def _user_id(email: str) -> int:
|
|
async with async_session_factory() as session:
|
|
user = (
|
|
await session.execute(select(User).where(User.email == email))
|
|
).scalar_one()
|
|
return user.id
|
|
|
|
|
|
async def _seed_tickets(specs: list[dict]) -> None:
|
|
"""Insert tickets directly so tests control status/timestamps exactly."""
|
|
async with async_session_factory() as session:
|
|
for i, spec in enumerate(specs):
|
|
session.add(Ticket(ticket_number=f"PAV-PERF-{i:05d}", **spec))
|
|
await session.commit()
|
|
|
|
|
|
async def _make_user(email: str, full_name: str, role: str = "Tech") -> int:
|
|
"""Create a user directly (no API) so tests can build same-name technicians."""
|
|
async with async_session_factory() as session:
|
|
user = User(email=email, password_hash="not-a-real-hash", full_name=full_name, role=role)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user.id
|
|
|
|
|
|
def _ticket(status: str, assigned_to: int | None, *, created_at: datetime | None = None,
|
|
closed_at: datetime | None = None) -> dict:
|
|
spec: dict = {
|
|
"status": status,
|
|
"priority": "medium",
|
|
"description": f"{status} ticket",
|
|
"assigned_to": assigned_to,
|
|
}
|
|
if created_at is not None:
|
|
spec["created_at"] = created_at
|
|
if closed_at is not None:
|
|
spec["closed_at"] = closed_at
|
|
return spec
|
|
|
|
|
|
# ── 3a. Real technician names on the workload data paths ─────────────
|
|
async def test_workload_sources_report_real_technician_names(client):
|
|
"""The FM workload card reads ``/api/tickets?page_size=200`` and the
|
|
performance report reads ``/api/tickets/tech-performance``. For an assigned
|
|
ticket both must expose the technician's real full name, never a
|
|
``Tech #<id>`` placeholder.
|
|
"""
|
|
prosper = await _user_id("prosper@denya.com")
|
|
await _seed_tickets([_ticket("In Progress", prosper), _ticket("Escalated", prosper)])
|
|
|
|
token = await _login(client)
|
|
list_resp = await client.get("/api/tickets?page_size=200", headers=_auth(token))
|
|
assert list_resp.status_code == 200
|
|
active = [
|
|
t for t in list_resp.json()["items"]
|
|
if t["status"] not in ("Closed", "Completed", "Cancelled")
|
|
]
|
|
|
|
# Same mapping the FM workload card builds from active assigned tickets.
|
|
workload: dict[int, str] = {}
|
|
for t in active:
|
|
if t["assigned_to"]:
|
|
workload.setdefault(t["assigned_to"], t["assigned_technician_name"] or "Unassigned")
|
|
assert workload == {prosper: "Prosper"}
|
|
|
|
perf_resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
assert perf_resp.status_code == 200
|
|
rows = {t["technician_id"]: t for t in perf_resp.json()["technicians"]}
|
|
assert rows[prosper]["name"] == "Prosper"
|
|
assert all(not row["name"].startswith("Tech #") for row in rows.values())
|
|
|
|
|
|
# ── Page + route wiring ──────────────────────────────────────────────
|
|
async def test_tech_performance_page_is_same_origin(client):
|
|
"""The new dashboard is reachable and uses only same-origin assets."""
|
|
resp = await client.get("/dashboard/tech-performance")
|
|
assert resp.status_code == 200
|
|
body = resp.text
|
|
assert "/api/tickets/tech-performance" in body
|
|
assert "https://" not in body
|
|
assert "cdn." not in body
|
|
|
|
|
|
async def test_base_nav_links_fm_and_ceo_to_tech_performance(client):
|
|
"""FM and CEO navigation exposes the report."""
|
|
for path in ("/dashboard/fm", "/dashboard/ceo"):
|
|
body = (await client.get(path)).text
|
|
assert "/dashboard/tech-performance" in body
|
|
|
|
|
|
async def test_tech_performance_requires_auth(client):
|
|
"""The aggregate endpoint is bearer-gated, like other dashboard data paths."""
|
|
resp = await client.get("/api/tickets/tech-performance")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
async def test_tech_performance_route_not_shadowed_by_ticket_id(client):
|
|
"""`/tech-performance` is a literal route, not an int ticket id (no 422)."""
|
|
token = await _login(client)
|
|
resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# ── 3b. Aggregation ──────────────────────────────────────────────────
|
|
async def test_tech_performance_empty_state(client):
|
|
"""No assigned tickets → empty rows and a fully zeroed roll-up."""
|
|
token = await _login(client)
|
|
resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["technicians"] == []
|
|
assert data["totals"]["technicians"] == 0
|
|
assert data["totals"]["total_assigned"] == 0
|
|
assert data["totals"]["completion_rate"] == 0.0
|
|
assert data["totals"]["avg_resolution_hours"] is None
|
|
|
|
|
|
async def test_tech_performance_aggregates_by_real_name(client):
|
|
"""Counts, completion rate, aging, sorting and totals per technician."""
|
|
prosper = await _user_id("prosper@denya.com")
|
|
sam = await _user_id("sam@denya.com")
|
|
|
|
await _seed_tickets(
|
|
[
|
|
# Prosper: 2 completed (one timestamped), 1 escalated, 1 in progress, 1 cancelled
|
|
_ticket("Completed", prosper, created_at=datetime(2026, 1, 1, 0, 0),
|
|
closed_at=datetime(2026, 1, 1, 10, 0)),
|
|
_ticket("Completed", prosper),
|
|
_ticket("Escalated", prosper),
|
|
_ticket("In Progress", prosper),
|
|
_ticket("Cancelled", prosper),
|
|
# Sam: 3 closed (timestamped), 1 assigned, 1 awaiting verification
|
|
_ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
|
|
closed_at=datetime(2026, 1, 2, 0, 0)),
|
|
_ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
|
|
closed_at=datetime(2026, 1, 3, 0, 0)),
|
|
_ticket("Closed", sam, created_at=datetime(2026, 1, 1, 0, 0),
|
|
closed_at=datetime(2026, 1, 4, 0, 0)),
|
|
_ticket("Assigned", sam),
|
|
_ticket("On-Field Verification", sam),
|
|
# Unassigned tickets are reported separately, never as a fake technician.
|
|
_ticket("Logged", None),
|
|
_ticket("New", None),
|
|
]
|
|
)
|
|
|
|
token = await _login(client)
|
|
resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
|
|
by_name = {t["name"]: t for t in data["technicians"]}
|
|
assert set(by_name) == {"Prosper", "Sam"}
|
|
assert all("Tech #" not in name for name in by_name)
|
|
|
|
prosper_row = by_name["Prosper"]
|
|
assert prosper_row["technician_id"] == prosper
|
|
assert prosper_row["total_assigned"] == 5
|
|
assert prosper_row["completed"] == 2
|
|
assert prosper_row["closed"] == 0
|
|
assert prosper_row["in_progress"] == 1
|
|
assert prosper_row["escalated"] == 1
|
|
assert prosper_row["cancelled"] == 1
|
|
assert prosper_row["pending"] == 0
|
|
# Buckets always reconcile with the assigned total.
|
|
assert (
|
|
prosper_row["completed"] + prosper_row["in_progress"] + prosper_row["escalated"]
|
|
+ prosper_row["cancelled"] + prosper_row["pending"]
|
|
) == prosper_row["total_assigned"]
|
|
assert prosper_row["open_tickets"] == 2
|
|
assert prosper_row["completion_rate"] == 40.0
|
|
assert prosper_row["avg_resolution_hours"] == 10.0
|
|
assert prosper_row["resolved_with_timestamps"] == 1
|
|
assert prosper_row["resolved_without_timestamps"] == 1
|
|
assert prosper_row["status_breakdown"] == {"Completed": 2, "Cancelled": 1, "Escalated": 1, "In Progress": 1}
|
|
|
|
sam_row = by_name["Sam"]
|
|
assert sam_row["total_assigned"] == 5
|
|
assert sam_row["completed"] == 3
|
|
assert sam_row["closed"] == 3
|
|
assert sam_row["in_progress"] == 0
|
|
assert sam_row["pending"] == 2
|
|
assert sam_row["open_tickets"] == 2
|
|
assert sam_row["completion_rate"] == 60.0
|
|
assert sam_row["avg_resolution_hours"] == 48.0
|
|
assert sam_row["resolved_without_timestamps"] == 0
|
|
assert sam_row["status_breakdown"] == {"Closed": 3, "Assigned": 1, "On-Field Verification": 1}
|
|
|
|
# Sorted by completed desc → Sam first.
|
|
assert [t["name"] for t in data["technicians"]] == ["Sam", "Prosper"]
|
|
|
|
totals = data["totals"]
|
|
assert totals["technicians"] == 2
|
|
assert totals["total_assigned"] == 10
|
|
assert totals["completed"] == 5
|
|
assert totals["closed"] == 3
|
|
assert totals["in_progress"] == 1
|
|
assert totals["escalated"] == 1
|
|
assert totals["cancelled"] == 1
|
|
assert totals["pending"] == 2
|
|
assert totals["open_tickets"] == 4
|
|
assert totals["completion_rate"] == 50.0
|
|
# Fleet average is weighted over tickets, not an average of per-tech averages.
|
|
assert totals["avg_resolution_hours"] == 38.5
|
|
assert totals["resolved_with_timestamps"] == 4
|
|
assert totals["resolved_without_timestamps"] == 1
|
|
assert totals["unassigned_tickets"] == 2
|
|
assert totals["total_tickets"] == 12
|
|
|
|
|
|
async def test_tech_performance_reports_counts_when_timestamps_absent(client):
|
|
"""Finished tasks without ``closed_at`` yield no average but a count."""
|
|
afful = await _user_id("afful@denya.com")
|
|
await _seed_tickets([
|
|
_ticket("Completed", afful),
|
|
_ticket("Closed", afful),
|
|
_ticket("Completed", afful),
|
|
])
|
|
|
|
token = await _login(client)
|
|
resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
row = next(t for t in resp.json()["technicians"] if t["name"] == "Afful")
|
|
assert row["completed"] == 3
|
|
assert row["closed"] == 1
|
|
assert row["avg_resolution_hours"] is None
|
|
assert row["resolved_with_timestamps"] == 0
|
|
assert row["resolved_without_timestamps"] == 3
|
|
|
|
|
|
async def test_tech_performance_groups_same_name_by_user_id(client):
|
|
"""Two technicians sharing a display name stay separate rows."""
|
|
first = await _make_user("kwame.one@denya.com", "Kwame Mensah")
|
|
second = await _make_user("kwame.two@denya.com", "Kwame Mensah")
|
|
await _seed_tickets([
|
|
_ticket("Completed", first),
|
|
_ticket("Completed", second),
|
|
_ticket("Escalated", second),
|
|
])
|
|
|
|
token = await _login(client)
|
|
resp = await client.get("/api/tickets/tech-performance", headers=_auth(token))
|
|
rows = [t for t in resp.json()["technicians"] if t["name"] == "Kwame Mensah"]
|
|
assert len(rows) == 2
|
|
assert {r["technician_id"] for r in rows} == {first, second}
|
|
by_id = {r["technician_id"]: r for r in rows}
|
|
assert by_id[first]["total_assigned"] == 1
|
|
assert by_id[second]["total_assigned"] == 2
|
|
assert by_id[second]["escalated"] == 1
|