badblocks/docker-easy-haproxy
1
0
Fork 0
Code Activity

Refactor E2E tests and configuration format

Browse source 
- Replaced `easymapping` configuration with `containers` for better maintainability and clarity.
- Introduced `DockerComposeFixture` class in `utils.py` to manage Docker Compose lifecycle and smart build strategy.
- Enhanced YAML-to-environment variable conversion in `ContainerEnv` for dynamic configuration support.
- Updated HAProxy configurations and test fixtures to reflect the new format.
- Improved test coverage for YAML parsing, environment variable handling, and HAProxy config generation.
This commit is contained in:
Joao Gilberto Magalhaes 2026-02-13 11:44:26 -05:00
parent c125985150
commit 97845b8a52
22 changed files with 1042 additions and 442 deletions
Download patch file Download diff file Expand all files Collapse all files

9
tests_e2e/static/README.md
View file

@ -79,13 +79,10 @@ stats:
password: password
port: 1936
easymapping:
- port: 443
containers:
"host1.local:443":
ip: ["container:8080"] # Can also be IP:PORT for external backends
ssl: true
hosts:
host1.local:
containers:
- container:8080 # Can also be IP:PORT for external backends
```
See `conf/` directory for complete examples.

22
tests_e2e/static/conf/config-basic.yml
View file

@ -15,17 +15,17 @@ stats:
customerrors: true # Optional (default false)
easymapping:
# HTTP - Redirect to HTTPS
- port: 80
redirect:
host1.local: https://host1.local
www.host1.local: https://host1.local
containers:
# HTTP - Redirect to HTTPS using redirect_ssl
"host1.local:80":
ip: ["container:8080"]
redirect_ssl: true
"www.host1.local:80":
ip: ["container:8080"]
redirect_ssl: true
# HTTPS - Serve application
- port: 443
"host1.local:443":
ip: ["container:8080"]
ssl: true
hosts:
host1.local:
containers:
- container:8080

70
tests_e2e/static/conf/config-certbot.yml
View file

@ -35,54 +35,48 @@ stats:
customerrors: true
easymapping:
containers:
# HTTP Port 80
# Required for ACME HTTP-01 challenge and redirect
- port: 80
hosts:
# Domain with certbot enabled
example.com:
containers:
- webapp:8080
# Enable certbot for this domain
certbot: true
# Redirect HTTP to HTTPS after cert is issued
redirect_ssl: true
# Additional domain with certbot
app.example.com:
containers:
- app:3000
certbot: true
redirect_ssl: true
# Domain with certbot enabled
"example.com:80":
ip: ["webapp:8080"]
# Enable certbot for this domain
certbot: true
# Redirect HTTP to HTTPS after cert is issued
redirect_ssl: true
# Domain without certbot (uses custom certificate)
custom.example.com:
containers:
- custom-app:8080
# No certbot - expects certificate at /certs/haproxy/custom.example.com.pem
# Additional domain with certbot
"app.example.com:80":
ip: ["app:3000"]
certbot: true
redirect_ssl: true
# Domain without certbot (uses custom certificate)
"custom.example.com:80":
ip: ["custom-app:8080"]
# No certbot - expects certificate at /certs/haproxy/custom.example.com.pem
# HTTPS Port 443
# Serves HTTPS traffic with auto-generated certificates
- port: 443
"example.com:443":
ip: ["webapp:8080"]
ssl: true
hosts:
example.com:
containers:
- webapp:8080
# Certificate path (auto-generated by certbot)
# /certs/certbot/example.com/fullchain.pem
# Certificate path (auto-generated by certbot)
# /certs/certbot/example.com/fullchain.pem
app.example.com:
containers:
- app:3000
"app.example.com:443":
ip: ["app:3000"]
ssl: true
# Custom certificate example
custom.example.com:
containers:
- custom-app:8080
# Place your certificate at:
# /certs/haproxy/custom.example.com.pem
# Custom certificate example
"custom.example.com:443":
ip: ["custom-app:8080"]
ssl: true
# Place your certificate at:
# /certs/haproxy/custom.example.com.pem
# Multiple domains with different backends
# Certbot will request separate certificates for each domain

75
tests_e2e/static/conf/config-deny-pages.yml
View file

@ -33,46 +33,39 @@ plugins:
- /config
status_code: 404 # Hide existence of these paths
easymapping:
- port: 80
hosts:
# Domain 1: Uses global deny_pages configuration
host1.local:
containers:
- webapp1:8080
# No plugins specified = uses global configuration
containers:
# Domain 1: Uses global deny_pages configuration
"host1.local:80":
ip: ["webapp1:8080"]
# No plugins specified = uses global configuration
# Domain 2: WordPress site with custom blocked paths
host2.local:
containers:
- wordpress:80
# Override global plugin configuration for this domain
plugins:
- deny_pages
plugin_config:
deny_pages:
paths:
- /wp-admin
- /wp-login.php
- /xmlrpc.php
- /wp-config.php
status_code: 403 # Return forbidden instead of 404
# Domain 2: WordPress site with custom blocked paths
"host2.local:80":
ip: ["wordpress:80"]
# Override global plugin configuration for this domain
plugins: [deny_pages]
plugin:
deny_pages:
paths:
- /wp-admin
- /wp-login.php
- /xmlrpc.php
- /wp-config.php
status_code: 403 # Return forbidden instead of 404
# Domain 3: Public site with stricter blocking
host3.local:
containers:
- publicsite:3000
plugins:
- deny_pages
plugin_config:
deny_pages:
paths:
- /admin
- /administrator
- /manager
- /phpmyadmin
- /.git
- /.env
- /config
- /backup
status_code: 404
# Domain 3: Public site with stricter blocking
"host3.local:80":
ip: ["publicsite:3000"]
plugins: [deny_pages]
plugin:
deny_pages:
paths:
- /admin
- /administrator
- /manager
- /phpmyadmin
- /.git
- /.env
- /config
- /backup
status_code: 404

88
tests_e2e/static/conf/config-jwt-validator.yml
View file

@ -32,55 +32,45 @@ stats:
customerrors: true
easymapping:
- port: 80
hosts:
# Public API with full JWT validation
api.local:
containers:
- api-server:8080
plugins:
- jwt_validator
plugin_config:
jwt_validator:
algorithm: RS256
issuer: https://auth.example.com/
audience: https://api.example.com
pubkey_path: /etc/haproxy/jwt_keys/api_pubkey.pem
containers:
# Public API with full JWT validation
"api.local:80":
ip: ["api-server:8080"]
plugins: [jwt_validator]
plugin:
jwt_validator:
algorithm: RS256
issuer: https://auth.example.com/
audience: https://api.example.com
pubkey_path: /etc/haproxy/jwt_keys/api_pubkey.pem
# Internal API - validate signature only (no issuer/audience check)
internal-api.local:
containers:
- internal-api:3000
plugins:
- jwt_validator
plugin_config:
jwt_validator:
algorithm: RS256
# No issuer/audience = skip those validations
pubkey_path: /etc/haproxy/jwt_keys/api_pubkey.pem
# Internal API - validate signature only (no issuer/audience check)
"internal-api.local:80":
ip: ["internal-api:3000"]
plugins: [jwt_validator]
plugin:
jwt_validator:
algorithm: RS256
# No issuer/audience = skip those validations
pubkey_path: /etc/haproxy/jwt_keys/api_pubkey.pem
# Admin API - different issuer and key
admin-api.local:
containers:
- admin-api:4000
plugins:
- jwt_validator
- deny_pages # Also block internal paths
plugin_config:
jwt_validator:
algorithm: RS256
issuer: https://admin-auth.example.com/
audience: https://admin.example.com
pubkey_path: /etc/haproxy/jwt_keys/admin_pubkey.pem
deny_pages:
paths:
- /internal
- /debug
status_code: 403
# Admin API - different issuer and key
"admin-api.local:80":
ip: ["admin-api:4000"]
plugins: [jwt_validator, deny_pages] # Also block internal paths
plugin:
jwt_validator:
algorithm: RS256
issuer: https://admin-auth.example.com/
audience: https://admin.example.com
pubkey_path: /etc/haproxy/jwt_keys/admin_pubkey.pem
deny_pages:
paths:
- /internal
- /debug
status_code: 403
# Public website - no JWT required
website.local:
containers:
- website:8080
# No plugins = public access
# Public website - no JWT required
"website.local:80":
ip: ["website:8080"]
# No plugins = public access

57
tests_e2e/static/docker-compose.yml
View file

@ -60,17 +60,70 @@
services:
haproxy:
image: byjg/easy-haproxy:5.0.0
image: byjg/easy-haproxy:local
build:
context: ../..
dockerfile: build/Dockerfile
volumes:
- ./conf/:/etc/haproxy/static/
- ./host1.local.pem:/certs/haproxy/host1.local.pem
- ../static/host1.local.pem:/certs/haproxy/host1.local.pem:ro
- ../docker/jwt_pubkey.pem:/etc/haproxy/jwt_keys/api_pubkey.pem:ro
- ../docker/jwt_pubkey.pem:/etc/haproxy/jwt_keys/admin_pubkey.pem:ro
- /var/run/docker.sock:/var/run/docker.sock
environment:
EASYHAPROXY_DISCOVER: static
HAPROXY_USERNAME: admin
HAPROXY_PASSWORD: password
ports:
- "80:80/tcp"
- "443:443/tcp"
- "1936:1936/tcp"
# Main container for basic tests
container:
image: byjg/static-httpserver
container_name: container
# Containers for deny-pages tests
webapp1:
image: byjg/static-httpserver
container_name: webapp1
environment:
TITLE: "WebApp 1"
wordpress:
image: byjg/static-httpserver
container_name: wordpress
environment:
TITLE: "WordPress Site"
publicsite:
image: byjg/static-httpserver
container_name: publicsite
environment:
TITLE: "Public Site"
# Containers for JWT validator tests
api-server:
image: byjg/static-httpserver
container_name: api-server
environment:
TITLE: "Protected API"
internal-api:
image: byjg/static-httpserver
container_name: internal-api
environment:
TITLE: "Internal API"
admin-api:
image: byjg/static-httpserver
container_name: admin-api
environment:
TITLE: "Admin API"
website:
image: byjg/static-httpserver
container_name: website
environment:
TITLE: "Public Website"

87
tests_e2e/test_docker_compose.py
View file

@ -33,17 +33,15 @@ import pytest
import requests
import jwt as jwt_lib
from typing import Generator
from utils import extract_backend_block
from utils import extract_backend_block, DockerComposeFixture
# Base directory for docker-compose files
BASE_DIR = Path(__file__).parent.absolute()
DOCKER_DIR = BASE_DIR / "docker"
# Track if cloudflare_ips.lst has been created in this test session
_cloudflare_ips_created = False
# Track if Docker image has been built in this test session
_docker_image_built = False
def create_cloudflare_ips_file():
"""
@ -89,75 +87,10 @@ def create_cloudflare_ips_file():
_cloudflare_ips_created = True
class DockerComposeFixture:
"""Helper class to manage docker-compose lifecycle"""
def __init__(self, compose_file: str, startup_wait: int = 3, build: bool = None):
self.compose_file = str(BASE_DIR / "docker" / compose_file)
self.startup_wait = startup_wait
# Smart build strategy: build on first call, skip on subsequent calls
global _docker_image_built
if build is None:
self.build = not _docker_image_built
else:
self.build = build
def up(self):
"""Start docker-compose services"""
global _docker_image_built
compose_name = Path(self.compose_file).name
print() # Newline for better test output formatting
print(f" → Starting services from {compose_name}...")
cmd = ["docker", "compose", "-f", self.compose_file, "up", "-d"]
if self.build:
cmd.append("--build")
result = subprocess.run(
cmd,
capture_output=True,
text=True
)
if result.returncode != 0:
print(f" ✗ ERROR: Failed to start services!")
print(f" stdout: {result.stdout}")
print(f" stderr: {result.stderr}")
raise subprocess.CalledProcessError(result.returncode, cmd, result.stdout, result.stderr)
# Mark image as built for this test session
if self.build:
_docker_image_built = True
print(f" ✓ Services started, waiting {self.startup_wait}s for initialization...")
time.sleep(self.startup_wait)
print(f" ✓ Services ready")
def down(self):
"""Stop and remove docker-compose services"""
compose_name = Path(self.compose_file).name
print(f" → Stopping services from {compose_name}...")
result = subprocess.run(
["docker", "compose", "-f", self.compose_file, "down", "--remove-orphans"],
capture_output=True,
text=True
)
if result.returncode != 0:
print(f" ⚠ WARNING: Failed to stop services cleanly")
print(f" stderr: {result.stderr}")
# Don't raise error on cleanup, just warn
else:
print(f" ✓ Services stopped and cleaned up")
@pytest.fixture
def docker_compose_basic_ssl() -> Generator[None, None, None]:
"""Fixture for docker-compose.yml (Basic SSL)"""
fixture = DockerComposeFixture("docker-compose.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose.yml"))
fixture.up()
yield
fixture.down()
@ -166,7 +99,7 @@ def docker_compose_basic_ssl() -> Generator[None, None, None]:
@pytest.fixture
def docker_compose_jwt_validator() -> Generator[None, None, None]:
"""Fixture for docker-compose-jwt-validator.yml"""
fixture = DockerComposeFixture("docker-compose-jwt-validator.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-jwt-validator.yml"))
fixture.up()
yield
fixture.down()
@ -175,7 +108,7 @@ def docker_compose_jwt_validator() -> Generator[None, None, None]:
@pytest.fixture
def docker_compose_multi_containers() -> Generator[None, None, None]:
"""Fixture for docker-compose-multi-containers.yml"""
fixture = DockerComposeFixture("docker-compose-multi-containers.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-multi-containers.yml"))
fixture.up()
yield
fixture.down()
@ -184,7 +117,7 @@ def docker_compose_multi_containers() -> Generator[None, None, None]:
@pytest.fixture
def docker_compose_php_fpm() -> Generator[None, None, None]:
"""Fixture for docker-compose-php-fpm.yml"""
fixture = DockerComposeFixture("docker-compose-php-fpm.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-php-fpm.yml"))
fixture.up()
yield
fixture.down()
@ -196,7 +129,7 @@ def docker_compose_plugins_combined() -> Generator[None, None, None]:
# Create cloudflare_ips.lst (required by this compose file)
create_cloudflare_ips_file()
fixture = DockerComposeFixture("docker-compose-plugins-combined.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-plugins-combined.yml"))
fixture.up()
yield
fixture.down()
@ -205,7 +138,7 @@ def docker_compose_plugins_combined() -> Generator[None, None, None]:
@pytest.fixture
def docker_compose_ip_whitelist() -> Generator[None, None, None]:
"""Fixture for docker-compose-ip-whitelist.yml"""
fixture = DockerComposeFixture("docker-compose-ip-whitelist.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-ip-whitelist.yml"))
fixture.up()
yield
fixture.down()
@ -217,7 +150,7 @@ def docker_compose_cloudflare() -> Generator[None, None, None]:
# Create cloudflare_ips.lst (required by this compose file)
create_cloudflare_ips_file()
fixture = DockerComposeFixture("docker-compose-cloudflare.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-cloudflare.yml"))
fixture.up()
yield
fixture.down()
@ -787,7 +720,7 @@ class TestCloudflare:
@pytest.fixture
def docker_compose_changed_label() -> Generator[None, None, None]:
"""Fixture for docker-compose-changed-label.yml"""
fixture = DockerComposeFixture("docker-compose-changed-label.yml")
fixture = DockerComposeFixture(str(DOCKER_DIR / "docker-compose-changed-label.yml"))
fixture.up()
yield
fixture.down()

259
tests_e2e/test_static.py Normal file
View file

@ -0,0 +1,259 @@
"""
Pytest test suite for EasyHAProxy Static Configuration Mode
These tests verify static YAML configuration mode (EASYHAPROXY_DISCOVER=static).
Tests are organized by configuration file and can be run individually or as a suite.
Requirements:
- pytest
- requests
- PyJWT
- cryptography
- docker-compose
Usage:
# Run all static tests
pytest test_static.py -v
# Run specific test class
pytest test_static.py::TestStaticBasic -v
# Run specific test
pytest test_static.py::TestStaticBasic::test_https_host1 -v
"""
import subprocess
import shutil
from pathlib import Path
import pytest
import requests
from typing import Generator
from utils import extract_backend_block, DockerComposeFixture
# Base directory for static configuration
BASE_DIR = Path(__file__).parent.absolute()
STATIC_DIR = BASE_DIR / "static"
CONF_DIR = STATIC_DIR / "conf"
class StaticDockerComposeFixture(DockerComposeFixture):
"""Helper class to manage static docker-compose lifecycle with config file switching"""
def __init__(self, config_file: str, startup_wait: int = 3, build: bool = None):
# Initialize parent with static docker-compose.yml path
super().__init__(str(STATIC_DIR / "docker-compose.yml"), startup_wait, build)
self.config_file = config_file
self.config_source = CONF_DIR / config_file
self.config_target = CONF_DIR / "config.yml"
def up(self):
"""Start docker-compose services with specified config"""
print() # Newline for better test output formatting
print(f" → Using static config: {self.config_file}")
# Copy the config file to config.yml
shutil.copy(self.config_source, self.config_target)
print(f" ✓ Config copied to config.yml")
# Call parent's up() method to start services
super().up()
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def static_basic() -> Generator[None, None, None]:
"""Fixture for config-basic.yml"""
fixture = StaticDockerComposeFixture("config-basic.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def static_deny_pages() -> Generator[None, None, None]:
"""Fixture for config-deny-pages.yml"""
fixture = StaticDockerComposeFixture("config-deny-pages.yml")
fixture.up()
yield
fixture.down()
@pytest.fixture
def static_jwt_validator() -> Generator[None, None, None]:
"""Fixture for config-jwt-validator.yml"""
fixture = StaticDockerComposeFixture("config-jwt-validator.yml")
fixture.up()
yield
fixture.down()
# =============================================================================
# Test: config-basic.yml - Basic HTTP→HTTPS Redirect
# =============================================================================
@pytest.mark.static
class TestStaticBasic:
"""Tests for static config-basic.yml"""
def test_haproxy_config(self, static_basic):
"""Test HAProxy configuration has SSL and redirect configurations"""
result = subprocess.run(
["docker", "exec", "static-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Test HTTPS backend for host1
https_host1_block = extract_backend_block(config, "srv_host1_local_443")
assert https_host1_block, "Backend srv_host1_local_443 not found"
assert "mode http" in https_host1_block
# Verify SSL frontend exists
assert "frontend https_in_443" in config or "bind *:443" in config
# Verify HTTP to HTTPS redirect (new format uses http-request redirect scheme)
assert "http-request redirect scheme https code 301" in config
def test_https_host1(self, static_basic):
"""Test HTTPS access to host1.local"""
response = requests.get(
"https://127.0.0.1/",
headers={"Host": "host1.local"},
verify=False
)
assert response.status_code == 200
def test_http_redirect_host1(self, static_basic):
"""Test HTTP to HTTPS redirect for host1.local"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host1.local"},
allow_redirects=False
)
assert response.status_code == 301
assert "https://host1.local" in response.headers.get("location", "")
def test_haproxy_stats(self, static_basic):
"""Test HAProxy stats interface"""
from conftest import verify_haproxy_stats
verify_haproxy_stats()
# =============================================================================
# Test: config-deny-pages.yml - Deny Pages Plugin
# =============================================================================
@pytest.mark.static
class TestStaticDenyPages:
"""Tests for static config-deny-pages.yml"""
def test_haproxy_config(self, static_deny_pages):
"""Test HAProxy configuration has deny pages rules"""
result = subprocess.run(
["docker", "exec", "static-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract backend for host1.local (should have global deny_pages config)
backend_block = extract_backend_block(config, "srv_host1_local_80")
assert backend_block, "Backend srv_host1_local_80 not found"
# Verify deny pages plugin is configured
assert "# Deny Pages - Block specific paths" in backend_block
assert "acl denied_path path_beg" in backend_block
assert "/admin" in backend_block
assert "/.env" in backend_block
assert "/config" in backend_block
assert "http-request deny" in backend_block
def test_normal_access(self, static_deny_pages):
"""Test normal access to allowed paths"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "host1.local"}
)
assert response.status_code == 200
def test_blocked_paths(self, static_deny_pages):
"""Test access to blocked paths"""
blocked_paths = ["/admin", "/.env", "/config"]
for path in blocked_paths:
response = requests.get(
f"http://127.0.0.1{path}",
headers={"Host": "host1.local"}
)
assert response.status_code == 404, f"Path {path} should be blocked with 404"
def test_haproxy_stats(self, static_deny_pages):
"""Test HAProxy stats interface"""
from conftest import verify_haproxy_stats
verify_haproxy_stats()
# =============================================================================
# Test: config-jwt-validator.yml - JWT Validator Plugin
# =============================================================================
@pytest.mark.static
class TestStaticJWTValidator:
"""Tests for static config-jwt-validator.yml"""
def test_haproxy_config(self, static_jwt_validator):
"""Test HAProxy configuration has JWT validation rules"""
result = subprocess.run(
["docker", "exec", "static-haproxy-1", "cat", "/etc/haproxy/haproxy.cfg"],
capture_output=True,
text=True,
check=True
)
config = result.stdout
# Extract backend for API (static mode uses different naming)
# Find any backend that contains JWT validation
assert "# JWT Validator - Validate JWT tokens" in config, \
"JWT Validator plugin comment not found"
assert "jwt_verify" in config, \
"JWT signature verification not found"
assert "Missing Authorization HTTP header" in config, \
"JWT authorization check not found"
def test_without_token(self, static_jwt_validator):
"""Test API access without JWT token (should fail)"""
response = requests.get(
"http://127.0.0.1/",
headers={"Host": "api.local"}
)
assert response.status_code == 403
assert "Missing Authorization HTTP header" in response.text
def test_with_valid_token(self, static_jwt_validator, jwt_token):
"""Test API access with valid JWT token (should succeed)"""
response = requests.get(
"http://127.0.0.1/",
headers={
"Host": "api.local",
"Authorization": f"Bearer {jwt_token}"
}
)
assert response.status_code == 200
def test_haproxy_stats(self, static_jwt_validator):
"""Test HAProxy stats interface"""
from conftest import verify_haproxy_stats
verify_haproxy_stats()
if __name__ == "__main__":
print("This is a pytest test suite. Run with: pytest test_static.py -v")
print("\nAvailable test classes:")
print(" - TestStaticBasic: Basic static configuration tests")
print(" - TestStaticDenyPages: Deny pages plugin tests")
print(" - TestStaticJWTValidator: JWT validator plugin tests")

68
tests_e2e/utils.py
View file

@ -14,6 +14,74 @@ import jwt as jwt_lib
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
# Track if Docker image has been built in this test session
_docker_image_built = False
class DockerComposeFixture:
"""Helper class to manage docker-compose lifecycle"""
def __init__(self, compose_file: str, startup_wait: int = 3, build: bool = None):
self.compose_file = compose_file
self.startup_wait = startup_wait
# Smart build strategy: build on first call, skip on subsequent calls
global _docker_image_built
if build is None:
self.build = not _docker_image_built
else:
self.build = build
def up(self):
"""Start docker-compose services"""
global _docker_image_built
compose_name = Path(self.compose_file).name
print() # Newline for better test output formatting
print(f" → Starting services from {compose_name}...")
cmd = ["docker", "compose", "-f", self.compose_file, "up", "-d"]
if self.build:
cmd.append("--build")
result = subprocess.run(
cmd,
capture_output=True,
text=True
)
if result.returncode != 0:
print(f" ✗ ERROR: Failed to start services!")
print(f" stdout: {result.stdout}")
print(f" stderr: {result.stderr}")
raise subprocess.CalledProcessError(result.returncode, cmd, result.stdout, result.stderr)
# Mark image as built for this test session
if self.build:
_docker_image_built = True
print(f" ✓ Services started, waiting {self.startup_wait}s for initialization...")
time.sleep(self.startup_wait)
print(f" ✓ Services ready")
def down(self):
"""Stop and remove docker-compose services"""
compose_name = Path(self.compose_file).name
print(f" → Stopping services from {compose_name}...")
result = subprocess.run(
["docker", "compose", "-f", self.compose_file, "down", "--remove-orphans"],
capture_output=True,
text=True
)
if result.returncode != 0:
print(f" ⚠ WARNING: Failed to stop services cleanly")
print(f" stderr: {result.stderr}")
# Don't raise error on cleanup, just warn
else:
print(f" ✓ Services stopped and cleaned up")
def generate_jwt_token(
private_key_path: Path,