diff --git a/tests/test_container_integration.py b/tests/test_container_integration.py new file mode 100644 index 0000000..b7c9141 --- /dev/null +++ b/tests/test_container_integration.py @@ -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}"