100 lines
3.6 KiB
Python
100 lines
3.6 KiB
Python
from fastapi import status
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
from unittest.mock import patch
|
|
|
|
from tests.helpers import generators
|
|
|
|
# Test admin routes require admin privileges
|
|
|
|
|
|
def test_read_admin_unauthorized(client: TestClient) -> None:
|
|
"""Test accessing admin route without authentication."""
|
|
response = client.get("/api/admin/")
|
|
assert response.status_code == status.HTTP_401_UNAUTHORIZED
|
|
|
|
|
|
def test_read_admin_forbidden(db: Session, client: TestClient) -> None:
|
|
"""Test accessing admin route as a non-admin user."""
|
|
user, password = generators.create_user(db, is_admin=False) # Use is_admin=False
|
|
login_rsp = generators.login(db, user.username, password)
|
|
access_token = login_rsp["access_token"]
|
|
|
|
response = client.get(
|
|
"/api/admin/", headers={"Authorization": f"Bearer {access_token}"}
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|
|
|
|
|
|
def test_read_admin_success(db: Session, client: TestClient) -> None:
|
|
"""Test accessing admin route as an admin user."""
|
|
admin_user, password = generators.create_user(
|
|
db, is_admin=True
|
|
) # Use is_admin=True
|
|
login_rsp = generators.login(db, admin_user.username, password)
|
|
access_token = login_rsp["access_token"]
|
|
|
|
response = client.get(
|
|
"/api/admin/", headers={"Authorization": f"Bearer {access_token}"}
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == {"message": "Admin route"}
|
|
|
|
|
|
@patch("modules.admin.api.cleardb.delay") # Mock the celery task
|
|
def test_clear_db_soft(mock_cleardb_delay, db: Session, client: TestClient) -> None:
|
|
"""Test soft clearing the database as admin."""
|
|
admin_user, password = generators.create_user(
|
|
db, is_admin=True
|
|
) # Use is_admin=True
|
|
login_rsp = generators.login(db, admin_user.username, password)
|
|
access_token = login_rsp["access_token"]
|
|
|
|
response = client.post(
|
|
"/api/admin/cleardb",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
json={"hard": False},
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == {
|
|
"message": "Clearing database in the background",
|
|
"hard": False,
|
|
}
|
|
mock_cleardb_delay.assert_called_once_with(False)
|
|
|
|
|
|
@patch("modules.admin.api.cleardb.delay") # Mock the celery task
|
|
def test_clear_db_hard(mock_cleardb_delay, db: Session, client: TestClient) -> None:
|
|
"""Test hard clearing the database as admin."""
|
|
admin_user, password = generators.create_user(
|
|
db, is_admin=True
|
|
) # Use is_admin=True
|
|
login_rsp = generators.login(db, admin_user.username, password)
|
|
access_token = login_rsp["access_token"]
|
|
|
|
response = client.post(
|
|
"/api/admin/cleardb",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
json={"hard": True},
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
assert response.json() == {
|
|
"message": "Clearing database in the background",
|
|
"hard": True,
|
|
}
|
|
mock_cleardb_delay.assert_called_once_with(True)
|
|
|
|
|
|
def test_clear_db_forbidden(db: Session, client: TestClient) -> None:
|
|
"""Test clearing the database as a non-admin user."""
|
|
user, password = generators.create_user(db, is_admin=False) # Use is_admin=False
|
|
login_rsp = generators.login(db, user.username, password)
|
|
access_token = login_rsp["access_token"]
|
|
|
|
response = client.post(
|
|
"/api/admin/cleardb",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
json={"hard": False},
|
|
)
|
|
assert response.status_code == status.HTTP_403_FORBIDDEN
|