File size: 8,011 Bytes
eeff6a0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 | 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
|