135 lines
5.2 KiB
Python
135 lines
5.2 KiB
Python
import dataclasses
|
|
from datetime import datetime
|
|
import json
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import Any, Generator
|
|
|
|
import pytest
|
|
|
|
from tests.common.component_orchestrator import ComponentOrchestrator
|
|
from tests.common import constants
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_PASSWORD = "password"
|
|
EXTRA_TEST_USERS = 5
|
|
SERVICE_USERS = [
|
|
{"username": "admin", "password": "admin", "full_name": "Admin Service Account"},
|
|
{"username": "risk_gateway", "password": "risk_gateway", "full_name": "Risk Gateway Service Account"},
|
|
]
|
|
|
|
_performance_data: list[dict[str, Any]] = []
|
|
|
|
|
|
def pytest_addoption(parser: pytest.Parser) -> None:
|
|
parser.addoption("--venv-path", action="store", default="", help="The path to the virtual environment")
|
|
parser.addoption(
|
|
"--deployment-config",
|
|
action="store",
|
|
default="deployment_config.json",
|
|
help="Path to the deployment_config.json for the deployment under test",
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def venv_path(request: pytest.FixtureRequest) -> Path:
|
|
venv_path = request.config.getoption("--venv-path")
|
|
assert venv_path is not None, "Virtual environment path is not set"
|
|
assert Path(venv_path).exists(), f"Virtual environment path {venv_path} does not exist"
|
|
return Path(venv_path)
|
|
|
|
|
|
@pytest.fixture
|
|
def deployment_config(request: pytest.FixtureRequest) -> dict:
|
|
config_path = Path(request.config.getoption("--deployment-config"))
|
|
assert config_path.exists(), f"Deployment config not found: {config_path}"
|
|
with config_path.open() as f:
|
|
return json.load(f)
|
|
|
|
def _create_data_file_with_users(test_name: str) -> Path:
|
|
"""Create a temporary data file with service accounts and test-specific users.
|
|
|
|
Creates the primary test user plus numbered extras (e.g. ``<test>_2`` …
|
|
``<test>_6``) so that multi-client tests can log in with distinct identities.
|
|
"""
|
|
users = list(SERVICE_USERS) + [
|
|
{"username": test_name, "password": DEFAULT_PASSWORD, "full_name": f"Test User ({test_name})"},
|
|
]
|
|
for i in range(2, EXTRA_TEST_USERS + 2):
|
|
username = f"{test_name}_{i}"
|
|
users.append({"username": username, "password": DEFAULT_PASSWORD, "full_name": f"Test User ({username})"})
|
|
data = {"users": users}
|
|
tmp = tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".json", prefix="data_file_")
|
|
json.dump(data, tmp, indent=2)
|
|
tmp.close()
|
|
logger.info(f"Created temporary data file with {len(users)} users: {tmp.name}")
|
|
return Path(tmp.name)
|
|
|
|
|
|
def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
|
|
"""Write collected performance data to ``performance_report.json``.
|
|
|
|
With pytest-xdist each worker writes a partial file; the controller
|
|
(or a single-process run) merges them into the final report.
|
|
"""
|
|
log_dir = Path(constants.LOG_DIRECTORY)
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
worker_input = getattr(session.config, "workerinput", None)
|
|
if worker_input is not None:
|
|
worker_id = worker_input["workerid"]
|
|
partial = log_dir / f"_perf_partial_{worker_id}.json"
|
|
partial.write_text(json.dumps(_performance_data, indent=2))
|
|
else:
|
|
all_data: list[dict[str, Any]] = list(_performance_data)
|
|
for partial in sorted(log_dir.glob("_perf_partial_*.json")):
|
|
all_data.extend(json.loads(partial.read_text()))
|
|
partial.unlink()
|
|
if all_data:
|
|
now = datetime.now().strftime('%Y%m%d_%H%M%S')
|
|
report_path = log_dir / f"performance_report_{now}.json"
|
|
report_path.write_text(json.dumps(all_data, indent=2))
|
|
logger.info(f"Performance report written to {report_path}")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def start_components(
|
|
request: pytest.FixtureRequest, venv_path: Path, deployment_config: dict
|
|
) -> Generator[None, None, None]:
|
|
"""Automatically start all components needed for the test class's declared PROTOCOL.
|
|
|
|
Any test class with a ``PROTOCOL`` class attribute (matching a protocol name
|
|
from the deployment-config schema) will have the required components started
|
|
before each test and stopped afterwards. The component implementing the
|
|
declared protocol is exposed as ``self.server_address`` on the test instance.
|
|
"""
|
|
protocol: str | None = getattr(request.cls, "PROTOCOL", None)
|
|
if protocol is None:
|
|
yield
|
|
return
|
|
|
|
test_name = request.node.name
|
|
data_file_path = _create_data_file_with_users(test_name)
|
|
|
|
try:
|
|
orchestrator = ComponentOrchestrator(venv_path, deployment_config, data_file_path)
|
|
orchestrator.start_for_protocol(protocol)
|
|
|
|
if request.instance is not None:
|
|
request.instance.server_address = orchestrator.get_server_address(protocol)
|
|
request.instance.orchestrator = orchestrator
|
|
request.instance.auth_required = orchestrator.is_auth_required(protocol)
|
|
|
|
yield
|
|
|
|
perf_stats = orchestrator.stop_all()
|
|
if perf_stats:
|
|
_performance_data.append({
|
|
"test": test_name,
|
|
"components": {name: dataclasses.asdict(stats) for name, stats in perf_stats.items()},
|
|
})
|
|
finally:
|
|
data_file_path.unlink(missing_ok=True)
|