pesco-ncr/backend/tests/test_permissions.py

"""Server-side role enforcement per stage and admin-area authorization."""
from .util import (
    create_ncr,
    do_initial_disposition,
    hdr,
    to_costing,
    to_operations,
    to_qc_inspection,
    user_id_by_email,
)


async def test_stage_actions_require_stage_role(client, team):
    ncr = await create_ncr(client, team)

    # a plain requester can't perform initial disposition
    r = await do_initial_disposition(client, team, ncr["id"], as_user=team["requester"])
    assert r.status_code == 403
    # nor can operations/qc/costing roles
    r = await do_initial_disposition(client, team, ncr["id"], as_user=team["ops"])
    assert r.status_code == 403

    await to_operations(client, team, ncr["id"])
    # only operations can mark complete
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["requester"])
    )
    assert r.status_code == 403
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["qc"])
    )
    assert r.status_code == 403

    r = await client.post(
        f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["ops"])
    )
    assert r.status_code == 200

    # only QC can edit inspection fields
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_approval": "yes"},
        headers=hdr(team["ops"]),
    )
    assert r.status_code == 403

    # only costing can cost
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_approval": "yes", "qc_closed": True},
        headers=hdr(team["qc"]),
    )
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing",
        json={"labor_cost": "1", "material_cost": "1", "service_cost": "1", "other_cost": "1"},
        headers=hdr(team["qc"]),
    )
    assert r.status_code == 403


async def test_admin_can_act_at_every_stage(client, team):
    ncr = await create_ncr(client, team)
    r = await do_initial_disposition(client, team, ncr["id"], as_user=team["admin"])
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/operations-complete", headers=hdr(team["admin"])
    )
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/inspection",
        json={"qc_approval": "yes", "qc_closed": True},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/capa",
        json={
            "corrective_action_required": False,
            "corrective_action_justification": "Isolated incident; no systemic cause.",
        },
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 200
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/costing",
        json={"labor_cost": "1", "material_cost": "1", "service_cost": "1", "other_cost": "1"},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 200
    assert r.json()["ncr"]["stage"] == "closed"


async def test_secondary_restricted_to_assignees(client, team):
    ncr = await create_ncr(client, team)
    second_id = await user_id_by_email(
        client, team["dispo"], team["second"], "secondary_disposition_authority"
    )
    await do_initial_disposition(
        client, team, ncr["id"], secondary=True, secondary_ids=[second_id]
    )

    # another user holding the secondary role but NOT assigned is rejected
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/secondary-disposition",
        json={"release": True},
        headers=hdr(team["second2"]),
    )
    assert r.status_code == 403

    # the assignee is allowed
    r = await client.post(
        f"/api/ncrs/{ncr['id']}/secondary-disposition",
        json={"release": True},
        headers=hdr(team["second"]),
    )
    assert r.status_code == 200
    assert r.json()["ncr"]["stage"] == "operations"


async def test_secondary_queue_filtered_by_identity(client, team):
    ncr = await create_ncr(client, team)
    second_id = await user_id_by_email(
        client, team["dispo"], team["second"], "secondary_disposition_authority"
    )
    await do_initial_disposition(
        client, team, ncr["id"], secondary=True, secondary_ids=[second_id]
    )

    r = await client.get("/api/ncrs?queue=secondary", headers=hdr(team["second"]))
    assert any(item["id"] == ncr["id"] for item in r.json()["items"])

    r = await client.get("/api/ncrs?queue=secondary", headers=hdr(team["second2"]))
    assert not any(item["id"] == ncr["id"] for item in r.json()["items"])


async def test_create_requires_valid_disposition_authority(client, team):
    from .util import lookup_ids

    dept_id, cat_id = await lookup_ids(client, team["requester"])
    ops_id = await user_id_by_email(client, team["requester"], team["ops"], "operations")
    r = await client.post(
        "/api/ncrs",
        json={
            "job_number": "J1",
            "department_id": dept_id,
            "deviation_category_id": cat_id,
            "disposition_authority_id": ops_id,  # lacks the role
            "deviation_detail": "Detail long enough.",
        },
        headers=hdr(team["requester"]),
    )
    assert r.status_code == 422


async def test_secondary_assignees_must_hold_role(client, team):
    ncr = await create_ncr(client, team)
    ops_id = await user_id_by_email(client, team["dispo"], team["ops"], "operations")
    r = await do_initial_disposition(
        client, team, ncr["id"], secondary=True, secondary_ids=[ops_id]
    )
    assert r.status_code == 422


async def test_audit_endpoint_restricted(client, team):
    ncr = await create_ncr(client, team)
    r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["requester"]))
    assert r.status_code == 403
    r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["qc"]))
    assert r.status_code == 200
    r = await client.get(f"/api/ncrs/{ncr['id']}/audit", headers=hdr(team["admin"]))
    assert r.status_code == 200


async def test_admin_area_requires_admin(client, team):
    for path in ("/api/admin/users", "/api/admin/departments", "/api/admin/settings",
                 "/api/admin/audit"):
        r = await client.get(path, headers=hdr(team["requester"]))
        assert r.status_code == 403, path
        r = await client.get(path, headers=hdr(team["admin"]))
        assert r.status_code == 200, path


async def test_role_assignment_and_lockout_protection(client, team, make_user):
    target = await make_user(["requester"])
    r = await client.get("/api/admin/users", headers=hdr(team["admin"]))
    target_id = next(u["id"] for u in r.json() if u["email"] == target)
    admin_id = next(u["id"] for u in r.json() if u["email"] == team["admin"])

    r = await client.put(
        f"/api/admin/users/{target_id}/roles",
        json={"roles": ["requester", "qc_inspector"]},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 200
    assert set(r.json()["roles"]) == {"requester", "qc_inspector"}

    # unknown role rejected
    r = await client.put(
        f"/api/admin/users/{target_id}/roles",
        json={"roles": ["superuser"]},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 422

    # admin cannot remove their own admin role
    r = await client.put(
        f"/api/admin/users/{admin_id}/roles",
        json={"roles": ["requester"]},
        headers=hdr(team["admin"]),
    )
    assert r.status_code == 422


async def test_everyone_can_view_and_search(client, team):
    ncr = await create_ncr(client, team)
    r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["ops"]))
    assert r.status_code == 200
    # available_actions reflect the viewer's role
    assert "initial_disposition" not in r.json()["available_actions"]
    r = await client.get(f"/api/ncrs/{ncr['id']}", headers=hdr(team["dispo"]))
    assert "initial_disposition" in r.json()["available_actions"]

    r = await client.get(
        f"/api/ncrs?q={ncr['ncr_number']}", headers=hdr(team["requester"])
    )
    assert r.json()["total"] >= 1


async def test_unknown_dev_user_rejected(client):
    r = await client.get("/api/me", headers=hdr("ghost@pescoinc.biz"))
    assert r.status_code == 401