243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""
|
|
Integration tests for Options API endpoints.
|
|
Tests the actual API endpoints with real database connections.
|
|
"""
|
|
|
|
import pytest
|
|
import secrets
|
|
from fastapi.testclient import TestClient
|
|
from modules.datamodels.datamodelUam import User
|
|
from modules.interfaces.interfaceDbApp import getRootInterface
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create FastAPI app instance for testing."""
|
|
from app import app as fastapi_app
|
|
return fastapi_app
|
|
|
|
|
|
@pytest.fixture
|
|
def testClient(app):
|
|
"""Create test client for API testing."""
|
|
return TestClient(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def csrfToken():
|
|
"""Generate a valid CSRF token for testing."""
|
|
# Generate a hex string between 16-64 characters (CSRF validation requirement)
|
|
return secrets.token_hex(16) # 32 character hex string
|
|
|
|
|
|
@pytest.fixture
|
|
def testUser() -> User:
|
|
"""Create a test user for API testing."""
|
|
# Use getRootInterface for system operations like user creation
|
|
# The root interface automatically uses the root mandate
|
|
rootInterface = getRootInterface()
|
|
user = rootInterface.createUser(
|
|
username="testuser_options",
|
|
email="testuser_options@example.com",
|
|
password="testpass123",
|
|
roleLabels=["user"]
|
|
)
|
|
return user
|
|
|
|
|
|
class TestOptionsAPI:
|
|
"""Test Options API endpoints."""
|
|
|
|
def testGetOptionsUserRole(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/user.role endpoint."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get options
|
|
response = testClient.get(
|
|
"/api/options/user.role",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
options = response.json()
|
|
|
|
assert isinstance(options, list)
|
|
assert len(options) >= 4 # At least sysadmin, admin, user, viewer
|
|
|
|
# Check structure
|
|
for option in options:
|
|
assert "value" in option
|
|
assert "label" in option
|
|
assert isinstance(option["label"], dict)
|
|
|
|
# Check specific values
|
|
values = [opt["value"] for opt in options]
|
|
assert "sysadmin" in values
|
|
assert "admin" in values
|
|
assert "user" in values
|
|
assert "viewer" in values
|
|
|
|
def testGetOptionsAuthAuthority(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/auth.authority endpoint."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get options
|
|
response = testClient.get(
|
|
"/api/options/auth.authority",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
options = response.json()
|
|
|
|
assert isinstance(options, list)
|
|
assert len(options) == 3 # local, google, msft
|
|
|
|
# Check structure
|
|
for option in options:
|
|
assert "value" in option
|
|
assert "label" in option
|
|
|
|
# Check specific values
|
|
values = [opt["value"] for opt in options]
|
|
assert "local" in values
|
|
assert "google" in values
|
|
assert "msft" in values
|
|
|
|
def testGetOptionsConnectionStatus(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/connection.status endpoint."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get options
|
|
response = testClient.get(
|
|
"/api/options/connection.status",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
options = response.json()
|
|
|
|
assert isinstance(options, list)
|
|
assert len(options) >= 4 # active, inactive, expired, pending, revoked, error
|
|
|
|
# Check structure
|
|
for option in options:
|
|
assert "value" in option
|
|
assert "label" in option
|
|
|
|
def testGetOptionsUserConnection(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/user.connection endpoint (context-aware)."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get options (should return empty list if no connections)
|
|
response = testClient.get(
|
|
"/api/options/user.connection",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
options = response.json()
|
|
|
|
# Should return a list (may be empty)
|
|
assert isinstance(options, list)
|
|
|
|
def testGetOptionsList(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/ endpoint (list all available options)."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get available options names
|
|
response = testClient.get(
|
|
"/api/options/",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
optionsNames = response.json()
|
|
|
|
assert isinstance(optionsNames, list)
|
|
assert "user.role" in optionsNames
|
|
assert "auth.authority" in optionsNames
|
|
assert "connection.status" in optionsNames
|
|
assert "user.connection" in optionsNames
|
|
|
|
def testGetOptionsUnknown(self, testClient, testUser, csrfToken):
|
|
"""Test GET /api/options/unknown.options endpoint (should return 400)."""
|
|
# Get auth token (stored in cookie)
|
|
response = testClient.post(
|
|
"/api/local/login",
|
|
data={"username": testUser.username, "password": "testpass123"},
|
|
headers={"X-CSRF-Token": csrfToken}
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
# Extract token from cookie for Bearer header
|
|
token = response.cookies.get("auth_token")
|
|
assert token is not None
|
|
|
|
# Get unknown options (should return error)
|
|
response = testClient.get(
|
|
"/api/options/unknown.options",
|
|
headers={"Authorization": f"Bearer {token}"}
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
def testGetOptionsUnauthorized(self, testClient):
|
|
"""Test GET /api/options/user.role without authentication."""
|
|
# Try to get options without auth token
|
|
response = testClient.get("/api/options/user.role")
|
|
|
|
# Should require authentication
|
|
assert response.status_code == 401
|