badblocks/docker-easy-haproxy
1
0
Fork 0
Code Activity
docker-easy-haproxy/tests_e2e/test_swarm.py
Joao Gilberto Magalhaes d9c7ff89cc Set pytest.fixture scope to class, add Swarm E2E tests, and update Docker Swarm configurations 
- Modified `pytest.fixture` scope to `class` across tests to improve fixture lifecycle and reduce setup overhead.
- Introduced a new `tests_e2e/test_swarm.py` test suite for Docker Swarm stack configurations.
- Added E2E tests for Swarm mode with scenarios such as basic services and combined plugins.
- Updated HAProxy image in Swarm stack `.yml` files to `byjg/easy-haproxy:local` for local testing consistency.
- Added `swarm` marker in `pyproject.toml` for differentiating Docker Swarm-specific tests.
- Updated GitHub workflows to include a new Swarm E2E test pipeline.
2026-02-18 15:28:07 -05:00

635 lines
No EOL
24 KiB
Python
Raw Blame History

"""
Pytest test suite for EasyHAProxy Docker Swarm examples
These tests verify the functionality of Docker Swarm stack configurations.
Tests require Docker with Swarm support.
Test scenarios:
1. TestSwarmBasicServices: easyhaproxy.yml + services.yml
- HTTPS for host1.local and host2.local (embedded SSL cert)
- HTTP to HTTPS redirect
- HAProxy stats interface
- HAProxy config verification
2. TestSwarmPluginsCombined: plugins-combined.yml
- Public website (Cloudflare + deny_pages)
- Protected API (JWT validator + deny_pages)
- Admin panel (IP whitelist)
- HAProxy stats interface
Requirements:
- Docker with Swarm support
- pytest, requests, PyJWT, cryptography
Usage:
# Run all swarm tests
pytest tests_e2e/test_swarm.py -v
# Run specific test class
pytest tests_e2e/test_swarm.py::TestSwarmBasicServices -v
# Run with markers
pytest tests_e2e/test_swarm.py -m swarm -v
"""
import subprocess
import tempfile
import time
from pathlib import Path
from typing import Generator
import pytest
import requests
BASE_DIR = Path(__file__).parent.absolute()
SWARM_DIR = BASE_DIR / "swarm"
# Session-level init state (prevent redundant work within a session;
# swarm and network are left running after the session so subsequent runs skip setup)
_swarm_image_built = False
_swarm_initialized = False
_swarm_network_created = False
# Session-level cloudflare config state
_cloudflare_config_created = False
# =============================================================================
# Image and Swarm Infrastructure Helpers
# =============================================================================
def ensure_haproxy_image():
"""Build byjg/easy-haproxy:local from source at the start of the test session.
Always builds on the first call so tests run against the current codebase,
never a stale image left over from a previous session or docker compose run.
Subsequent calls within the same session are no-ops (image already built).
"""
global _swarm_image_built
if _swarm_image_built:
return
project_root = BASE_DIR.parent
print("\n → Building byjg/easy-haproxy:local from source...")
subprocess.run(
[
"docker", "build",
"-t", "byjg/easy-haproxy:local",
"-f", str(project_root / "deploy/docker/Dockerfile"),
str(project_root),
],
check=True,
)
print(" ✓ Image built as byjg/easy-haproxy:local")
_swarm_image_built = True
def init_swarm() -> None:
"""Initialize Docker Swarm if not already done this session.
Uses a session-level flag (like ensure_haproxy_image) so the check runs at most
once per pytest session regardless of how many fixtures call it.
Swarm is left running after the session so subsequent runs skip initialization.
"""
global _swarm_initialized
if _swarm_initialized:
return
result = subprocess.run(
["docker", "info", "--format", "{{.Swarm.LocalNodeState}}"],
capture_output=True, text=True, check=True
)
if result.stdout.strip() != "active":
print("\n → Initializing Docker Swarm...")
subprocess.run(["docker", "swarm", "init"], check=True, capture_output=True)
print(" ✓ Docker Swarm initialized")
else:
print("\n ✓ Docker Swarm already active")
_swarm_initialized = True
def create_overlay_network(network_name: str = "easyhaproxy") -> None:
"""Create an attachable overlay network if not already done this session.
Uses a session-level flag (like ensure_haproxy_image / init_swarm) so the check
runs at most once per pytest session. The network is left running after the
session so subsequent runs skip creation.
"""
global _swarm_network_created
if _swarm_network_created:
return
result = subprocess.run(
["docker", "network", "ls", "--filter", f"name={network_name}", "--format", "{{.Name}}"],
capture_output=True, text=True, check=True
)
existing = [n.strip() for n in result.stdout.strip().split("\n") if n.strip()]
if network_name not in existing:
print(f"\n → Creating overlay network '{network_name}'...")
subprocess.run(
["docker", "network", "create", "--driver", "overlay", "--attachable", network_name],
check=True, capture_output=True
)
print(f" ✓ Overlay network '{network_name}' created")
else:
print(f"\n ✓ Overlay network '{network_name}' already exists")
_swarm_network_created = True
def wait_for_swarm_services(stack_name: str, timeout: int = 120) -> bool:
"""Poll until all services in a stack have reached their target replica count."""
start_time = time.time()
print(f"\n → Waiting for stack '{stack_name}' services (timeout: {timeout}s)...")
while time.time() - start_time < timeout:
result = subprocess.run(
["docker", "stack", "services", stack_name, "--format", "{{.Replicas}}"],
capture_output=True, text=True
)
if result.returncode != 0 or not result.stdout.strip():
time.sleep(2)
continue
replicas = [r.strip() for r in result.stdout.strip().split("\n") if "/" in r]
if not replicas:
time.sleep(2)
continue
all_ready = all(r.split("/")[0] == r.split("/")[1] for r in replicas)
if all_ready:
elapsed = time.time() - start_time
print(f" ✓ All {len(replicas)} service(s) ready ({elapsed:.1f}s)")
return True
time.sleep(2)
return False
def wait_for_http(
url: str,
headers: dict = None,
expected_status: list = None,
timeout: int = 120,
verify_ssl: bool = False,
) -> bool:
"""Poll an HTTP endpoint until it responds with an expected status code."""
if expected_status is None:
expected_status = [200, 301, 302, 403, 401, 404]
start_time = time.time()
while time.time() - start_time < timeout:
try:
resp = requests.get(
url,
headers=headers or {},
verify=verify_ssl,
allow_redirects=False,
timeout=5,
)
if resp.status_code in expected_status:
return True
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
pass
time.sleep(3)
return False
def get_haproxy_container_id(stack_name: str, service_name: str = "haproxy") -> str:
"""Return the container ID for a swarm service (e.g., 'easyhaproxy_haproxy')."""
result = subprocess.run(
["docker", "ps", "--filter", f"name={stack_name}_{service_name}", "--format", "{{.ID}}"],
capture_output=True, text=True, check=True
)
container_ids = [c.strip() for c in result.stdout.strip().split("\n") if c.strip()]
if not container_ids:
raise RuntimeError(f"No container found for {stack_name}_{service_name}")
return container_ids[0]
def create_swarm_config(config_name: str, file_path: Path) -> None:
"""Create a Docker swarm config from a file, removing any existing one first."""
subprocess.run(["docker", "config", "rm", config_name], capture_output=True)
subprocess.run(
["docker", "config", "create", config_name, str(file_path)],
check=True, capture_output=True
)
print(f" ✓ Docker config '{config_name}' created")
def remove_swarm_config(config_name: str) -> None:
"""Remove a Docker swarm config, ignoring errors if it doesn't exist."""
subprocess.run(["docker", "config", "rm", config_name], capture_output=True)
def create_cloudflare_config():
"""Download Cloudflare IP ranges and create a Docker swarm config.
Adds the 10.0.0.0/8 range so that requests routed through Docker Swarm's
ingress network (typically 10.x.x.x) are treated as Cloudflare IPs in tests.
"""
global _cloudflare_config_created
if _cloudflare_config_created:
return
print("\n → Creating 'cloudflare_ips' Docker config...")
with tempfile.NamedTemporaryFile(mode="w", suffix=".lst", delete=False) as f:
temp_path = Path(f.name)
try:
subprocess.run(
["curl", "-s", "https://www.cloudflare.com/ips-v4"],
stdout=open(temp_path, "w"), check=True
)
with open(temp_path, "a") as f:
f.write("\n")
subprocess.run(
["curl", "-s", "https://www.cloudflare.com/ips-v6"],
stdout=open(temp_path, "a"), check=True
)
# Add Docker ingress range so test requests appear to come from Cloudflare
with open(temp_path, "a") as f:
f.write("\n10.0.0.0/8\n")
create_swarm_config("cloudflare_ips", temp_path)
_cloudflare_config_created = True
finally:
temp_path.unlink(missing_ok=True)
# =============================================================================
# SwarmFixture: manages docker stack lifecycle
# =============================================================================
class SwarmFixture:
"""Helper class to manage Docker Swarm stack lifecycle."""
def __init__(self, stack_files, stack_name: str, timeout: int = 120):
self.stack_files = stack_files if isinstance(stack_files, list) else [stack_files]
self.stack_name = stack_name
self.timeout = timeout
def up(self):
"""Deploy the stack and wait for all services to be running."""
for stack_file in self.stack_files:
name = Path(stack_file).name
print(f"\n → Deploying stack '{self.stack_name}' from {name}...")
result = subprocess.run(
[
"docker", "stack", "deploy",
"--resolve-image", "never", # use local image, never pull from registry
"-c", stack_file, self.stack_name,
],
capture_output=True, text=True
)
if result.returncode != 0:
print(f" ✗ Deploy failed:\n stdout: {result.stdout}\n stderr: {result.stderr}")
raise subprocess.CalledProcessError(
result.returncode, result.args, result.stdout, result.stderr
)
if not wait_for_swarm_services(self.stack_name, self.timeout):
raise TimeoutError(
f"Stack '{self.stack_name}' services did not become ready within {self.timeout}s"
)
def down(self):
"""Force-kill all stack containers, then remove the stack definition."""
print(f"\n → Removing stack '{self.stack_name}'...")
# Force-kill running containers immediately (no graceful shutdown period)
result = subprocess.run(
["docker", "ps", "-q", "--filter", f"name={self.stack_name}_"],
capture_output=True, text=True
)
container_ids = [c.strip() for c in result.stdout.strip().split("\n") if c.strip()]
if container_ids:
subprocess.run(["docker", "kill"] + container_ids, capture_output=True)
# Remove the stack definition (services, configs, secrets)
subprocess.run(
["docker", "stack", "rm", self.stack_name],
capture_output=True, text=True
)
# Poll until no containers from this stack remain, so ports are free
start = time.time()
while time.time() - start < 60:
result = subprocess.run(
["docker", "ps", "-q", "--filter", f"name={self.stack_name}_"],
capture_output=True, text=True
)
if not result.stdout.strip():
break
time.sleep(2)
print(f" ✓ Stack '{self.stack_name}' removed")
# =============================================================================
# Session-level setup: swarm mode + overlay network + image build
# =============================================================================
@pytest.fixture(scope="session", autouse=True)
def swarm_setup():
"""Session fixture: build image, initialize Docker Swarm, and create overlay network.
Each step runs at most once per session (guarded by module-level flags).
Swarm mode and the overlay network are left running after the session so that
a subsequent test run can skip setup — the same pattern as the image build.
Only deployed stacks (SwarmFixture) are torn down between test classes.
"""
ensure_haproxy_image()
init_swarm()
create_overlay_network()
yield
# No teardown of swarm/network: they persist for subsequent runs (faster second run)
# =============================================================================
# Per-test fixtures
# =============================================================================
@pytest.fixture(scope="class")
def swarm_basic_services(generate_ssl_certificates) -> Generator[None, None, None]:
"""Fixture for easyhaproxy.yml + services.yml.
Deploys two stacks:
- easyhaproxy: HAProxy in swarm discovery mode
- services: Two backends (host1.local, host2.local) with embedded SSL
EasyHAProxy auto-attaches services to the easyhaproxy overlay network on
each refresh cycle (default: every 10 seconds), then regenerates HAProxy
config to include the discovered backends.
"""
easyhaproxy = SwarmFixture(str(SWARM_DIR / "easyhaproxy.yml"), "easyhaproxy", timeout=120)
services = SwarmFixture(str(SWARM_DIR / "services.yml"), "services", timeout=60)
easyhaproxy.up()
services.up()
# Wait for EasyHAProxy to auto-attach services, regenerate config, and
# for HAProxy to start serving traffic (up to 2 refresh cycles = ~20s).
print("\n → Waiting for HAProxy to discover and configure swarm services...")
ready = wait_for_http(
"https://127.0.0.1/",
headers={"Host": "host1.local"},
expected_status=[200, 301, 302, 503],
timeout=120,
)
if not ready:
print(" ⚠ Warning: Services may not be fully configured yet")
else:
print(" ✓ Services are reachable through HAProxy")
yield
services.down()
easyhaproxy.down()
@pytest.fixture(scope="class")
def swarm_plugins_combined(generate_ssl_certificates) -> Generator[None, None, None]:
"""Fixture for plugins-combined.yml.
Creates required Docker swarm configs (jwt_api_pubkey, cloudflare_ips) and
deploys a self-contained stack containing:
- HAProxy with plugin support
- Public website (Cloudflare + deny_pages)
- Protected API (JWT validator + deny_pages)
- Admin panel (IP whitelist: 203.0.113.0/24, 10.0.0.0/8)
"""
certs = generate_ssl_certificates
print("\n → Setting up Docker configs for plugins-combined stack...")
create_swarm_config("jwt_api_pubkey", certs["jwt_pubkey"])
create_cloudflare_config()
stack = SwarmFixture(str(SWARM_DIR / "plugins-combined.yml"), "production", timeout=120)
stack.up()
# Wait for HAProxy to discover services and apply plugin configurations
print("\n → Waiting for HAProxy to configure plugin backends...")
ready = wait_for_http(
"http://127.0.0.1/",
headers={"Host": "website.example.com"},
expected_status=[200, 404, 403, 401, 503],
timeout=120,
)
if not ready:
print(" ⚠ Warning: Services may not be fully configured yet")
else:
print(" ✓ Services are reachable through HAProxy")
yield
stack.down()
print("\n → Cleaning up Docker configs...")
remove_swarm_config("jwt_api_pubkey")
remove_swarm_config("cloudflare_ips")
global _cloudflare_config_created
_cloudflare_config_created = False
# =============================================================================
# Tests: easyhaproxy.yml + services.yml - Basic Services with SSL in Swarm
# =============================================================================
@pytest.mark.swarm
@pytest.mark.ssl
class TestSwarmBasicServices:
"""Tests for Swarm mode with basic SSL services.
Uses easyhaproxy.yml (HAProxy) + services.yml (two SSL backends).
EasyHAProxy discovers services via Swarm API and auto-attaches them
to the shared overlay network.
"""
def test_services_running(self, swarm_basic_services):
"""Verify all expected swarm services are running."""
result = subprocess.run(
["docker", "service", "ls", "--format", "{{.Name}}"],
capture_output=True, text=True, check=True
)
service_names = result.stdout
assert "easyhaproxy_haproxy" in service_names, "easyhaproxy_haproxy service not found"
assert "services_container" in service_names, "services_container service not found"
assert "services_container2" in service_names, "services_container2 service not found"
def test_haproxy_config(self, swarm_basic_services):
"""Verify HAProxy configuration contains backends for both swarm services."""
from utils import extract_backend_block
container_id = get_haproxy_container_id("easyhaproxy")
result = subprocess.run(
["docker", "exec", container_id, "cat", "/etc/easyhaproxy/haproxy/haproxy.cfg"],
capture_output=True, text=True, check=True
)
config = result.stdout
# Both HTTPS backends must be present
assert "backend srv_host1_local_443" in config, "HTTPS backend for host1.local not found"
assert "backend srv_host2_local_443" in config, "HTTPS backend for host2.local not found"
# HTTP to HTTPS redirect must be configured
assert "redirect scheme https" in config or "redirect prefix https://" in config, \
"HTTP to HTTPS redirect not found in HAProxy config"
def test_https_host1(self, swarm_basic_services):
"""Test HTTPS access to host1.local through Swarm HAProxy."""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host1.local"},
verify=False,
timeout=10,
)
assert response.status_code == 200
def test_https_host2(self, swarm_basic_services):
"""Test HTTPS access to host2.local through Swarm HAProxy."""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host2.local"},
verify=False,
timeout=10,
)
assert response.status_code == 200
def test_http_redirect_host1(self, swarm_basic_services):
"""Test HTTP to HTTPS permanent redirect for host1.local."""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host1.local"},
allow_redirects=False,
timeout=10,
)
assert response.status_code == 301
assert response.headers.get("location") == "https://host1.local/"
def test_http_redirect_host2(self, swarm_basic_services):
"""Test HTTP to HTTPS permanent redirect for host2.local."""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host2.local"},
allow_redirects=False,
timeout=10,
)
assert response.status_code == 301
assert response.headers.get("location") == "https://host2.local/"
def test_haproxy_stats(self, swarm_basic_services):
"""Test HAProxy stats interface is accessible."""
from conftest import verify_haproxy_stats
verify_haproxy_stats()
# =============================================================================
# Tests: plugins-combined.yml - Multiple Security Plugins in Swarm
# =============================================================================
@pytest.mark.swarm
@pytest.mark.plugins
class TestSwarmPluginsCombined:
"""Tests for multiple combined security plugins in Swarm mode.
Uses plugins-combined.yml which is a self-contained stack:
- HAProxy with Cloudflare + JWT + IP-whitelist plugins
- website.example.com: Cloudflare IP restoration + deny_pages
- api.example.com: JWT validator + deny_pages
- admin.example.com: IP whitelist (203.0.113.0/24, 10.0.0.0/8)
"""
def test_services_running(self, swarm_plugins_combined):
"""Verify all four services are running in the production stack."""
result = subprocess.run(
["docker", "service", "ls", "--format", "{{.Name}}"],
capture_output=True, text=True, check=True
)
service_names = result.stdout
assert "production_haproxy" in service_names, "production_haproxy service not found"
assert "production_website" in service_names, "production_website service not found"
assert "production_api" in service_names, "production_api service not found"
assert "production_admin" in service_names, "production_admin service not found"
def test_website_normal_access(self, swarm_plugins_combined):
"""Test normal GET request reaches the public website (Cloudflare + deny_pages)."""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "website.example.com"},
timeout=10,
)
assert response.status_code == 200
def test_website_blocked_paths(self, swarm_plugins_combined):
"""Test deny_pages plugin blocks sensitive paths with HTTP 404."""
blocked_paths = ["/admin", "/wp-admin", "/.env", "/config"]
for path in blocked_paths:
response = requests.get(
f"http://127.0.0.1{path}",
headers={"Host": "website.example.com"},
timeout=10,
)
assert response.status_code == 404, \
f"Expected 404 for blocked path '{path}', got {response.status_code}"
def test_api_without_token(self, swarm_plugins_combined):
"""Test JWT validator rejects requests that have no Authorization header."""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "api.example.com"},
timeout=10,
)
assert response.status_code in (401, 403), \
f"Expected 401/403 without JWT, got {response.status_code}"
assert "Missing Authorization HTTP header" in response.text
def test_api_with_valid_token(self, swarm_plugins_combined, jwt_token):
"""Test JWT validator allows requests with a valid RS256 JWT token."""
response = requests.get(
"http://127.0.0.1/",
headers={
"Host": "api.example.com",
"Authorization": f"Bearer {jwt_token}",
},
timeout=10,
)
assert response.status_code == 200
def test_api_blocked_paths_with_token(self, swarm_plugins_combined, jwt_token):
"""Test deny_pages blocks internal API paths even with a valid JWT token."""
blocked_paths = ["/internal", "/debug", "/metrics"]
for path in blocked_paths:
response = requests.get(
f"http://127.0.0.1{path}",
headers={
"Host": "api.example.com",
"Authorization": f"Bearer {jwt_token}",
},
timeout=10,
)
assert response.status_code == 403, \
f"Expected 403 for blocked API path '{path}', got {response.status_code}"
def test_admin_panel_ip_whitelist(self, swarm_plugins_combined):
"""Test IP whitelist plugin on admin panel.
The whitelist is '203.0.113.0/24,10.0.0.0/8'.
In Docker Swarm ingress mode, requests from the host arrive at HAProxy
with the Docker ingress router IP (typically in 10.0.0.0/8), so the
admin panel should be accessible.
"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "admin.example.com"},
timeout=10,
)
# Docker Swarm ingress IPs (10.x.x.x) are in the 10.0.0.0/8 whitelist
assert response.status_code == 200, \
(f"Expected admin access from Docker ingress IP (in 10.0.0.0/8), "
f"got {response.status_code}")
def test_haproxy_stats(self, swarm_plugins_combined):
"""Test HAProxy stats interface is accessible."""
from conftest import verify_haproxy_stats
verify_haproxy_stats()
View git blame Copy permalink