feat(phase-9): implement MVP polishing and closure
T-032: User management admin panel - backend users router with CRUD, frontend UsersPage with modals T-033: Audit log viewer - backend audit router with filters/pagination, frontend AuditLogPage T-034: Global error handling - ErrorBoundary, LoadingSpinner, ErrorMessage, Toast components T-035: Backend tests - pytest setup with SQLite, tests for health/auth/techniques/tests T-036: Documentation - Updated README with testing section, created docs/API.md
This commit is contained in:
1
backend/tests/__init__.py
Normal file
1
backend/tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# Tests package
|
||||
118
backend/tests/conftest.py
Normal file
118
backend/tests/conftest.py
Normal file
@@ -0,0 +1,118 @@
|
||||
"""Pytest fixtures and configuration for backend tests."""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.main import app
|
||||
from app.database import Base, get_db
|
||||
from app.auth import hash_password
|
||||
from app.models.user import User
|
||||
|
||||
# Use in-memory SQLite for tests
|
||||
SQLALCHEMY_DATABASE_URL = "sqlite:///:memory:"
|
||||
|
||||
engine = create_engine(
|
||||
SQLALCHEMY_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
def override_get_db():
|
||||
"""Override the database dependency for testing."""
|
||||
db = TestingSessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db():
|
||||
"""Create a fresh database for each test."""
|
||||
Base.metadata.create_all(bind=engine)
|
||||
db = TestingSessionLocal()
|
||||
yield db
|
||||
db.close()
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def client(db):
|
||||
"""Create a test client with database override."""
|
||||
app.dependency_overrides[get_db] = override_get_db
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def admin_user(db):
|
||||
"""Create an admin user for testing."""
|
||||
user = User(
|
||||
username="admin",
|
||||
email="admin@test.com",
|
||||
hashed_password=hash_password("admin123"),
|
||||
role="admin",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def red_tech_user(db):
|
||||
"""Create a red_tech user for testing."""
|
||||
user = User(
|
||||
username="redtech",
|
||||
email="redtech@test.com",
|
||||
hashed_password=hash_password("redtech123"),
|
||||
role="red_tech",
|
||||
is_active=True,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
return user
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def admin_token(client, admin_user):
|
||||
"""Get an auth token for the admin user."""
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "admin", "password": "admin123"},
|
||||
)
|
||||
return response.json()["access_token"]
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def red_tech_token(client, red_tech_user):
|
||||
"""Get an auth token for the red_tech user."""
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "redtech", "password": "redtech123"},
|
||||
)
|
||||
return response.json()["access_token"]
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def auth_headers(admin_token):
|
||||
"""Return authorization headers for admin user."""
|
||||
return {"Authorization": f"Bearer {admin_token}"}
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def red_tech_headers(red_tech_token):
|
||||
"""Return authorization headers for red_tech user."""
|
||||
return {"Authorization": f"Bearer {red_tech_token}"}
|
||||
81
backend/tests/test_auth.py
Normal file
81
backend/tests/test_auth.py
Normal file
@@ -0,0 +1,81 @@
|
||||
"""Tests for authentication endpoints."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_login_success(client, admin_user):
|
||||
"""Test successful login returns a token."""
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "admin", "password": "admin123"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
|
||||
def test_login_wrong_password(client, admin_user):
|
||||
"""Test login with wrong password returns 400."""
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "admin", "password": "wrongpassword"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_login_nonexistent_user(client):
|
||||
"""Test login with non-existent user returns 400."""
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "nobody", "password": "password"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_login_inactive_user(client, db):
|
||||
"""Test login with inactive user returns 400."""
|
||||
from app.auth import hash_password
|
||||
from app.models.user import User
|
||||
|
||||
user = User(
|
||||
username="inactive",
|
||||
hashed_password=hash_password("password"),
|
||||
role="viewer",
|
||||
is_active=False,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/auth/login",
|
||||
data={"username": "inactive", "password": "password"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_get_me_with_token(client, admin_user, admin_token):
|
||||
"""Test /auth/me returns current user with valid token."""
|
||||
response = client.get(
|
||||
"/api/v1/auth/me",
|
||||
headers={"Authorization": f"Bearer {admin_token}"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["username"] == "admin"
|
||||
assert data["role"] == "admin"
|
||||
|
||||
|
||||
def test_get_me_without_token(client):
|
||||
"""Test /auth/me returns 401 without token."""
|
||||
response = client.get("/api/v1/auth/me")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_get_me_invalid_token(client):
|
||||
"""Test /auth/me returns 401 with invalid token."""
|
||||
response = client.get(
|
||||
"/api/v1/auth/me",
|
||||
headers={"Authorization": "Bearer invalidtoken"},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
8
backend/tests/test_health.py
Normal file
8
backend/tests/test_health.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""Tests for the health endpoint."""
|
||||
|
||||
|
||||
def test_health_endpoint(client):
|
||||
"""Test that the health endpoint returns 200 OK."""
|
||||
response = client.get("/health")
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"status": "ok"}
|
||||
128
backend/tests/test_techniques.py
Normal file
128
backend/tests/test_techniques.py
Normal file
@@ -0,0 +1,128 @@
|
||||
"""Tests for technique endpoints."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def test_list_techniques_requires_auth(client):
|
||||
"""Test that listing techniques requires authentication."""
|
||||
response = client.get("/api/v1/techniques")
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_list_techniques_empty(client, auth_headers):
|
||||
"""Test listing techniques when none exist."""
|
||||
response = client.get("/api/v1/techniques", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.json() == []
|
||||
|
||||
|
||||
def test_create_technique_requires_admin(client, red_tech_headers):
|
||||
"""Test that creating techniques requires admin role."""
|
||||
response = client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1001", "name": "Test Technique"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_create_technique_success(client, auth_headers):
|
||||
"""Test successful technique creation."""
|
||||
response = client.post(
|
||||
"/api/v1/techniques",
|
||||
json={
|
||||
"mitre_id": "T1059",
|
||||
"name": "Command and Scripting Interpreter",
|
||||
"description": "Test description",
|
||||
"tactic": "execution",
|
||||
},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["mitre_id"] == "T1059"
|
||||
assert data["name"] == "Command and Scripting Interpreter"
|
||||
assert data["status_global"] == "not_evaluated"
|
||||
|
||||
|
||||
def test_create_duplicate_technique(client, auth_headers):
|
||||
"""Test creating a technique with duplicate mitre_id fails."""
|
||||
# Create first technique
|
||||
client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1001", "name": "First"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Try to create duplicate
|
||||
response = client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1001", "name": "Second"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
def test_get_technique_by_mitre_id(client, auth_headers):
|
||||
"""Test getting a single technique by mitre_id."""
|
||||
# Create a technique first
|
||||
client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1059", "name": "Test Technique"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Get it by mitre_id
|
||||
response = client.get("/api/v1/techniques/T1059", headers=auth_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["mitre_id"] == "T1059"
|
||||
|
||||
|
||||
def test_get_nonexistent_technique(client, auth_headers):
|
||||
"""Test getting a non-existent technique returns 404."""
|
||||
response = client.get("/api/v1/techniques/T9999", headers=auth_headers)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_update_technique(client, auth_headers):
|
||||
"""Test updating a technique."""
|
||||
# Create technique
|
||||
client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1059", "name": "Original Name"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Update it
|
||||
response = client.patch(
|
||||
"/api/v1/techniques/T1059",
|
||||
json={"name": "Updated Name", "description": "New description"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["name"] == "Updated Name"
|
||||
|
||||
|
||||
def test_filter_techniques_by_tactic(client, auth_headers):
|
||||
"""Test filtering techniques by tactic."""
|
||||
# Create techniques in different tactics
|
||||
client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1001", "name": "Exec", "tactic": "execution"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1002", "name": "Persist", "tactic": "persistence"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Filter by execution
|
||||
response = client.get(
|
||||
"/api/v1/techniques?tactic=execution",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
techniques = response.json()
|
||||
assert len(techniques) == 1
|
||||
assert techniques[0]["mitre_id"] == "T1001"
|
||||
165
backend/tests/test_tests.py
Normal file
165
backend/tests/test_tests.py
Normal file
@@ -0,0 +1,165 @@
|
||||
"""Tests for security test endpoints."""
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def technique(client, auth_headers):
|
||||
"""Create a technique for test association."""
|
||||
response = client.post(
|
||||
"/api/v1/techniques",
|
||||
json={"mitre_id": "T1059", "name": "Test Technique"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
return response.json()
|
||||
|
||||
|
||||
def test_create_test_requires_auth(client, technique):
|
||||
"""Test that creating a test requires authentication."""
|
||||
response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={
|
||||
"technique_id": technique["id"],
|
||||
"name": "Test Name",
|
||||
},
|
||||
)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_create_test_success(client, red_tech_headers, technique):
|
||||
"""Test successful test creation."""
|
||||
response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={
|
||||
"technique_id": technique["id"],
|
||||
"name": "My Security Test",
|
||||
"description": "Test description",
|
||||
"platform": "windows",
|
||||
},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["name"] == "My Security Test"
|
||||
assert data["state"] == "draft"
|
||||
assert data["technique_id"] == technique["id"]
|
||||
|
||||
|
||||
def test_create_test_nonexistent_technique(client, red_tech_headers):
|
||||
"""Test creating a test with non-existent technique fails."""
|
||||
response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={
|
||||
"technique_id": "00000000-0000-0000-0000-000000000000",
|
||||
"name": "Test",
|
||||
},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
def test_get_test_by_id(client, red_tech_headers, technique):
|
||||
"""Test getting a test by ID."""
|
||||
# Create a test
|
||||
create_response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={"technique_id": technique["id"], "name": "Test"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
test_id = create_response.json()["id"]
|
||||
|
||||
# Get it
|
||||
response = client.get(f"/api/v1/tests/{test_id}", headers=red_tech_headers)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["id"] == test_id
|
||||
|
||||
|
||||
def test_validate_test(client, auth_headers, red_tech_headers, technique):
|
||||
"""Test validating a test updates status correctly."""
|
||||
# Create a test
|
||||
create_response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={"technique_id": technique["id"], "name": "Test"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
test_id = create_response.json()["id"]
|
||||
|
||||
# Validate it (requires lead/admin)
|
||||
response = client.post(
|
||||
f"/api/v1/tests/{test_id}/validate",
|
||||
json={"result": "detected"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["state"] == "validated"
|
||||
assert data["result"] == "detected"
|
||||
assert data["validated_by"] is not None
|
||||
|
||||
|
||||
def test_validate_test_updates_technique_status(client, auth_headers, red_tech_headers, technique):
|
||||
"""Test that validating a test recalculates technique status."""
|
||||
# Create and validate a test
|
||||
create_response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={"technique_id": technique["id"], "name": "Test"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
test_id = create_response.json()["id"]
|
||||
|
||||
client.post(
|
||||
f"/api/v1/tests/{test_id}/validate",
|
||||
json={"result": "detected"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Check technique status was updated
|
||||
response = client.get(
|
||||
f"/api/v1/techniques/{technique['mitre_id']}",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.json()["status_global"] == "validated"
|
||||
|
||||
|
||||
def test_reject_test(client, auth_headers, red_tech_headers, technique):
|
||||
"""Test rejecting a test."""
|
||||
# Create a test
|
||||
create_response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={"technique_id": technique["id"], "name": "Test"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
test_id = create_response.json()["id"]
|
||||
|
||||
# Reject it
|
||||
response = client.post(
|
||||
f"/api/v1/tests/{test_id}/reject",
|
||||
headers=auth_headers,
|
||||
)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["state"] == "rejected"
|
||||
|
||||
|
||||
def test_update_test_only_in_draft(client, auth_headers, red_tech_headers, technique):
|
||||
"""Test that tests can only be updated when in draft/rejected state."""
|
||||
# Create and validate a test
|
||||
create_response = client.post(
|
||||
"/api/v1/tests",
|
||||
json={"technique_id": technique["id"], "name": "Test"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
test_id = create_response.json()["id"]
|
||||
|
||||
client.post(
|
||||
f"/api/v1/tests/{test_id}/validate",
|
||||
json={"result": "detected"},
|
||||
headers=auth_headers,
|
||||
)
|
||||
|
||||
# Try to update validated test
|
||||
response = client.patch(
|
||||
f"/api/v1/tests/{test_id}",
|
||||
json={"name": "New Name"},
|
||||
headers=red_tech_headers,
|
||||
)
|
||||
assert response.status_code == 400
|
||||
Reference in New Issue
Block a user