forked from username/flaskpaste
test: add container integration tests
Tests verify: - Container image builds successfully - Health endpoint responds - Paste creation/retrieval works - Security headers present - Non-root execution - Gunicorn workers running Skipped by default, run with: FLASKPASTE_INTEGRATION=1 pytest tests/test_container_integration.py
This commit is contained in:
Username
365
tests/test_container_integration.py
Normal file
365
tests/test_container_integration.py
Normal file
@@ -0,0 +1,365 @@
|
||||
"""Container integration tests for FlaskPaste.
|
||||
|
||||
These tests verify the container build and runtime behavior.
|
||||
They require podman or docker to be available on the system.
|
||||
|
||||
Run with:
|
||||
FLASKPASTE_INTEGRATION=1 pytest tests/test_container_integration.py -v
|
||||
|
||||
Skip in normal test runs:
|
||||
pytest tests/ # automatically skips these tests
|
||||
|
||||
Note: S603 (subprocess call) warnings are suppressed as these are
|
||||
legitimate container management operations in test code.
|
||||
"""
|
||||
# ruff: noqa: S603
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
# Configuration
|
||||
CONTAINER_NAME = "flaskpaste-test"
|
||||
IMAGE_NAME = "flaskpaste:test"
|
||||
HOST_PORT = 5099 # Use non-standard port to avoid conflicts
|
||||
CONTAINER_PORT = 5000
|
||||
BASE_URL = f"http://localhost:{HOST_PORT}"
|
||||
BUILD_TIMEOUT = 180 # seconds (increased for slow systems)
|
||||
START_TIMEOUT = 30 # seconds
|
||||
REQUEST_TIMEOUT = 5 # seconds
|
||||
|
||||
|
||||
def get_container_runtime() -> str | None:
|
||||
"""Detect available container runtime (podman or docker)."""
|
||||
for runtime in ["podman", "docker"]:
|
||||
if shutil.which(runtime):
|
||||
return runtime
|
||||
return None
|
||||
|
||||
|
||||
def should_run_integration_tests() -> bool:
|
||||
"""Check if integration tests should run.
|
||||
|
||||
Integration tests only run when FLASKPASTE_INTEGRATION=1 environment
|
||||
variable is set AND a container runtime is available.
|
||||
"""
|
||||
if not os.environ.get("FLASKPASTE_INTEGRATION"):
|
||||
return False
|
||||
return get_container_runtime() is not None
|
||||
|
||||
|
||||
# Skip all tests unless explicitly enabled
|
||||
pytestmark = pytest.mark.skipif(
|
||||
not should_run_integration_tests(),
|
||||
reason="Set FLASKPASTE_INTEGRATION=1 to run container integration tests",
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def container_runtime():
|
||||
"""Return the available container runtime."""
|
||||
runtime = get_container_runtime()
|
||||
if runtime is None:
|
||||
pytest.skip("No container runtime available")
|
||||
return runtime
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def project_root():
|
||||
"""Return the project root directory."""
|
||||
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def built_image(container_runtime, project_root):
|
||||
"""Build the container image for testing."""
|
||||
# Build the image
|
||||
result = subprocess.run(
|
||||
[container_runtime, "build", "-t", IMAGE_NAME, "."],
|
||||
cwd=project_root,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=BUILD_TIMEOUT,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
pytest.fail(f"Failed to build container image:\n{result.stderr}")
|
||||
|
||||
yield IMAGE_NAME
|
||||
|
||||
# Cleanup: remove the test image
|
||||
subprocess.run(
|
||||
[container_runtime, "rmi", "-f", IMAGE_NAME],
|
||||
capture_output=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def running_container(container_runtime, built_image):
|
||||
"""Start a container for testing and yield the base URL."""
|
||||
# Stop any existing container with same name
|
||||
subprocess.run(
|
||||
[container_runtime, "rm", "-f", CONTAINER_NAME],
|
||||
capture_output=True,
|
||||
timeout=30,
|
||||
)
|
||||
|
||||
# Start container
|
||||
result = subprocess.run(
|
||||
[
|
||||
container_runtime,
|
||||
"run",
|
||||
"-d",
|
||||
"--name",
|
||||
CONTAINER_NAME,
|
||||
"-p",
|
||||
f"{HOST_PORT}:{CONTAINER_PORT}",
|
||||
"-e",
|
||||
"FLASK_ENV=production",
|
||||
built_image,
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
pytest.fail(f"Failed to start container:\n{result.stderr}")
|
||||
|
||||
# Wait for container to be healthy
|
||||
start_time = time.time()
|
||||
healthy = False
|
||||
last_error = None
|
||||
|
||||
while time.time() - start_time < START_TIMEOUT:
|
||||
try:
|
||||
response = requests.get(f"{BASE_URL}/health", timeout=REQUEST_TIMEOUT)
|
||||
if response.status_code == 200:
|
||||
healthy = True
|
||||
break
|
||||
except requests.RequestException as e:
|
||||
last_error = e
|
||||
time.sleep(1)
|
||||
|
||||
if not healthy:
|
||||
# Get container logs for debugging
|
||||
logs = subprocess.run(
|
||||
[container_runtime, "logs", CONTAINER_NAME],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
subprocess.run(
|
||||
[container_runtime, "rm", "-f", CONTAINER_NAME], capture_output=True, timeout=30
|
||||
)
|
||||
pytest.fail(
|
||||
f"Container failed to become healthy within {START_TIMEOUT}s.\n"
|
||||
f"Last error: {last_error}\n"
|
||||
f"Container logs:\n{logs.stdout}\n{logs.stderr}"
|
||||
)
|
||||
|
||||
yield BASE_URL
|
||||
|
||||
# Cleanup: stop and remove container
|
||||
subprocess.run(
|
||||
[container_runtime, "rm", "-f", CONTAINER_NAME], capture_output=True, timeout=30
|
||||
)
|
||||
|
||||
|
||||
class TestContainerBuild:
|
||||
"""Test container image build process."""
|
||||
|
||||
def test_image_builds_successfully(self, built_image):
|
||||
"""Container image should build without errors."""
|
||||
assert built_image == IMAGE_NAME
|
||||
|
||||
def test_image_has_expected_labels(self, container_runtime, built_image):
|
||||
"""Container image should have expected labels."""
|
||||
result = subprocess.run(
|
||||
[container_runtime, "inspect", "--format", "{{.Config.Labels}}", built_image],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
# Check for expected label content
|
||||
labels = result.stdout
|
||||
assert "flaskpaste" in labels.lower() or "pastebin" in labels.lower()
|
||||
|
||||
def test_image_has_healthcheck(self, container_runtime, built_image):
|
||||
"""Container image should define a healthcheck."""
|
||||
result = subprocess.run(
|
||||
[container_runtime, "inspect", "--format", "{{.Config.Healthcheck}}", built_image],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
# Should have healthcheck defined
|
||||
assert "CMD" in result.stdout or "python" in result.stdout
|
||||
|
||||
|
||||
class TestContainerRuntime:
|
||||
"""Test container runtime behavior."""
|
||||
|
||||
def test_health_endpoint_responds(self, running_container):
|
||||
"""Health endpoint should respond with 200."""
|
||||
response = requests.get(f"{running_container}/health", timeout=REQUEST_TIMEOUT)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data.get("status") == "healthy"
|
||||
|
||||
def test_paste_creation(self, running_container):
|
||||
"""Should be able to create a paste."""
|
||||
response = requests.post(
|
||||
running_container,
|
||||
data=b"Test content from container integration test",
|
||||
headers={"Content-Type": "text/plain"},
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert "id" in data
|
||||
assert "url" in data
|
||||
|
||||
def test_paste_retrieval(self, running_container):
|
||||
"""Should be able to create and retrieve a paste."""
|
||||
# Create
|
||||
content = b"Retrieve test content"
|
||||
create_response = requests.post(
|
||||
running_container,
|
||||
data=content,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
)
|
||||
assert create_response.status_code == 201
|
||||
paste_id = create_response.json()["id"]
|
||||
|
||||
# Retrieve
|
||||
get_response = requests.get(
|
||||
f"{running_container}/raw/{paste_id}", timeout=REQUEST_TIMEOUT
|
||||
)
|
||||
assert get_response.status_code == 200
|
||||
assert get_response.content == content
|
||||
|
||||
def test_paste_info(self, running_container):
|
||||
"""Should be able to get paste info."""
|
||||
# Create
|
||||
create_response = requests.post(
|
||||
running_container,
|
||||
data=b"Info test content",
|
||||
headers={"Content-Type": "text/plain"},
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
)
|
||||
paste_id = create_response.json()["id"]
|
||||
|
||||
# Get info
|
||||
info_response = requests.get(
|
||||
f"{running_container}/{paste_id}", timeout=REQUEST_TIMEOUT
|
||||
)
|
||||
assert info_response.status_code == 200
|
||||
info = info_response.json()
|
||||
assert info["id"] == paste_id
|
||||
assert info["size"] == len(b"Info test content")
|
||||
assert info["mime_type"] == "text/plain"
|
||||
|
||||
def test_security_headers(self, running_container):
|
||||
"""Response should include security headers."""
|
||||
response = requests.get(f"{running_container}/health", timeout=REQUEST_TIMEOUT)
|
||||
|
||||
# Check for security headers
|
||||
assert response.headers.get("X-Content-Type-Options") == "nosniff"
|
||||
assert response.headers.get("X-Frame-Options") == "DENY"
|
||||
assert "no-store" in response.headers.get("Cache-Control", "")
|
||||
|
||||
def test_json_responses(self, running_container):
|
||||
"""API responses should be JSON."""
|
||||
# Health endpoint
|
||||
health_response = requests.get(f"{running_container}/health", timeout=REQUEST_TIMEOUT)
|
||||
assert health_response.headers.get("Content-Type") == "application/json"
|
||||
|
||||
# Create paste
|
||||
create_response = requests.post(
|
||||
running_container,
|
||||
data=b"JSON test",
|
||||
timeout=REQUEST_TIMEOUT,
|
||||
)
|
||||
assert "application/json" in create_response.headers.get("Content-Type", "")
|
||||
|
||||
def test_not_found_returns_json(self, running_container):
|
||||
"""404 errors should return JSON."""
|
||||
response = requests.get(
|
||||
f"{running_container}/raw/nonexistent1234", timeout=REQUEST_TIMEOUT
|
||||
)
|
||||
assert response.status_code == 404
|
||||
assert "application/json" in response.headers.get("Content-Type", "")
|
||||
data = response.json()
|
||||
assert "error" in data
|
||||
|
||||
def test_pow_challenge_endpoint(self, running_container):
|
||||
"""PoW challenge endpoint should be accessible."""
|
||||
response = requests.get(f"{running_container}/pow/challenge", timeout=REQUEST_TIMEOUT)
|
||||
# PoW may or may not be enabled, but endpoint should exist
|
||||
assert response.status_code in (200, 400, 404)
|
||||
|
||||
def test_large_paste_creation(self, running_container):
|
||||
"""Should handle reasonably sized pastes."""
|
||||
# Create 100KB paste
|
||||
content = b"x" * (100 * 1024)
|
||||
response = requests.post(
|
||||
running_container,
|
||||
data=content,
|
||||
headers={"Content-Type": "application/octet-stream"},
|
||||
timeout=REQUEST_TIMEOUT * 2,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
|
||||
# Verify retrieval
|
||||
paste_id = response.json()["id"]
|
||||
get_response = requests.get(
|
||||
f"{running_container}/raw/{paste_id}", timeout=REQUEST_TIMEOUT * 2
|
||||
)
|
||||
assert get_response.status_code == 200
|
||||
assert len(get_response.content) == len(content)
|
||||
|
||||
|
||||
class TestContainerEnvironment:
|
||||
"""Test container environment configuration."""
|
||||
|
||||
def test_runs_as_non_root(self, container_runtime, running_container):
|
||||
"""Container should run as non-root user."""
|
||||
result = subprocess.run(
|
||||
[container_runtime, "exec", CONTAINER_NAME, "id", "-u"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
uid = int(result.stdout.strip())
|
||||
assert uid != 0, "Container should not run as root (UID 0)"
|
||||
|
||||
def test_data_directory_writable(self, container_runtime, running_container):
|
||||
"""Data directory should be writable."""
|
||||
result = subprocess.run(
|
||||
[container_runtime, "exec", CONTAINER_NAME, "touch", "/app/data/test-write"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
|
||||
def test_gunicorn_workers(self, container_runtime, running_container):
|
||||
"""Should have multiple gunicorn workers."""
|
||||
result = subprocess.run(
|
||||
[container_runtime, "exec", CONTAINER_NAME, "pgrep", "-c", "gunicorn"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
)
|
||||
assert result.returncode == 0
|
||||
worker_count = int(result.stdout.strip())
|
||||
# Should have master + at least 2 workers
|
||||
assert worker_count >= 3, f"Expected at least 3 gunicorn processes, got {worker_count}"
|
||||
Reference in New Issue
Block a user