Add security fixes, testing infrastructure, and CI/CD improvements
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
@@ -0,0 +1,186 @@
|
||||
"""
|
||||
Pytest configuration and shared fixtures for DocuElevate tests.
|
||||
"""
|
||||
import os
|
||||
import tempfile
|
||||
import pytest
|
||||
from typing import Generator
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
# Set test environment variables before importing app
|
||||
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
|
||||
os.environ["REDIS_URL"] = "redis://localhost:6379/1"
|
||||
os.environ["OPENAI_API_KEY"] = "test-key"
|
||||
os.environ["AZURE_AI_KEY"] = "test-key"
|
||||
os.environ["AZURE_REGION"] = "test"
|
||||
os.environ["AZURE_ENDPOINT"] = "https://test.cognitiveservices.azure.com/"
|
||||
os.environ["GOTENBERG_URL"] = "http://localhost:3000"
|
||||
os.environ["WORKDIR"] = "/tmp"
|
||||
os.environ["AUTH_ENABLED"] = "False"
|
||||
os.environ["SESSION_SECRET"] = "test_secret_key_for_testing_must_be_at_least_32_characters_long"
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.main import app as fastapi_app
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def test_workdir() -> Generator[str, None, None]:
|
||||
"""Create a temporary work directory for tests."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield tmpdir
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def db_session():
|
||||
"""Create a fresh database session for each test."""
|
||||
# Create an in-memory SQLite database
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
|
||||
# Create all tables
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
# Create a session
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
session = TestingSessionLocal()
|
||||
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture(scope="function")
|
||||
def client(db_session) -> TestClient:
|
||||
"""Create a test client with a fresh database."""
|
||||
|
||||
# Override the get_db dependency to use our test database
|
||||
def override_get_db():
|
||||
try:
|
||||
yield db_session
|
||||
finally:
|
||||
pass
|
||||
|
||||
fastapi_app.dependency_overrides[get_db] = override_get_db
|
||||
|
||||
with TestClient(fastapi_app) as test_client:
|
||||
yield test_client
|
||||
|
||||
# Clean up
|
||||
fastapi_app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_pdf_path(test_workdir) -> str:
|
||||
"""Create a sample PDF file for testing."""
|
||||
pdf_path = os.path.join(test_workdir, "test.pdf")
|
||||
|
||||
# Create a minimal valid PDF
|
||||
pdf_content = b"""%PDF-1.4
|
||||
1 0 obj
|
||||
<<
|
||||
/Type /Catalog
|
||||
/Pages 2 0 R
|
||||
>>
|
||||
endobj
|
||||
2 0 obj
|
||||
<<
|
||||
/Type /Pages
|
||||
/Kids [3 0 R]
|
||||
/Count 1
|
||||
>>
|
||||
endobj
|
||||
3 0 obj
|
||||
<<
|
||||
/Type /Page
|
||||
/Parent 2 0 R
|
||||
/MediaBox [0 0 612 792]
|
||||
>>
|
||||
endobj
|
||||
xref
|
||||
0 4
|
||||
0000000000 65535 f
|
||||
0000000009 00000 n
|
||||
0000000058 00000 n
|
||||
0000000115 00000 n
|
||||
trailer
|
||||
<<
|
||||
/Size 4
|
||||
/Root 1 0 R
|
||||
>>
|
||||
startxref
|
||||
197
|
||||
%%EOF
|
||||
"""
|
||||
|
||||
with open(pdf_path, 'wb') as f:
|
||||
f.write(pdf_content)
|
||||
|
||||
return pdf_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_text_file(test_workdir) -> str:
|
||||
"""Create a sample text file for testing."""
|
||||
text_path = os.path.join(test_workdir, "test.txt")
|
||||
|
||||
with open(text_path, 'w') as f:
|
||||
f.write("This is a test document.\nWith multiple lines.\n")
|
||||
|
||||
return text_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_openai_response():
|
||||
"""Mock OpenAI API response for testing."""
|
||||
return {
|
||||
"choices": [{
|
||||
"message": {
|
||||
"content": '{"document_type": "invoice", "summary": "Test invoice", "tags": ["test", "invoice"]}'
|
||||
}
|
||||
}]
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_azure_response():
|
||||
"""Mock Azure Document Intelligence API response for testing."""
|
||||
return {
|
||||
"analyzeResult": {
|
||||
"content": "Test document content extracted by OCR",
|
||||
"pages": [{"pageNumber": 1}]
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
# Markers for categorizing tests
|
||||
def pytest_configure(config):
|
||||
"""Configure custom pytest markers."""
|
||||
config.addinivalue_line(
|
||||
"markers", "unit: Unit tests for individual functions/methods"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "integration: Integration tests for API endpoints and workflows"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "slow: Tests that take significant time to run"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "security: Security-related tests"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "requires_external: Tests requiring external services"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "requires_db: Tests requiring database"
|
||||
)
|
||||
config.addinivalue_line(
|
||||
"markers", "requires_redis: Tests requiring Redis"
|
||||
)
|
||||
@@ -0,0 +1,84 @@
|
||||
"""
|
||||
Integration tests for API endpoints.
|
||||
"""
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestHealthEndpoints:
|
||||
"""Tests for health check and status endpoints."""
|
||||
|
||||
def test_root_endpoint(self, client: TestClient):
|
||||
"""Test that root endpoint redirects to UI."""
|
||||
response = client.get("/", follow_redirects=False)
|
||||
assert response.status_code in [200, 307, 308] # OK or redirect
|
||||
|
||||
def test_docs_endpoint(self, client: TestClient):
|
||||
"""Test that API documentation is accessible."""
|
||||
response = client.get("/docs")
|
||||
assert response.status_code == 200
|
||||
assert "swagger" in response.text.lower() or "openapi" in response.text.lower()
|
||||
|
||||
def test_openapi_schema(self, client: TestClient):
|
||||
"""Test that OpenAPI schema is accessible."""
|
||||
response = client.get("/openapi.json")
|
||||
assert response.status_code == 200
|
||||
schema = response.json()
|
||||
assert "openapi" in schema
|
||||
assert "info" in schema
|
||||
assert schema["info"]["title"] == "DocuElevate"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestFileEndpoints:
|
||||
"""Tests for file management endpoints."""
|
||||
|
||||
def test_list_files_empty(self, client: TestClient):
|
||||
"""Test listing files when database is empty."""
|
||||
response = client.get("/api/files")
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert isinstance(data, list)
|
||||
assert len(data) == 0
|
||||
|
||||
def test_get_nonexistent_file(self, client: TestClient):
|
||||
"""Test getting a file that doesn't exist."""
|
||||
response = client.get("/api/files/99999")
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.requires_external
|
||||
class TestProcessingEndpoints:
|
||||
"""Tests for document processing endpoints."""
|
||||
|
||||
def test_process_endpoint_exists(self, client: TestClient):
|
||||
"""Test that process endpoint is registered."""
|
||||
# This will return 422 (validation error) without proper data,
|
||||
# but confirms the endpoint exists
|
||||
response = client.post("/api/process")
|
||||
assert response.status_code in [400, 422] # Bad request or validation error
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestConfigEndpoints:
|
||||
"""Tests for configuration endpoints."""
|
||||
|
||||
def test_config_status_endpoint(self, client: TestClient):
|
||||
"""Test configuration status endpoint if it exists."""
|
||||
# Some apps have a /status or /config/status endpoint
|
||||
response = client.get("/api/status")
|
||||
# Endpoint may not exist, which is fine
|
||||
assert response.status_code in [200, 404]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestAuthEndpoints:
|
||||
"""Tests for authentication endpoints (when auth is disabled in tests)."""
|
||||
|
||||
def test_unauthenticated_access_with_auth_disabled(self, client: TestClient):
|
||||
"""Test that API is accessible when auth is disabled."""
|
||||
# With AUTH_ENABLED=False, API should be accessible
|
||||
response = client.get("/api/files")
|
||||
assert response.status_code == 200
|
||||
@@ -0,0 +1,163 @@
|
||||
"""
|
||||
Unit tests for configuration and security validation.
|
||||
"""
|
||||
import pytest
|
||||
import os
|
||||
from pydantic import ValidationError
|
||||
from app.config import Settings
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestConfigurationValidation:
|
||||
"""Tests for configuration validation."""
|
||||
|
||||
def test_session_secret_required_with_auth(self):
|
||||
"""Test that SESSION_SECRET is required when auth is enabled."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=True,
|
||||
session_secret=None
|
||||
)
|
||||
assert "SESSION_SECRET must be set" in str(exc_info.value)
|
||||
|
||||
def test_session_secret_minimum_length(self):
|
||||
"""Test that SESSION_SECRET must be at least 32 characters."""
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=True,
|
||||
session_secret="short"
|
||||
)
|
||||
assert "at least 32 characters" in str(exc_info.value)
|
||||
|
||||
def test_valid_configuration(self):
|
||||
"""Test that valid configuration is accepted."""
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test_key",
|
||||
azure_ai_key="test_key",
|
||||
azure_region="eastus",
|
||||
azure_endpoint="https://test.cognitiveservices.azure.com/",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=True,
|
||||
session_secret="a" * 32 # 32 character secret
|
||||
)
|
||||
assert config.auth_enabled is True
|
||||
assert len(config.session_secret) == 32
|
||||
|
||||
def test_auth_disabled_no_session_secret_required(self):
|
||||
"""Test that SESSION_SECRET is not required when auth is disabled."""
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test_key",
|
||||
azure_ai_key="test_key",
|
||||
azure_region="eastus",
|
||||
azure_endpoint="https://test.cognitiveservices.azure.com/",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=False,
|
||||
session_secret=None
|
||||
)
|
||||
assert config.auth_enabled is False
|
||||
assert config.session_secret is None
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestNotificationConfiguration:
|
||||
"""Tests for notification configuration parsing."""
|
||||
|
||||
def test_notification_urls_from_string(self):
|
||||
"""Test parsing notification URLs from comma-separated string."""
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=False,
|
||||
notification_urls="discord://webhook1,telegram://webhook2"
|
||||
)
|
||||
assert len(config.notification_urls) == 2
|
||||
assert "discord://webhook1" in config.notification_urls
|
||||
assert "telegram://webhook2" in config.notification_urls
|
||||
|
||||
def test_notification_urls_from_list(self):
|
||||
"""Test that notification URLs can be provided as a list."""
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=False,
|
||||
notification_urls=["discord://webhook1", "telegram://webhook2"]
|
||||
)
|
||||
assert len(config.notification_urls) == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.security
|
||||
class TestSecurityConfiguration:
|
||||
"""Tests for security-related configuration."""
|
||||
|
||||
def test_no_default_credentials_in_config(self):
|
||||
"""Test that no default credentials are present in configuration."""
|
||||
# This test ensures we don't accidentally have hardcoded credentials
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=False
|
||||
)
|
||||
|
||||
# Ensure optional credentials are actually optional (None)
|
||||
assert config.dropbox_app_key is None
|
||||
assert config.dropbox_app_secret is None
|
||||
assert config.nextcloud_password is None
|
||||
assert config.paperless_ngx_api_token is None
|
||||
|
||||
def test_optional_services_dont_require_credentials(self):
|
||||
"""Test that application can start without optional service credentials."""
|
||||
config = Settings(
|
||||
database_url="sqlite:///test.db",
|
||||
redis_url="redis://localhost:6379",
|
||||
openai_api_key="test",
|
||||
azure_ai_key="test",
|
||||
azure_region="test",
|
||||
azure_endpoint="https://test.example.com",
|
||||
gotenberg_url="http://localhost:3000",
|
||||
workdir="/tmp",
|
||||
auth_enabled=False
|
||||
)
|
||||
|
||||
# Should not raise an error
|
||||
assert config.database_url == "sqlite:///test.db"
|
||||
Reference in New Issue
Block a user