copilot-swe-agent[bot]
ff9a3ff49f
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
323 lines
8.6 KiB
Python
323 lines
8.6 KiB
Python
"""
|
|
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
|
|
from pathlib import Path
|
|
from typing import Generator
|
|
|
|
import pytest
|
|
|
|
# Import testcontainers
|
|
pytest.importorskip("testcontainers", reason="testcontainers not installed")
|
|
from testcontainers.core.container import DockerContainer
|
|
from testcontainers.minio import MinioContainer
|
|
from testcontainers.postgres import PostgresContainer
|
|
from testcontainers.redis import RedisContainer
|
|
|
|
_TEST_CREDENTIAL = "testpass" # noqa: S105
|
|
|
|
|
|
@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", _TEST_CREDENTIAL)
|
|
|
|
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": _TEST_CREDENTIAL,
|
|
}
|
|
|
|
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(f"testuser:{_TEST_CREDENTIAL}: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": _TEST_CREDENTIAL,
|
|
"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", _TEST_CREDENTIAL)
|
|
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": _TEST_CREDENTIAL,
|
|
"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)
|