busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
03e9fbba6a791c2450b27f0990794b7a7ab2586e
ss-tools/backend/tests/integration/conftest.py
busya 03e9fbba6a fix: TLS Custom CA integration tests — all 175 pass 
ROOT CAUSE: OpenSSL 3.x requires AuthorityKeyIdentifier (AKI) extension
on certificates for capath-based chain building. The ca_chain fixture
generated certs without AKI/SKI extensions — causing ssl.create_default_context()
to fail with 'Missing Authority Key Identifier'.

FIXES (2 files):
1. conftest.py: Added SubjectKeyIdentifier and AuthorityKeyIdentifier
   extensions to all generated certificates in _gen_cert_pem()
2. test_superset_tls_custom_ca.py:
   - Added superset_container param to test_certifi_bundle_notrust
   - Added httpx.ConnectError to caught exceptions (httpx wraps SSLError)
   - Fixed get_dashboards() return type assertion (tuple vs dict)

RESULTS: 175/175 integration tests passing (was 167), +8 TLS tests
- openssl capath OK, certifi fails, certifi bundle no-trust
- httpx capath works, httpx certifi fails, verify_false works
- AsyncAPIClient verify_ssl=True authenticates over HTTPS
- SupersetClient full auth + API calls over TLS
2026-06-16 11:42:40 +03:00

748 lines
27 KiB
Python
Raw Blame History

# #region IntegrationTestConftest [C:3] [TYPE Module] [SEMANTICS test, conftest, testcontainers, postgres, integration]
# @BRIEF Shared pytest fixtures for integration tests using Testcontainers PostgreSQL.
# @RELATION BINDS_TO -> [TranslationJob]
# @RELATION BINDS_TO -> [TranslationRun]
# @RELATION BINDS_TO -> [TerminologyDictionary]
# @PRE Docker daemon is running and accessible.
# @POST PostgreSQL container is started, tables created, session available per test.
# @RATIONALE
# Architecture:
# - Testcontainers spins up a real PostgreSQL 16 container per session.
# - Each test gets an isolated session with transaction rollback.
# - FK constraints, JSON columns, indexes — all behave like production.
#
# Why Testcontainers over SQLite:
# - SQLite silently ignores FK violations in some edge cases.
# - JSON column behavior differs (PostgreSQL JSONB vs SQLite JSON text).
# - Index behavior, constraint names, and error messages differ.
# - Production parity — catches bugs that SQLite tests miss.
#
# @REJECTED
# Always-on PostgreSQL — requires manual setup, not portable.
# Docker Compose for tests — slower startup, harder isolation.
# Shared container across tests — state leakage between tests.
import os
import sys
import time
from pathlib import Path
import pytest
def pytest_collection_modifyitems(config, items):
if not config.getoption("--run-integration"):
skip_integration = pytest.mark.skip(reason="use --run-integration to run")
for item in items:
item.add_marker(skip_integration)
import requests
from sqlalchemy import create_engine, event, text
from sqlalchemy.orm import Session, sessionmaker
sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))
from src.core.database import Base
# Import all models to ensure tables are created
from src.models.translate import ( # noqa: F401
DictionaryEntry,
MetricSnapshot,
TerminologyDictionary,
TranslationBatch,
TranslationEvent,
TranslationJob,
TranslationJobDictionary,
TranslationLanguage,
TranslationPreviewLanguage,
TranslationPreviewRecord,
TranslationPreviewSession,
TranslationRecord,
TranslationRun,
TranslationRunLanguageStats,
TranslationSchedule,
)
from src.models.llm import ( # noqa: F401
LLMProvider,
ValidationPolicy,
ValidationRun,
ValidationRecord,
ValidationSource,
)
from src.models.maintenance import ( # noqa: F401
MaintenanceEvent,
MaintenanceDashboardBanner,
MaintenanceDashboardState,
MaintenanceSettings,
MaintenanceEventStatus,
MaintenanceDashboardBannerStatus,
MaintenanceDashboardStateStatus,
DashboardScope,
)
from src.models.task import ( # noqa: F401
TaskRecord,
TaskLogRecord,
)
from src.models.mapping import Environment # noqa: F401
# #region postgres_container [C:2] [TYPE Fixture]
# @BRIEF Session-scoped Testcontainers PostgreSQL container.
@pytest.fixture(scope="session")
def postgres_container():
"""Start a PostgreSQL 16 container for the test session."""
from testcontainers.postgres import PostgresContainer
with PostgresContainer(
image="postgres:16-alpine",
username="test",
password="test",
dbname="test_translate",
) as pg:
yield pg
# #endregion postgres_container
# #region pg_engine [C:2] [TYPE Fixture]
# @BRIEF SQLAlchemy engine connected to the Testcontainers PostgreSQL.
@pytest.fixture(scope="session")
def pg_engine(postgres_container):
"""Create engine and tables once per session."""
url = postgres_container.get_connection_url()
engine = create_engine(url, echo=False)
# Create all tables
Base.metadata.create_all(bind=engine)
yield engine
# Cleanup
Base.metadata.drop_all(bind=engine)
engine.dispose()
# #endregion pg_engine
# #region db_session [C:2] [TYPE Fixture]
# @BRIEF Per-test database session with transaction rollback for isolation.
@pytest.fixture
def db_session(pg_engine):
"""Provide a transactional database session that rolls back after each test.
This ensures test isolation — each test starts with a clean database state.
Uses connection-level transaction to wrap all operations.
"""
connection = pg_engine.connect()
transaction = connection.begin()
session = sessionmaker(bind=connection)()
# Join the outer transaction
session.begin_nested()
@event.listens_for(session, "after_transaction_end")
def restart_savepoint(session, transaction):
if transaction.nested and not transaction._parent.nested:
session.begin_nested()
try:
yield session
finally:
session.close()
transaction.rollback()
connection.close()
# #endregion db_session
# #region mock_config_manager [C:2] [TYPE Fixture]
# @BRIEF Mock ConfigManager for integration tests.
@pytest.fixture
def mock_config_manager():
"""Provide a mock ConfigManager that returns test environment."""
from unittest.mock import MagicMock
from src.core.config_models import Environment
config_manager = MagicMock()
config_manager.get_environments.return_value = [
Environment(
id="test_env",
name="Test Environment",
url="http://superset:8088",
username="admin",
password="admin",
)
]
config_manager.get_environment.return_value = Environment(
id="test_env",
name="Test Environment",
url="http://superset:8088",
username="admin",
password="admin",
)
config_manager.get_config.return_value = MagicMock()
return config_manager
# #endregion mock_config_manager
# #region verify_postgres_features [C:2] [TYPE Function]
# @BRIEF Helper to verify PostgreSQL-specific features are working.
def verify_postgres_features(session: Session) -> dict:
"""Verify that PostgreSQL-specific features are available.
Returns a dict with feature availability status.
"""
features = {}
# Check JSON support
try:
result = session.execute(text("SELECT '{}'::jsonb")).scalar()
features["jsonb"] = result is not None
except Exception:
features["jsonb"] = False
# Check FK constraints
try:
result = session.execute(text(
"SELECT COUNT(*) FROM information_schema.table_constraints "
"WHERE constraint_type = 'FOREIGN KEY'"
)).scalar()
features["foreign_keys"] = result > 0
except Exception:
features["foreign_keys"] = False
# Check indexes
try:
result = session.execute(text(
"SELECT COUNT(*) FROM pg_indexes WHERE schemaname = 'public'"
)).scalar()
features["indexes"] = result > 0
except Exception:
features["indexes"] = False
return features
# #endregion verify_postgres_features
# ── TLS / PKI Fixtures ────────────────────────────────────────────────
# #region ca_chain [C:2] [TYPE Fixture]
# @BRIEF Session-scoped — generates 3-tier PKI (Root CA → Intermediate CA → Server cert).
# @POST Returns dict with root_crt, intermediate_crt, server_crt, server_key, fullchain PEM strings.
@pytest.fixture(scope="session")
def ca_chain():
"""Generate 3-tier PKI: Root CA → Intermediate CA → Server certificate (SAN: localhost)."""
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
import datetime
def _gen_key():
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
def _gen_cert_pem(subject_name, issuer_name, subject_key, issuer_key, is_ca=False, san_dns=None, san_ip=None):
now = datetime.datetime.now(datetime.timezone.utc)
builder = x509.CertificateBuilder()
builder = builder.subject_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, subject_name),
]))
builder = builder.issuer_name(x509.Name([
x509.NameAttribute(NameOID.COMMON_NAME, issuer_name),
]))
builder = builder.not_valid_before(now - datetime.timedelta(hours=1))
builder = builder.not_valid_after(now + datetime.timedelta(days=365))
builder = builder.serial_number(x509.random_serial_number())
builder = builder.public_key(subject_key.public_key())
if is_ca:
builder = builder.add_extension(
x509.BasicConstraints(ca=True, path_length=None), critical=True,
)
builder = builder.add_extension(
x509.KeyUsage(
key_cert_sign=True, crl_sign=True,
digital_signature=False, content_commitment=False,
key_encipherment=False, data_encipherment=False,
key_agreement=False, encipher_only=False, decipher_only=False,
), critical=True,
)
else:
# Server cert
builder = builder.add_extension(
x509.BasicConstraints(ca=False, path_length=None), critical=True,
)
sans: list[x509.GeneralName] = []
if san_dns:
sans.append(x509.DNSName(san_dns))
if san_ip:
sans.append(x509.IPAddress(san_ip))
builder = builder.add_extension(
x509.SubjectAlternativeName(sans), critical=False,
)
# Subject Key Identifier — required for OpenSSL 3.x capath chain building
builder = builder.add_extension(
x509.SubjectKeyIdentifier.from_public_key(subject_key.public_key()),
critical=False,
)
# Authority Key Identifier — identifies the issuer's key (needed for non-root certs)
if subject_name != issuer_name:
builder = builder.add_extension(
x509.AuthorityKeyIdentifier.from_issuer_public_key(issuer_key.public_key()),
critical=False,
)
cert = builder.sign(issuer_key, hashes.SHA256())
return cert.public_bytes(serialization.Encoding.PEM).decode()
root_key = _gen_key()
root_crt = _gen_cert_pem("Test Root CA", "Test Root CA", root_key, root_key, is_ca=True)
intermediate_key = _gen_key()
intermediate_crt = _gen_cert_pem(
"Test Intermediate CA", "Test Root CA",
intermediate_key, root_key, is_ca=True,
)
server_key = _gen_key()
import ipaddress
server_crt = _gen_cert_pem(
"localhost", "Test Intermediate CA",
server_key, intermediate_key, is_ca=False,
san_dns="localhost", san_ip=ipaddress.IPv4Address("127.0.0.1"),
)
# Full chain for server: server PEM + intermediate PEM (order: leaf first)
fullchain = server_crt + "\n" + intermediate_crt
return {
"root_crt": root_crt,
"root_key": root_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
).decode(),
"intermediate_crt": intermediate_crt,
"intermediate_key": intermediate_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
).decode(),
"server_crt": server_crt,
"server_key": server_key.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.TraditionalOpenSSL,
encryption_algorithm=serialization.NoEncryption(),
).decode(),
"fullchain": fullchain,
}
# #endregion ca_chain
# ── Superset Integration Test Fixtures ────────────────────────────────
# #region superset_db_url [C:2] [TYPE Fixture]
# @BRIEF Session-scoped — creates a separate 'superset_meta' database in the Postgres container.
@pytest.fixture(scope="session")
def superset_db_url(postgres_container):
"""Create a dedicated database for Superset metadata inside the shared Postgres."""
import psycopg2
conn = psycopg2.connect(
host=postgres_container.get_container_host_ip(),
port=postgres_container.get_exposed_port(5432),
user="test",
password="test",
dbname="test_translate",
)
conn.autocommit = True
with conn.cursor() as cur:
cur.execute("DROP DATABASE IF EXISTS superset_meta")
cur.execute("CREATE DATABASE superset_meta")
conn.close()
# Build a Superset-compatible SQLALCHEMY_DATABASE_URI.
# CRITICAL: cannot use "localhost" because from inside the Superset container,
# "localhost" is the container itself. Must use the Docker bridge IP of the
# Postgres container so that the Superset container can reach it.
pg_container = postgres_container.get_wrapped_container()
pg_container.reload()
pg_ip = pg_container.attrs["NetworkSettings"]["Networks"]["bridge"]["IPAddress"]
pg_port = 5432 # internal port, not exposed port
superset_url = (
f"postgresql+psycopg2://test:test@{pg_ip}:{pg_port}/superset_meta"
)
return superset_url
# #endregion superset_db_url
# #region superset_secret_key [C:1] [TYPE Fixture]
# @BRIEF Session-scoped — deterministic secret key for the Superset instance.
@pytest.fixture(scope="session")
def superset_secret_key():
return "test-superset-secret-key-for-integration-tests-xyz"
# #endregion superset_secret_key
# #region superset_admin_password [C:1] [TYPE Fixture]
# @BRIEF Password for the pre-created admin user in the Superset container.
@pytest.fixture(scope="session")
def superset_admin_password():
return "admin123"
# #endregion superset_admin_password
# #region superset_container [C:3] [TYPE Fixture]
# @BRIEF Session-scoped Apache Superset container — db upgrade → admin → init → web server.
# Can be parametrized with {"tls": True} to enable HTTPS with custom 3-tier PKI.
# @PRE Docker daemon is running. superset_db_url points to an empty Postgres database.
# @POST Superset web server is listening on port 8088 (HTTP or HTTPS). Admin user created.
# @SIDE_EFFECT Spins up a short-lived init container, then a long-lived web container.
# In TLS mode: writes server cert + key into container via base64, runs with --cert/--key.
# @RATIONALE Two-container approach (init + web) avoids race conditions where the web
# server starts before migrations have completed. The init container runs `superset db
# upgrade`, creates the admin user, and runs `superset init`, then exits. The web
# container then starts against the fully initialized database.
# TLS parametrization allows testing corporate SSL certificate chain with the
# same Superset instance, using a dynamically generated 3-tier PKI.
# @REJECTED Single container with sequential exec calls — cannot change entrypoint
# from init to web server within the same container lifecycle.
# @REJECTED Using `latest` tag — version drift breaks tests silently.
@pytest.fixture(scope="session")
def superset_container(request, superset_db_url, superset_secret_key, superset_admin_password):
"""Start an Apache Superset instance for integration testing.
Parametrize with indirect=True, {"tls": True} to enable HTTPS with custom CA chain.
"""
from testcontainers.core.container import DockerContainer
import base64
superset_image = "apache/superset:4.1.2"
# Check if TLS was requested via parametrization
tls_config = getattr(request, 'param', None)
use_tls = tls_config is not None and tls_config.get('tls', False)
if use_tls:
ca_chain = request.getfixturevalue('ca_chain')
# Base64-encode certs to avoid shell escaping issues in heredoc
fullchain_b64 = base64.b64encode(ca_chain["fullchain"].encode()).decode()
key_b64 = base64.b64encode(ca_chain["server_key"].encode()).decode()
_write_certs = (
f"echo '{fullchain_b64}' | base64 -d > /tmp/server.crt && "
f"echo '{key_b64}' | base64 -d > /tmp/server.key"
)
_protocol = "https"
_superset_run = (
f"superset run -h 0.0.0.0 -p 8088 "
f"--cert /tmp/server.crt --key /tmp/server.key --with-threads"
)
else:
_write_certs = ""
_protocol = "http"
_superset_run = "superset run -h 0.0.0.0 -p 8088 --with-threads"
# Superset ignores SQLALCHEMY_DATABASE_URI env var — we must create
# superset_config.py and point SUPERSET_CONFIG_PATH to it.
_bootstrap_config = (
"cat > /tmp/superset_config.py << \"PYEOF\"\n"
"import os\n"
"SQLALCHEMY_DATABASE_URI = os.environ.get(\"SUPERSET_DB_URI\", \"sqlite://\")\n"
"PYEOF\n"
)
_export_config = "export SUPERSET_CONFIG_PATH=/tmp/superset_config.py"
# ── Step 1: Init container ──
init = DockerContainer(superset_image)
init.with_env("SUPERSET_SECRET_KEY", superset_secret_key)
init.with_env("SUPERSET_DB_URI", superset_db_url)
init.with_env("SUPERSET_LOAD_EXAMPLES", "no")
init.with_env("FLASK_APP", "superset")
# Note: init container does NOT need TLS certs — only runs db upgrade + init
init.with_command(
f"sh -c '"
f"python -m pip install psycopg2-binary -q && "
f"{_bootstrap_config}"
f"{_export_config} && "
f"superset db upgrade && "
f"superset fab create-admin "
f" --username admin --firstname Admin --lastname User "
f" --email admin@test.local --password {superset_admin_password} && "
f"superset init"
f"'"
)
try:
init.start()
# Block until the init command completes (container exits)
init.get_wrapped_container().wait()
exit_code = init.get_wrapped_container().attrs["State"]["ExitCode"]
if exit_code != 0:
logs = init.get_logs()
init.stop()
raise RuntimeError(
f"Superset init container failed (exit {exit_code}):\n{logs}"
)
finally:
init.stop()
# ── Step 2: Web server container ──
superset = DockerContainer(superset_image)
superset.with_env("SUPERSET_SECRET_KEY", superset_secret_key)
superset.with_env("SUPERSET_DB_URI", superset_db_url)
superset.with_env("SUPERSET_LOAD_EXAMPLES", "no")
superset.with_env("FLASK_APP", "superset")
# Build web server command: optionally write certs, then start superset
if use_tls and _write_certs:
web_cmd = (
f"sh -c '"
f"python -m pip install psycopg2-binary -q && "
f"{_write_certs} && "
f"{_bootstrap_config}"
f"{_export_config} && "
f"{_superset_run}'"
)
else:
web_cmd = (
f"sh -c '"
f"python -m pip install psycopg2-binary -q && "
f"{_bootstrap_config}"
f"{_export_config} && "
f"{_superset_run}'"
)
superset.with_command(web_cmd)
superset.with_exposed_ports(8088)
superset.start()
# Store protocol on the container object for superset_url fixture
superset._ss_protocol = _protocol
# Wait for the /health endpoint to respond
host = superset.get_container_host_ip()
port = superset.get_exposed_port(8088)
health_url = f"{_protocol}://{host}:{port}/health"
deadline = time.time() + 90
last_error = None
while time.time() < deadline:
try:
# verify=False in TLS mode because the custom CA hasn't been installed yet
resp = requests.get(health_url, timeout=3, verify=False)
if resp.status_code == 200:
break
except requests.RequestException as e:
last_error = e
time.sleep(2)
else:
logs = superset.get_logs()
superset.stop()
raise TimeoutError(
f"Superset did not become healthy within 90s. "
f"Last error: {last_error}\nContainer logs:\n{logs}"
)
yield superset
superset.stop()
# #endregion superset_container
# #region superset_url [C:1] [TYPE Fixture]
# @BRIEF Function-scoped — base URL of the running Superset container (auto-detects http/https).
@pytest.fixture
def superset_url(superset_container):
host = superset_container.get_container_host_ip()
port = superset_container.get_exposed_port(8088)
protocol = getattr(superset_container, '_ss_protocol', 'http')
return f"{protocol}://{host}:{port}"
# #endregion superset_url
# #region superset_admin_headers [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — authenticated session headers for Superset admin API calls.
@pytest.fixture
def superset_admin_headers(superset_url, superset_admin_password):
"""Log in as admin and return headers with CSRF token and session cookie."""
session = requests.Session()
# Step 1: Get CSRF token from login page
login_url = f"{superset_url}/login/"
resp = session.get(login_url, timeout=10)
resp.raise_for_status()
csrf_token = None
if "csrf_token" in resp.text:
import re
match = re.search(r'csrf_token"\s*:\s*"([^"]+)"', resp.text)
if match:
csrf_token = match.group(1)
# Step 2: Submit login form
resp = session.post(
login_url,
data={
"username": "admin",
"password": superset_admin_password,
"csrf_token": csrf_token or "",
},
headers={"Referer": login_url},
timeout=10,
)
resp.raise_for_status()
return session
# #endregion superset_admin_headers
# ── JWT-based Superset Fixtures ────────────────────────────────────
# ADR-0012: после установки psycopg2 + superset_config.py + Docker bridge IP,
# JWT-авторизация (/api/v1/security/login) работает в тестовом контейнере.
# #region superset_jwt_headers [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — JWT Bearer token headers for Superset REST API.
# @POST Returns dict with Authorization: Bearer <access_token>.
@pytest.fixture
def superset_jwt_headers(superset_url, superset_admin_password):
"""Obtain JWT access token from the running Superset container."""
resp = requests.post(
f"{superset_url}/api/v1/security/login",
json={
"username": "admin",
"password": superset_admin_password,
"provider": "db",
},
timeout=10,
)
resp.raise_for_status()
data = resp.json()
access_token = data["access_token"]
assert access_token, "No access_token in JWT response"
return {"Authorization": f"Bearer {access_token}"}
# #endregion superset_jwt_headers
# #region superset_env [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — real Environment pointing to the running Superset container.
@pytest.fixture
def superset_env(superset_url, superset_admin_password):
"""Create a real Environment config pointing to the Superset container."""
from src.core.config_models import Environment
return Environment(
id="test_env",
name="Test Superset Container",
url=superset_url,
username="admin",
password=superset_admin_password,
verify_ssl=False,
timeout=30,
)
# #endregion superset_env
# #region superset_client [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — real SupersetClient connected to the container.
@pytest.fixture
async def superset_client(superset_env):
"""Create and authenticate a SupersetClient against the real container."""
from src.core.superset_client import SupersetClient
client = SupersetClient(superset_env)
await client.authenticate()
return client
# #endregion superset_client
# ── TLS / Custom CA Fixtures ─────────────────────────────────────
# #region install_custom_ca [C:2] [TYPE Fixture]
# @BRIEF Function-scoped — installs Test Root CA into system trust store.
# Tries update-ca-certificates (root required), falls back to SSL_CERT_DIR.
# @POST Root CA accessible via ssl.create_default_context().
# @SIDE_EFFECT Modifies /usr/local/share/ca-certificates/custom/ and runs update-ca-certificates.
# On fallback: sets SSL_CERT_DIR env var, creates temp dir with hash symlinks.
@pytest.fixture
def install_custom_ca(ca_chain):
"""Install the custom Root CA into the system trust store.
Strategy 1 (prod-like): write .crt → update-ca-certificates (requires root).
Strategy 2 (fallback): SSL_CERT_DIR with hash symlinks.
"""
import subprocess
import tempfile
from pathlib import Path
ca_pem = ca_chain["root_crt"]
used_update_ca = False
ca_verify_path = None
cert_dir = None
# Strategy 1: update-ca-certificates (matches production install_certificates())
try:
cert_dir = Path("/usr/local/share/ca-certificates/custom")
cert_dir.mkdir(parents=True, exist_ok=True)
ca_path = cert_dir / "test_ss_tools_ca.crt"
ca_path.write_text(ca_pem)
subprocess.run(
["update-ca-certificates", "--fresh"],
check=True, timeout=30, capture_output=True,
)
ca_verify_path = "/etc/ssl/certs"
used_update_ca = True
except (subprocess.CalledProcessError, PermissionError, OSError):
# Strategy 2: fallback — SSL_CERT_DIR
cert_dir = Path(tempfile.mkdtemp(prefix="test_ca_"))
ca_path = cert_dir / "test_ss_tools_ca.pem"
ca_path.write_text(ca_pem)
# Create hash symlink for capath
result = subprocess.run(
["openssl", "x509", "-hash", "-noout"],
input=ca_pem, capture_output=True, text=True, timeout=10,
)
cert_hash = result.stdout.strip()
symlink_path = cert_dir / f"{cert_hash}.0"
if symlink_path.exists():
symlink_path.unlink()
symlink_path.symlink_to(ca_path.name)
os.environ["SSL_CERT_DIR"] = str(cert_dir)
ca_verify_path = str(cert_dir)
used_update_ca = False
yield {
"ca_verify_path": ca_verify_path,
"used_update_ca": used_update_ca,
"ca_pem": ca_pem,
}
# Cleanup
if used_update_ca:
ca_file = Path("/usr/local/share/ca-certificates/custom/test_ss_tools_ca.crt")
if ca_file.exists():
ca_file.unlink()
try:
subprocess.run(
["update-ca-certificates", "--fresh"],
timeout=30, capture_output=True,
)
except Exception:
pass
else:
if "SSL_CERT_DIR" in os.environ:
del os.environ["SSL_CERT_DIR"]
if cert_dir and cert_dir.exists():
import shutil
shutil.rmtree(str(cert_dir), ignore_errors=True)
# #endregion install_custom_ca
# #endregion IntegrationTestConftest
Reference in New Issue View Git Blame Copy Permalink