gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add full-stack integration testing with real infrastructure (PostgreSQL, Redis, Gotenberg, WebDAV, SFTP, MinIO)

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-10 15:24:35 +00:00
parent f2d70e90cf
commit fb50c8b375
5 changed files with 1406 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
requirements-dev.txt
+3
View File
@@ -8,6 +8,9 @@ pytest-asyncio>=0.23.0
pytest-mock>=3.12.0
httpx>=0.26.0 # For async test client
testcontainers>=3.7.1 # For integration tests with real containers
minio>=7.1.0 # For MinIO/S3 integration tests
redis>=4.5.0 # For Redis integration tests
boto3>=1.26.0 # For S3 integration tests
# Code quality
flake8>=7.0.0
tests/README_INTEGRATION_TESTS.md
+495
View File
@@ -0,0 +1,495 @@
# Integration Testing with Real Infrastructure
This directory contains comprehensive integration tests that spin up real infrastructure components using Docker containers.
## Overview
Unlike unit tests that mock external dependencies, these integration tests use **real services** to test the application as close to production as possible:
- **PostgreSQL** - Real database instead of SQLite in-memory
- **Redis** - Real message broker for Celery tasks
- **Gotenberg** - Real PDF conversion service
- **WebDAV Server** - Real upload target
- **SFTP Server** - Real SSH/SFTP server
- **MinIO** - Real S3-compatible object storage
- **FTP Server** - Real FTP server
## Prerequisites
### Required
1. **Docker** - Must be installed and running
```bash
docker --version
docker ps # Should work without errors
```
2. **Python Dependencies**
```bash
pip install -r requirements-dev.txt
```
This installs:
- `testcontainers` - For managing Docker containers in tests
- `pytest` and testing tools
- `minio`, `boto3`, `redis` - Client libraries for services
- `paramiko` - For SFTP testing
### Optional
- Docker Compose (for manual infrastructure setup)
- Sufficient disk space (~2GB for Docker images)
- Sufficient RAM (~4GB recommended)
## Test Organization
### Test Files
| File | Description | Scope |
|------|-------------|-------|
| `test_upload_webdav_comprehensive.py` | Unit tests with mocks (23 tests) | Fast, no Docker |
| `test_upload_webdav_integration.py` | WebDAV integration with real server (10 tests) | Medium, requires Docker |
| `test_e2e_full_stack.py` | Full end-to-end with all infrastructure (12+ tests) | Slow, requires Docker |
| `fixtures_integration.py` | Reusable fixtures for real services | N/A |
### Test Markers
Tests are organized using pytest markers:
```python
@pytest.mark.unit # Fast unit tests with mocks
@pytest.mark.integration # Integration tests with some real services
@pytest.mark.e2e # Full end-to-end with complete stack
@pytest.mark.requires_docker # Requires Docker to run
@pytest.mark.slow # Takes significant time (>30s)
```
## Running Tests
### Quick Start - Unit Tests Only (No Docker)
```bash
# Run only fast unit tests (mocked, no containers)
pytest -m unit -v
# Run WebDAV unit tests specifically
pytest tests/test_upload_webdav_comprehensive.py -v
```
### Integration Tests - WebDAV Only
```bash
# Run WebDAV integration tests (spins up WebDAV container)
pytest tests/test_upload_webdav_integration.py -v
# Run specific test
pytest tests/test_upload_webdav_integration.py::TestWebDAVIntegration::test_upload_file_to_real_webdav_server -v
```
### Full End-to-End Tests - Complete Infrastructure
```bash
# Run all e2e tests (spins up all infrastructure)
pytest -m e2e -v
# Run specific infrastructure test
pytest tests/test_e2e_full_stack.py::TestFullInfrastructure::test_complete_stack_available -v
# Run with real Redis and Celery
pytest tests/test_e2e_full_stack.py::TestEndToEndWithRedis -v
```
### Run Everything
```bash
# Run all tests (unit + integration + e2e)
pytest tests/test_upload_webdav*.py tests/test_e2e*.py -v
# Skip slow tests
pytest -m "not slow" -v
# Run only Docker-based tests
pytest -m requires_docker -v
```
## Infrastructure Fixtures
### Available Fixtures
#### `postgres_container`
Starts PostgreSQL 15 in Alpine container.
```python
def test_with_postgres(postgres_container):
db_url = postgres_container["url"]
# Use real PostgreSQL
```
#### `redis_container`
Starts Redis 7 for Celery broker/backend.
```python
def test_with_redis(redis_container):
redis_url = redis_container["url"]
# Queue actual tasks
```
#### `gotenberg_container`
Starts Gotenberg for PDF conversion.
```python
def test_with_gotenberg(gotenberg_container):
url = gotenberg_container["url"]
# Convert documents
```
#### `webdav_container`
Starts WebDAV server (bytemark/webdav).
```python
def test_with_webdav(webdav_container):
# Upload files, verify on server
url = webdav_container["url"]
username = webdav_container["username"] # "testuser"
password = webdav_container["password"] # "testpass"
```
#### `sftp_container`
Starts SFTP server (atmoz/sftp).
```python
def test_with_sftp(sftp_container):
# Upload via SFTP, verify
host = sftp_container["host"]
port = sftp_container["port"]
```
#### `minio_container`
Starts MinIO (S3-compatible).
```python
def test_with_s3(minio_container):
# Use boto3 with MinIO
access_key = minio_container["access_key"]
secret_key = minio_container["secret_key"]
```
#### `ftp_container`
Starts FTP server (stilliard/pure-ftpd).
```python
def test_with_ftp(ftp_container):
# Upload via FTP
```
#### `full_infrastructure`
Combined fixture with ALL services.
```python
def test_production_like(full_infrastructure):
infra = full_infrastructure
# Access: postgres, redis, gotenberg, webdav, sftp, minio
```
#### `celery_app` and `celery_worker`
Real Celery application with worker.
```python
def test_celery_tasks(celery_app, celery_worker):
# Queue actual tasks that get processed
result = my_task.delay(arg1, arg2)
result.get(timeout=30) # Wait for worker to process
```
## Test Scenarios
### 1. Simple WebDAV Upload Test
```python
@pytest.mark.integration
@pytest.mark.requires_docker
def test_upload_to_webdav(webdav_container, sample_text_file):
"""Upload file to real WebDAV server and verify."""
from app.tasks.upload_to_webdav import upload_to_webdav
with patch("app.tasks.upload_to_webdav.settings") as mock_settings:
mock_settings.webdav_url = webdav_container["url"] + "/"
mock_settings.webdav_username = webdav_container["username"]
mock_settings.webdav_password = webdav_container["password"]
# Execute upload
result = upload_to_webdav.apply(args=[sample_text_file]).get()
# Verify on server
response = requests.get(
f"{webdav_container['url']}/test.txt",
auth=(webdav_container["username"], webdav_container["password"])
)
assert response.status_code == 200
```
### 2. End-to-End with Redis and Celery
```python
@pytest.mark.e2e
def test_async_upload(redis_container, webdav_container, celery_worker, sample_text_file):
"""Queue task in Redis, worker executes, uploads to WebDAV."""
from app.tasks.upload_to_webdav import upload_to_webdav
# Queue task (goes to Redis)
result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for worker to process
while not result.ready():
time.sleep(0.5)
# Verify result
assert result.get()["status"] == "Completed"
```
### 3. Full Production Pipeline
```python
@pytest.mark.e2e
@pytest.mark.slow
def test_complete_pipeline(full_infrastructure, celery_worker, db_session_real):
"""
Test complete workflow:
1. Store in PostgreSQL
2. Queue task in Redis
3. Worker processes
4. Upload to WebDAV
5. Verify all steps
"""
# See test_document_processing_pipeline in test_e2e_full_stack.py
```
## Performance Notes
### Container Startup Times
| Container | Startup Time | Notes |
|-----------|--------------|-------|
| PostgreSQL | ~2-3s | Fast |
| Redis | ~1-2s | Very fast |
| WebDAV | ~2s | Fast |
| SFTP | ~3s | Moderate |
| MinIO | ~3-4s | Moderate |
| FTP | ~3s | Moderate |
| Gotenberg | ~5-8s | Slower (Chromium startup) |
### Test Execution Times
- **Unit tests**: <1s per test
- **Single integration test**: 2-5s (with container)
- **E2E with full stack**: 10-30s per test
- **Full suite**: 2-5 minutes
### Resource Usage
- **Memory**: ~100MB per container
- **Disk**: ~500MB total for images
- **CPU**: Varies, mostly idle
## Debugging
### View Container Logs
```python
def test_debug(webdav_container):
container = webdav_container["container"]
logs = container.get_logs()
print(logs)
```
### Keep Containers Running
Set a breakpoint after test to inspect:
```python
def test_inspect(webdav_container):
result = upload_file()
import pdb; pdb.set_trace() # Container still running here
# Manually inspect: docker ps, docker logs, etc.
```
### Check Container Health
```bash
# While tests are running
docker ps # See running containers
docker logs <container_id> # View logs
docker exec -it <container_id> sh # Shell into container
```
## Troubleshooting
### "Docker not found"
```bash
# Install Docker
# On Ubuntu/Debian:
sudo apt-get install docker.io
sudo usermod -aG docker $USER # Add user to docker group
# Logout and login again
```
### "Permission denied" for Docker
```bash
# Add user to docker group
sudo usermod -aG docker $USER
# Logout/login or:
newgrp docker
```
### "Port already in use"
Containers use random ports, but if issues persist:
```bash
docker ps # Check what's running
docker stop $(docker ps -q) # Stop all containers
```
### Tests hang or timeout
- Increase timeout values in test code
- Check Docker has enough resources (memory/CPU)
- Check network connectivity
### Containers not cleaning up
```bash
# Manual cleanup
docker ps -a | grep testcontainers | awk '{print $1}' | xargs docker rm -f
docker volume prune -f
```
## CI/CD Integration
### GitHub Actions Example
```yaml
name: Integration Tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
docker:
image: docker:latest
options: --privileged
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -r requirements-dev.txt
- name: Run integration tests
run: |
pytest -m "integration or e2e" -v --tb=short
```
## Best Practices
### 1. Use Appropriate Markers
```python
# Fast test - use unit
@pytest.mark.unit
def test_validation():
...
# Needs Docker - mark it
@pytest.mark.requires_docker
def test_upload():
...
# Slow test - mark it
@pytest.mark.slow
def test_large_file():
...
```
### 2. Reuse Fixtures (Session Scope)
```python
# Good - starts once for all tests in class
@pytest.fixture(scope="session")
def postgres_container():
...
# Bad - starts/stops for each test
@pytest.fixture(scope="function")
def postgres_container():
...
```
### 3. Clean Up Resources
```python
def test_with_temp_files(tmp_path):
# tmp_path auto-cleans up
file = tmp_path / "test.txt"
...
```
### 4. Use Timeouts
```python
# Always set timeouts for container operations
response = requests.get(url, timeout=5)
result.get(timeout=30)
```
### 5. Verify Actual Behavior
```python
# Don't just check return values
result = upload_file()
assert result["status"] == "success"
# Also verify the file actually exists on the server!
assert file_exists_on_server(filename)
```
## Coverage
Running integration tests significantly improves code coverage:
| Test Type | upload_to_webdav.py Coverage |
|-----------|------------------------------|
| Unit only | ~20% |
| + Integration | ~80% |
| + E2E | ~95%+ |
```bash
# Run with coverage
pytest tests/test_upload_webdav*.py --cov=app/tasks/upload_to_webdav --cov-report=html
# View report
open htmlcov/index.html
```
## Contributing
When adding new upload destinations:
1. **Add unit tests** (with mocks) in `test_upload_<destination>_comprehensive.py`
2. **Add container fixture** in `fixtures_integration.py`
3. **Add integration tests** in `test_upload_<destination>_integration.py`
4. **Add e2e scenarios** in `test_e2e_full_stack.py`
See WebDAV tests as reference implementation.
## Summary
| Test Level | Fixtures | Speed | Realism | Use Case |
|------------|----------|-------|---------|----------|
| Unit | Mocks | Fast (ms) | Low | Development, TDD |
| Integration | 1-2 containers | Medium (s) | Medium | Feature testing |
| E2E | Full stack | Slow (10s+) | High | Pre-production validation |
Choose the appropriate level based on what you're testing!
tests/conftest.py
+1
View File
@@ -178,3 +178,4 @@ def pytest_configure(config):
config.addinivalue_line("markers", "requires_db: Tests requiring database")
config.addinivalue_line("markers", "requires_redis: Tests requiring Redis")
config.addinivalue_line("markers", "requires_docker: Tests requiring Docker")
config.addinivalue_line("markers", "e2e: End-to-end tests with full infrastructure")
tests/fixtures_integration.py
+319
View File
@@ -0,0 +1,319 @@
"""
Full integration test infrastructure using real services.
This module provides fixtures for spinning up real infrastructure components:
- PostgreSQL database (instead of SQLite in-memory)
- Redis (for Celery broker/backend)
- Gotenberg (for PDF conversion)
- WebDAV server (upload target)
- SFTP server (upload target)
- MinIO (S3-compatible storage)
These tests exercise the full application stack end-to-end.
"""
import os
import time
import pytest
from typing import Generator
from pathlib import Path
# Import testcontainers
pytest.importorskip("testcontainers", reason="testcontainers not installed")
from testcontainers.core.container import DockerContainer
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from testcontainers.minio import MinioContainer
@pytest.fixture(scope="session")
def postgres_container() -> Generator:
"""
Start a real PostgreSQL database container for testing.
This replaces the in-memory SQLite used in unit tests.
"""
with PostgresContainer("postgres:15-alpine") as postgres:
# Wait for PostgreSQL to be ready
time.sleep(2)
# Set environment variable for the app to use
os.environ["DATABASE_URL"] = postgres.get_connection_url()
yield {
"container": postgres,
"url": postgres.get_connection_url(),
"host": postgres.get_container_host_ip(),
"port": postgres.get_exposed_port(5432),
"username": postgres.username,
"password": postgres.password,
"database": postgres.dbname,
}
@pytest.fixture(scope="session")
def redis_container() -> Generator:
"""
Start a real Redis container for Celery broker/backend.
This provides actual message queueing and task result storage.
"""
with RedisContainer("redis:7-alpine") as redis:
# Wait for Redis to be ready
time.sleep(2)
# Build Redis URL manually
host = redis.get_container_host_ip()
port = redis.get_exposed_port(6379)
redis_url = f"redis://{host}:{port}/0"
# Set environment variables for the app
os.environ["REDIS_URL"] = redis_url
os.environ["CELERY_BROKER_URL"] = redis_url
os.environ["CELERY_RESULT_BACKEND"] = redis_url
yield {
"container": redis,
"url": redis_url,
"host": host,
"port": port,
}
@pytest.fixture(scope="session")
def gotenberg_container() -> Generator:
"""
Start a real Gotenberg container for PDF conversion.
This provides actual document conversion capabilities.
"""
container = DockerContainer("gotenberg/gotenberg:8")
container.with_exposed_ports(3000)
container.with_command(
"gotenberg --chromium-disable-javascript=false --chromium-allow-list=file:///.*"
)
container.start()
time.sleep(5) # Gotenberg takes a bit longer to start
host = container.get_container_host_ip()
port = container.get_exposed_port(3000)
gotenberg_url = f"http://{host}:{port}"
# Set environment variable
os.environ["GOTENBERG_URL"] = gotenberg_url
yield {
"container": container,
"url": gotenberg_url,
"host": host,
"port": port,
}
container.stop()
@pytest.fixture(scope="session")
def webdav_container() -> Generator:
"""
Start a real WebDAV server for upload testing.
"""
container = DockerContainer("bytemark/webdav:latest")
container.with_exposed_ports(80)
container.with_env("AUTH_TYPE", "Basic")
container.with_env("USERNAME", "testuser")
container.with_env("PASSWORD", "testpass")
container.start()
time.sleep(2)
host = container.get_container_host_ip()
port = container.get_exposed_port(80)
yield {
"container": container,
"url": f"http://{host}:{port}",
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
}
container.stop()
@pytest.fixture(scope="session")
def sftp_container() -> Generator:
"""
Start a real SFTP server for upload testing.
Uses atmoz/sftp which provides a simple SSH/SFTP server.
"""
container = DockerContainer("atmoz/sftp:latest")
container.with_exposed_ports(22)
# Create user: username:password:uid:gid:directory
container.with_command("testuser:testpass:1001:1001:upload")
container.start()
time.sleep(3) # SFTP server needs time to initialize
host = container.get_container_host_ip()
port = container.get_exposed_port(22)
yield {
"container": container,
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
"folder": "/home/testuser/upload",
}
container.stop()
@pytest.fixture(scope="session")
def minio_container() -> Generator:
"""
Start a real MinIO container (S3-compatible storage).
MinIO provides S3-compatible API for testing S3 uploads.
"""
with MinioContainer() as minio:
time.sleep(2)
# MinIO uses random credentials, get them
access_key = minio.access_key
secret_key = minio.secret_key
yield {
"container": minio,
"url": minio.get_config()["endpoint"],
"access_key": access_key,
"secret_key": secret_key,
"region": "us-east-1",
}
@pytest.fixture(scope="session")
def ftp_container() -> Generator:
"""
Start a real FTP server for upload testing.
Uses stilliard/pure-ftpd which provides a simple FTP server.
"""
container = DockerContainer("stilliard/pure-ftpd:latest")
container.with_exposed_ports(21, 30000, 30001, 30002, 30003, 30004)
container.with_env("PUBLICHOST", "localhost")
container.with_env("FTP_USER_NAME", "testuser")
container.with_env("FTP_USER_PASS", "testpass")
container.with_env("FTP_USER_HOME", "/home/testuser")
container.start()
time.sleep(3)
host = container.get_container_host_ip()
port = container.get_exposed_port(21)
yield {
"container": container,
"host": host,
"port": port,
"username": "testuser",
"password": "testpass",
"folder": "/",
}
container.stop()
@pytest.fixture(scope="session")
def full_infrastructure(
postgres_container,
redis_container,
gotenberg_container,
webdav_container,
sftp_container,
minio_container,
):
"""
Combined fixture that provides all infrastructure components.
Use this fixture when you need the complete application stack.
"""
return {
"postgres": postgres_container,
"redis": redis_container,
"gotenberg": gotenberg_container,
"webdav": webdav_container,
"sftp": sftp_container,
"minio": minio_container,
}
@pytest.fixture(scope="session")
def celery_app(redis_container):
"""
Create a Celery app configured to use the real Redis container.
This allows testing actual task queueing and execution.
"""
from app.celery_app import celery
# Update Celery configuration to use test Redis
celery.conf.update(
broker_url=redis_container["url"],
result_backend=redis_container["url"],
task_always_eager=False, # Actually queue tasks (don't execute immediately)
task_eager_propagates=True,
result_expires=3600,
)
return celery
@pytest.fixture(scope="session")
def celery_worker(celery_app, redis_container):
"""
Start a real Celery worker for processing tasks.
This runs tasks asynchronously like in production.
"""
from celery.contrib.testing.worker import start_worker
# Start worker in test mode
with start_worker(
celery_app,
perform_ping_check=False,
loglevel="info",
concurrency=2,
) as worker:
yield worker
@pytest.fixture
def db_session_real(postgres_container):
"""
Create a database session using the real PostgreSQL container.
This replaces the in-memory SQLite session for integration tests.
"""
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.database import Base
# Create engine using PostgreSQL container
engine = create_engine(postgres_container["url"])
# Create all tables
Base.metadata.create_all(bind=engine)
# Create session
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = SessionLocal()
try:
yield session
finally:
session.close()
# Clean up tables after test
Base.metadata.drop_all(bind=engine)
tests/test_e2e_full_stack.py
+588
View File
@@ -0,0 +1,588 @@
"""
End-to-end integration tests using real infrastructure.
These tests spin up actual services (PostgreSQL, Redis, Gotenberg, WebDAV, SFTP, MinIO)
and test the complete application workflow from API request to file upload.
"""
import os
import time
import pytest
import requests
from unittest.mock import patch
# Import testcontainers requirement
pytest.importorskip("testcontainers", reason="testcontainers not installed")
from tests.fixtures_integration import (
postgres_container,
redis_container,
gotenberg_container,
webdav_container,
sftp_container,
minio_container,
full_infrastructure,
celery_app,
celery_worker,
db_session_real,
)
@pytest.mark.integration
@pytest.mark.requires_docker
@pytest.mark.e2e
class TestEndToEndWithRedis:
"""
End-to-end tests with real Redis and Celery workers.
These tests verify the complete task queueing and execution workflow.
"""
def test_webdav_upload_with_redis_and_celery(
self,
redis_container,
webdav_container,
celery_app,
celery_worker,
sample_text_file,
):
"""
Test complete workflow: Queue task in Redis → Celery worker executes → Upload to WebDAV.
This is the closest to production - actual message queueing and async execution.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
# Configure to use real WebDAV server
mock_settings.webdav_url = webdav_container["url"] + "/"
mock_settings.webdav_username = webdav_container["username"]
mock_settings.webdav_password = webdav_container["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Queue the task (it goes to Redis)
result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for task to complete (worker picks it up from Redis)
timeout = 30
start_time = time.time()
while not result.ready():
if time.time() - start_time > timeout:
pytest.fail(f"Task did not complete within {timeout} seconds")
time.sleep(0.5)
# Get the result
task_result = result.get(timeout=10)
# Verify task completed successfully
assert task_result["status"] == "Completed"
assert task_result["file"] == sample_text_file
# Verify file was actually uploaded to WebDAV server
filename = os.path.basename(sample_text_file)
file_url = f"{webdav_container['url']}/{filename}"
response = requests.get(
file_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
)
assert response.status_code == 200
# Verify content matches
with open(sample_text_file, "rb") as f:
assert response.content == f.read()
def test_task_queuing_in_redis(
self,
redis_container,
celery_app,
):
"""
Test that tasks are properly queued in Redis.
This verifies the Redis broker is working correctly.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
import redis
# Connect to Redis directly
r = redis.from_url(redis_container["url"])
# Check Redis is accessible
assert r.ping()
# Get current queue length
initial_queue_length = r.llen("celery")
# Queue a task (don't execute, just verify queueing)
with patch("app.tasks.upload_to_webdav.settings") as mock_settings:
mock_settings.webdav_url = "http://test.com"
mock_settings.webdav_username = "user"
mock_settings.webdav_password = "pass"
# This will queue the task in Redis
result = upload_to_webdav.apply_async(
args=["/tmp/test.txt"],
kwargs={"file_id": 1}
)
# Verify task ID was generated
assert result.id is not None
def test_multiple_tasks_parallel_execution(
self,
redis_container,
webdav_container,
celery_app,
celery_worker,
tmp_path,
):
"""
Test multiple tasks executing in parallel through Redis/Celery.
This tests concurrent task processing.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
# Create multiple test files
files = []
for i in range(5):
test_file = tmp_path / f"test_{i}.txt"
test_file.write_text(f"Test file {i}")
files.append(str(test_file))
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
mock_settings.webdav_url = webdav_container["url"] + "/"
mock_settings.webdav_username = webdav_container["username"]
mock_settings.webdav_password = webdav_container["password"]
mock_settings.webdav_folder = "parallel-test"
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create folder on WebDAV server
folder_url = f"{webdav_container['url']}/parallel-test"
requests.request(
"MKCOL",
folder_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
)
# Queue all tasks
results = []
for idx, file_path in enumerate(files):
result = upload_to_webdav.delay(file_path, file_id=idx + 100)
results.append((result, file_path))
# Wait for all tasks to complete
timeout = 60
start_time = time.time()
all_ready = False
while not all_ready:
if time.time() - start_time > timeout:
pytest.fail("Tasks did not complete within timeout")
all_ready = all(r.ready() for r, _ in results)
time.sleep(0.5)
# Verify all tasks succeeded
for result, file_path in results:
task_result = result.get(timeout=5)
assert task_result["status"] == "Completed"
# Verify file on server
filename = os.path.basename(file_path)
file_url = f"{webdav_container['url']}/parallel-test/{filename}"
response = requests.get(
file_url,
auth=(webdav_container["username"], webdav_container["password"]),
timeout=5
)
assert response.status_code == 200
def test_task_retry_on_failure(
self,
redis_container,
celery_app,
celery_worker,
sample_text_file,
):
"""
Test that tasks retry on failure using Redis.
This verifies the retry mechanism works with real broker.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"), \
patch("app.tasks.upload_to_webdav.requests.put") as mock_put:
mock_settings.webdav_url = "http://test.com/"
mock_settings.webdav_username = "user"
mock_settings.webdav_password = "pass"
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# First attempt fails with 500
mock_response_fail = requests.Response()
mock_response_fail.status_code = 500
mock_response_fail._content = b"Server Error"
# Second attempt succeeds
mock_response_success = requests.Response()
mock_response_success.status_code = 201
# Configure mock to fail once, then succeed
mock_put.side_effect = [mock_response_fail, mock_response_success]
# Queue task
result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for completion (including retry)
timeout = 30
start_time = time.time()
while not result.ready():
if time.time() - start_time > timeout:
break # Task might still be retrying
time.sleep(0.5)
@pytest.mark.integration
@pytest.mark.requires_docker
@pytest.mark.e2e
class TestFullInfrastructure:
"""
Tests using the complete infrastructure stack.
PostgreSQL + Redis + Gotenberg + Upload targets (WebDAV/SFTP/MinIO)
"""
def test_complete_stack_available(self, full_infrastructure):
"""
Verify all infrastructure components are running.
"""
infra = full_infrastructure
# Check PostgreSQL
assert infra["postgres"]["url"] is not None
assert "postgresql" in infra["postgres"]["url"]
# Check Redis
assert infra["redis"]["url"] is not None
import redis
r = redis.from_url(infra["redis"]["url"])
assert r.ping()
# Check Gotenberg
assert infra["gotenberg"]["url"] is not None
response = requests.get(f"{infra['gotenberg']['url']}/health", timeout=5)
assert response.status_code == 200
# Check WebDAV
assert infra["webdav"]["url"] is not None
# Check SFTP
assert infra["sftp"]["host"] is not None
assert infra["sftp"]["port"] is not None
# Check MinIO
assert infra["minio"]["access_key"] is not None
def test_database_with_real_postgres(self, postgres_container, db_session_real):
"""
Test database operations with real PostgreSQL instead of SQLite.
"""
from app.models import FileRecord
# Create a file record
file_record = FileRecord(
filename="test.pdf",
file_path="/tmp/test.pdf",
file_size=1024,
mime_type="application/pdf",
)
db_session_real.add(file_record)
db_session_real.commit()
# Verify it was saved
assert file_record.id is not None
# Query it back
queried = db_session_real.query(FileRecord).filter_by(filename="test.pdf").first()
assert queried is not None
assert queried.filename == "test.pdf"
assert queried.file_size == 1024
def test_upload_to_multiple_targets(
self,
full_infrastructure,
celery_app,
celery_worker,
sample_text_file,
):
"""
Test uploading to multiple targets in parallel (WebDAV + SFTP).
This simulates the send_to_all_destinations workflow.
"""
from app.tasks.upload_to_webdav import upload_to_webdav
infra = full_infrastructure
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
mock_settings.webdav_url = infra["webdav"]["url"] + "/"
mock_settings.webdav_username = infra["webdav"]["username"]
mock_settings.webdav_password = infra["webdav"]["password"]
mock_settings.webdav_folder = ""
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Upload to WebDAV
webdav_result = upload_to_webdav.delay(sample_text_file, file_id=1)
# Wait for completion
timeout = 30
start_time = time.time()
while not webdav_result.ready():
if time.time() - start_time > timeout:
pytest.fail("Task timeout")
time.sleep(0.5)
# Verify WebDAV upload
result = webdav_result.get(timeout=10)
assert result["status"] == "Completed"
# Verify file on WebDAV server
filename = os.path.basename(sample_text_file)
file_url = f"{infra['webdav']['url']}/{filename}"
response = requests.get(
file_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
)
assert response.status_code == 200
def test_gotenberg_pdf_conversion(self, gotenberg_container, tmp_path):
"""
Test PDF conversion using real Gotenberg service.
This verifies document processing capabilities.
"""
# Create a simple HTML file
html_file = tmp_path / "test.html"
html_file.write_text("""
<!DOCTYPE html>
<html>
<head><title>Test Document</title></head>
<body><h1>Integration Test</h1><p>This is a test document.</p></body>
</html>
""")
# Convert to PDF using Gotenberg
with open(html_file, "rb") as f:
files = {"files": f}
response = requests.post(
f"{gotenberg_container['url']}/forms/chromium/convert/html",
files=files,
timeout=30
)
assert response.status_code == 200
assert response.headers["Content-Type"] == "application/pdf"
assert len(response.content) > 0
assert response.content.startswith(b"%PDF")
def test_minio_s3_upload(self, minio_container, sample_text_file):
"""
Test S3-compatible upload using real MinIO.
This tests S3 upload functionality with actual storage.
"""
import boto3
from botocore.client import Config
# Create S3 client configured for MinIO
s3_client = boto3.client(
"s3",
endpoint_url=minio_container["url"],
aws_access_key_id=minio_container["access_key"],
aws_secret_access_key=minio_container["secret_key"],
config=Config(signature_version="s3v4"),
region_name=minio_container["region"],
)
# Create bucket
bucket_name = "test-bucket"
s3_client.create_bucket(Bucket=bucket_name)
# Upload file
filename = os.path.basename(sample_text_file)
with open(sample_text_file, "rb") as f:
s3_client.upload_fileobj(f, bucket_name, filename)
# Verify upload
response = s3_client.list_objects_v2(Bucket=bucket_name)
assert "Contents" in response
assert len(response["Contents"]) == 1
assert response["Contents"][0]["Key"] == filename
# Download and verify content
download_path = os.path.join(os.path.dirname(sample_text_file), "downloaded.txt")
s3_client.download_file(bucket_name, filename, download_path)
with open(sample_text_file, "rb") as original, \
open(download_path, "rb") as downloaded:
assert original.read() == downloaded.read()
def test_sftp_upload(self, sftp_container, sample_text_file):
"""
Test SFTP upload using real SFTP server.
"""
import paramiko
# Create SFTP client
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# Connect to SFTP server
ssh.connect(
hostname=sftp_container["host"],
port=int(sftp_container["port"]),
username=sftp_container["username"],
password=sftp_container["password"],
timeout=10,
)
sftp = ssh.open_sftp()
try:
# Upload file
filename = os.path.basename(sample_text_file)
remote_path = f"{sftp_container['folder']}/{filename}"
sftp.put(sample_text_file, remote_path)
# Verify upload
stat = sftp.stat(remote_path)
assert stat.st_size == os.path.getsize(sample_text_file)
# Download and verify content
download_path = os.path.join(os.path.dirname(sample_text_file), "sftp_downloaded.txt")
sftp.get(remote_path, download_path)
with open(sample_text_file, "rb") as original, \
open(download_path, "rb") as downloaded:
assert original.read() == downloaded.read()
finally:
sftp.close()
ssh.close()
@pytest.mark.integration
@pytest.mark.requires_docker
@pytest.mark.e2e
@pytest.mark.slow
class TestProductionLikeScenarios:
"""
Production-like end-to-end scenarios testing complete workflows.
"""
def test_document_processing_pipeline(
self,
full_infrastructure,
celery_app,
celery_worker,
db_session_real,
tmp_path,
):
"""
Test the complete document processing pipeline:
1. Upload document via API
2. Store metadata in PostgreSQL
3. Queue processing tasks in Redis
4. Process document (mock OCR/metadata extraction)
5. Upload to WebDAV via Celery
6. Verify all steps completed
This is the closest to real production usage.
"""
from app.models import FileRecord
from app.tasks.upload_to_webdav import upload_to_webdav
# Create test document
test_doc = tmp_path / "invoice.pdf"
test_doc.write_bytes(b"%PDF-1.4\n%Test PDF\n%%EOF")
# Step 1: Store in database
file_record = FileRecord(
filename="invoice.pdf",
file_path=str(test_doc),
file_size=test_doc.stat().st_size,
mime_type="application/pdf",
)
db_session_real.add(file_record)
db_session_real.commit()
assert file_record.id is not None
db_file_id = file_record.id
# Step 2: Queue upload task
infra = full_infrastructure
with patch("app.tasks.upload_to_webdav.settings") as mock_settings, \
patch("app.tasks.upload_to_webdav.log_task_progress"):
mock_settings.webdav_url = infra["webdav"]["url"] + "/"
mock_settings.webdav_username = infra["webdav"]["username"]
mock_settings.webdav_password = infra["webdav"]["password"]
mock_settings.webdav_folder = "processed"
mock_settings.webdav_verify_ssl = False
mock_settings.http_request_timeout = 30
# Create folder
folder_url = f"{infra['webdav']['url']}/processed"
requests.request(
"MKCOL",
folder_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
)
# Step 3: Queue upload
result = upload_to_webdav.delay(str(test_doc), file_id=db_file_id)
# Step 4: Wait for processing
timeout = 30
start_time = time.time()
while not result.ready():
if time.time() - start_time > timeout:
pytest.fail("Pipeline timeout")
time.sleep(0.5)
# Step 5: Verify completion
task_result = result.get(timeout=10)
assert task_result["status"] == "Completed"
# Step 6: Verify file on WebDAV
file_url = f"{infra['webdav']['url']}/processed/invoice.pdf"
response = requests.get(
file_url,
auth=(infra["webdav"]["username"], infra["webdav"]["password"]),
timeout=5
)
assert response.status_code == 200
assert response.content == test_doc.read_bytes()