tests/test_permissions.py

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.core.database import Base
from app.core.security import create_access_token, hash_password
from app.dependencies import get_db
from app.models.department import Department
from app.models.query import Query
from app.models.user import User
from app.routes import admin, auth, queries


@pytest.fixture(scope="module")
def password_hash():
    return hash_password("test-password-123")


@pytest.fixture
def workspace(password_hash):
    engine = create_engine(
        "sqlite+pysqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(engine)
    with sessionmaker(bind=engine)() as db:
        departments = [
            Department(id=11, name="Examination Department", code="EXAM", keywords="exam"),
            Department(id=22, name="Finance Department", code="FIN", keywords="fees"),
            Department(id=33, name="Registrar", code="REG", keywords="transcript"),
            Department(id=44, name="IT Dept", code="IT", keywords="wifi"),
            Department(id=55, name="Instructor", code="INST", keywords="feedback"),
            Department(id=66, name="Admin", code="ADMIN", keywords="general"),
        ]
        db.add_all(departments)
        db.flush()
        users = {
            "student": User(email="student@example.com", role="STUDENT"),
            "other_student": User(email="other@example.com", role="STUDENT"),
            "admin": User(email="admin@example.com", role="ADMIN", is_admin=True),
            "unassigned_staff": User(email="unassigned@example.com", role="STAFF"),
        }
        for department in departments:
            users[department.code] = User(
                email=f"staff{department.id}@example.com",
                role=department.name,
                department_id=department.id,
            )
        for user in users.values():
            user.hashed_password = password_hash
        db.add_all(users.values())
        db.flush()
        for query_id, department in enumerate(departments, start=101):
            db.add(Query(
                id=query_id, student_id=users["student"].id,
                sender_email=users["student"].email,
                message=f"Question for {department.name}",
                department_id=department.id,
            ))
        db.add_all([
            Query(id=107, student_id=users["other_student"].id, sender_email="other@example.com",
                  message="Another student's fees query", department_id=22),
            Query(id=108, student_id=users["other_student"].id, sender_email="other@example.com",
                  message="Unassigned query"),
            Query(id=109, student_id=users["FIN"].id, sender_email=users["FIN"].email,
                  message="Staff's own exam question", department_id=11),
        ])
        db.commit()

        app = FastAPI()
        for router in (auth.router, queries.router, admin.router):
            app.include_router(router)

        def test_db():
            yield db

        app.dependency_overrides[get_db] = test_db
        with TestClient(app) as client:
            yield client, db, users
    engine.dispose()


def headers(user):
    return {"Authorization": f"Bearer {create_access_token(str(user.id))}"}


@pytest.mark.parametrize(("account", "expected_ids"), [
    ("EXAM", {101, 109}), ("FIN", {102, 107}), ("REG", {103}),
    ("IT", {104}), ("INST", {105}), ("ADMIN", {106}),
])
def test_department_staff_only_receive_assigned_queries(workspace, account, expected_ids):
    client, _, users = workspace
    response = client.get("/queries/", headers=headers(users[account]))
    assert response.status_code == 200
    assert {item["id"] for item in response.json()} == expected_ids


@pytest.mark.parametrize("path", ["/queries/", "/admin/queries"])
def test_admin_receives_all_departments_and_unassigned_queries(workspace, path):
    client, _, users = workspace
    response = client.get(path, headers=headers(users["admin"]))
    assert response.status_code == 200
    assert {item["id"] for item in response.json()} == set(range(101, 110))


@pytest.mark.parametrize(("account", "expected_ids"), [
    ("student", {101, 102, 103, 104, 105, 106}), ("other_student", {107, 108}),
])
def test_students_only_receive_their_own_queries(workspace, account, expected_ids):
    client, _, users = workspace
    response = client.get("/queries/", headers=headers(users[account]))
    assert response.status_code == 200
    assert {item["id"] for item in response.json()} == expected_ids


def test_student_role_does_not_grant_department_access(workspace):
    client, db, users = workspace
    users["student"].department_id = 22
    db.commit()
    response = client.get("/queries/", headers=headers(users["student"]))
    assert {item["id"] for item in response.json()} == {101, 102, 103, 104, 105, 106}


def test_staff_without_department_cannot_see_unassigned_queries(workspace):
    client, _, users = workspace
    response = client.get("/queries/", headers=headers(users["unassigned_staff"]))
    assert response.status_code == 200
    assert response.json() == []


@pytest.mark.parametrize("account", ["EXAM", "FIN", "REG", "IT", "INST", "ADMIN", "student"])
def test_only_admin_can_use_all_queries_endpoint(workspace, account):
    client, _, users = workspace
    assert client.get("/admin/queries", headers=headers(users[account])).status_code == 403


@pytest.mark.parametrize("path", ["/queries/", "/admin/queries"])
def test_reading_queries_requires_login(workspace, path):
    client, _, _ = workspace
    assert client.get(path).status_code == 401


def test_request_parameters_cannot_change_a_staff_members_scope(workspace):
    client, _, users = workspace
    response = client.get("/queries/?department_id=11&role=Admin&is_admin=true",
                          headers=headers(users["FIN"]))
    assert response.status_code == 200
    assert {item["id"] for item in response.json()} == {102, 107}


@pytest.mark.parametrize(("keyword", "department_id", "staff_account"), [
    ("exam", 11, "EXAM"), ("fees", 22, "FIN"), ("transcript", 33, "REG"),
    ("wifi", 44, "IT"), ("feedback", 55, "INST"), ("general", 66, "ADMIN"),
])
def test_submitted_query_is_visible_to_its_department_student_and_admin(workspace, keyword, department_id, staff_account):
    client, _, users = workspace
    response = client.post("/queries/", headers=headers(users["student"]), json={"message": keyword})
    assert response.status_code == 201
    query_id = response.json()["id"]
    assert response.json()["department_id"] == department_id
    for account in ("student", staff_account, "admin"):
        response = client.get("/queries/", headers=headers(users[account]))
        assert query_id in {item["id"] for item in response.json()}
    for account in ("EXAM", "FIN", "REG", "IT", "INST", "ADMIN", "other_student"):
        if account != staff_account:
            response = client.get("/queries/", headers=headers(users[account]))
            assert query_id not in {item["id"] for item in response.json()}


def test_login_uses_saved_account_role_for_query_access(workspace):
    client, _, users = workspace
    response = client.post("/auth/login", json={
        "email": users["FIN"].email, "password": "test-password-123",
        "selectedRole": "Admin", "is_admin": True,
    })
    assert response.status_code == 200
    assert response.json()["user"]["is_admin"] is False
    assert response.json()["user"]["department_id"] == 22
    token_headers = {"Authorization": f"Bearer {response.json()['access_token']}"}
    response = client.get("/queries/", headers=token_headers)
    assert {item["id"] for item in response.json()} == {102, 107}
    assert client.get("/admin/queries", headers=token_headers).status_code == 403