a040fad0b8
Some checks failed
CI / Lint & Format (push) Failing after 16s
CI / Unit Tests (push) Has been skipped
CI / Memory Leak Check (push) Has been skipped
CI / SBOM Generation (push) Has been skipped
CI / Security Scan (push) Failing after 21s
CI / Security Tests (push) Has been skipped
366 lines
12 KiB
Python
366 lines
12 KiB
Python
"""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
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
# Skip module entirely if integration tests are disabled
|
|
# This allows the file to be collected without requiring 'requests'
|
|
_INTEGRATION_ENABLED = bool(os.environ.get("FLASKPASTE_INTEGRATION"))
|
|
|
|
requests: Any = None
|
|
if _INTEGRATION_ENABLED:
|
|
import requests # type: ignore[import-untyped,no-redef]
|
|
|
|
# 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 _INTEGRATION_ENABLED:
|
|
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}"
|