290 lines
8.6 KiB
Python
290 lines
8.6 KiB
Python
import pytest
|
|
import os
|
|
import json
|
|
|
|
from tests.browser import create_chrome_driver
|
|
|
|
|
|
def get_run_id(config, prefix=None):
|
|
|
|
run_prefix = prefix or _get_run_prefix(config)
|
|
run_ids = getattr(config, "_pxil_run_ids", {})
|
|
if run_prefix in run_ids:
|
|
return run_ids[run_prefix]
|
|
|
|
base_folder = os.path.join("reports", "screenshots")
|
|
os.makedirs(base_folder, exist_ok=True)
|
|
|
|
existing = [
|
|
name for name in os.listdir(base_folder)
|
|
if name.startswith(f"{run_prefix}_")
|
|
]
|
|
|
|
if not existing:
|
|
next_number = 1
|
|
else:
|
|
numbers = [
|
|
int(name.split("_")[-1])
|
|
for name in existing
|
|
if name.split("_")[-1].isdigit()
|
|
]
|
|
next_number = max(numbers, default=0) + 1
|
|
|
|
run_id = f"{run_prefix}_{next_number:03d}"
|
|
run_ids[run_prefix] = run_id
|
|
config._pxil_run_ids = run_ids
|
|
return run_id
|
|
|
|
|
|
def pytest_configure(config):
|
|
|
|
run_id = get_run_id(config)
|
|
|
|
if config.pluginmanager.hasplugin("html"):
|
|
reports_dir = os.path.abspath(os.path.join("reports", "html"))
|
|
os.makedirs(reports_dir, exist_ok=True)
|
|
config.option.htmlpath = os.path.join(reports_dir, f"{run_id}.html")
|
|
config.option.self_contained_html = True
|
|
|
|
|
|
def pytest_collection_modifyitems(config, items):
|
|
"""Keep the full setup run limited to the shared combined flow."""
|
|
invocation = " ".join(
|
|
str(arg).replace("\\", "/") for arg in config.invocation_params.args
|
|
).lower()
|
|
full_setup_run = (
|
|
"tests/setup" in invocation
|
|
and "tests/setup/sessions" not in invocation
|
|
and "tests/setup/tokens" not in invocation
|
|
and "test_session.py" not in invocation
|
|
and "test_trading_token.py" not in invocation
|
|
)
|
|
if not full_setup_run:
|
|
return
|
|
|
|
deselected = [
|
|
item for item in items
|
|
if "/tests/setup/sessions/" in f"/{item.nodeid.replace('\\', '/')}"
|
|
or "/tests/setup/tokens/" in f"/{item.nodeid.replace('\\', '/')}"
|
|
]
|
|
items[:] = [item for item in items if item not in deselected]
|
|
config.hook.pytest_deselected(items=deselected)
|
|
|
|
|
|
# -----------------------------
|
|
# WebDriver Fixture
|
|
# -----------------------------
|
|
@pytest.fixture(scope="module")
|
|
def driver():
|
|
|
|
driver = create_chrome_driver()
|
|
driver.maximize_window()
|
|
|
|
yield driver
|
|
|
|
driver.quit()
|
|
|
|
|
|
# -----------------------------
|
|
# Generate Incremental Test ID
|
|
# -----------------------------
|
|
@pytest.fixture
|
|
def test_context(request):
|
|
|
|
test_id = get_run_id(request.config, _setup_item_prefix(request.node))
|
|
test_name = request.node.name
|
|
|
|
folder = os.path.abspath(
|
|
os.path.join("reports", "screenshots", test_id, test_name)
|
|
)
|
|
log_folder = os.path.abspath(
|
|
os.path.join("reports", "logs", test_id, test_name)
|
|
)
|
|
|
|
os.makedirs(folder, exist_ok=True)
|
|
os.makedirs(log_folder, exist_ok=True)
|
|
|
|
print(f"\nTEST ID: {test_id}")
|
|
|
|
return {
|
|
"test_id": test_id,
|
|
"folder": folder,
|
|
"log_folder": log_folder,
|
|
"steps": [],
|
|
"extra_folders": [],
|
|
}
|
|
|
|
|
|
# -----------------------------
|
|
# Attach Screenshot Folder Link
|
|
# -----------------------------
|
|
@pytest.hookimpl(hookwrapper=True)
|
|
def pytest_runtest_makereport(item, call):
|
|
|
|
outcome = yield
|
|
report = outcome.get_result()
|
|
|
|
if report.when == "call":
|
|
|
|
context = item.funcargs.get("test_context", None)
|
|
|
|
if context:
|
|
|
|
folder = context["folder"]
|
|
steps = context.get("steps", [])
|
|
|
|
if report.failed:
|
|
driver = item.funcargs.get("driver", None)
|
|
failure_index = len(steps) + 1
|
|
failure_name = "failure_state"
|
|
failure_filename = f"{failure_index:02d}_{failure_name}.png"
|
|
failure_path = os.path.join(folder, failure_filename)
|
|
|
|
if driver:
|
|
try:
|
|
driver.save_screenshot(failure_path)
|
|
steps.append(
|
|
{
|
|
"index": failure_index,
|
|
"name": failure_name,
|
|
"path": failure_path,
|
|
"status": "failed",
|
|
}
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
if item.config.pluginmanager.hasplugin("html"):
|
|
|
|
from pytest_html import extras
|
|
|
|
extras_list = getattr(report, "extras", [])
|
|
|
|
for step in steps:
|
|
extras_list.append(
|
|
extras.url(
|
|
_as_file_url(step["path"]),
|
|
name=f"Step {step['index']:02d} [{step['status']}] {step['name']}"
|
|
)
|
|
)
|
|
|
|
if steps:
|
|
extras_list.append(
|
|
extras.html(_build_steps_html(steps))
|
|
)
|
|
|
|
extras_list.append(
|
|
extras.url(_as_file_url(folder), name="Screenshots Folder")
|
|
)
|
|
|
|
log_folder = context.get("log_folder")
|
|
if log_folder:
|
|
extras_list.append(
|
|
extras.url(_as_file_url(log_folder), name="Logs Folder")
|
|
)
|
|
|
|
for extra_folder in context.get("extra_folders", []):
|
|
extras_list.append(
|
|
extras.url(
|
|
_as_file_url(extra_folder["path"]),
|
|
name=extra_folder["name"]
|
|
)
|
|
)
|
|
|
|
report.extras = extras_list
|
|
|
|
|
|
def _as_file_url(path):
|
|
|
|
absolute_path = os.path.abspath(path).replace("\\", "/")
|
|
return f"file:///{absolute_path}"
|
|
|
|
|
|
def _build_steps_html(steps):
|
|
|
|
rows = []
|
|
for step in steps:
|
|
step_url = _as_file_url(step["path"])
|
|
rows.append(
|
|
"<tr>"
|
|
f"<td>{step['index']:02d}</td>"
|
|
f"<td>{step['status'].capitalize()}</td>"
|
|
f"<td>{step['name']}</td>"
|
|
f"<td><a href=\"{step_url}\">Open Screenshot</a></td>"
|
|
"</tr>"
|
|
)
|
|
|
|
return (
|
|
"<div>"
|
|
"<h4>Step Details</h4>"
|
|
"<table border='1' cellpadding='4' cellspacing='0' "
|
|
"style='border-collapse: collapse; margin-top: 8px;'>"
|
|
"<thead><tr><th>#</th><th>Status</th><th>Step</th><th>Screenshot</th></tr></thead>"
|
|
f"<tbody>{''.join(rows)}</tbody>"
|
|
"</table>"
|
|
"</div>"
|
|
)
|
|
|
|
|
|
def _setup_item_prefix(item):
|
|
"""Use the collected product and flow, including when -k selects a subset."""
|
|
nodeid = item.nodeid.replace("\\", "/")
|
|
if "tests/setup/" not in nodeid:
|
|
return None
|
|
callspec = getattr(item, "callspec", None)
|
|
product_config = callspec.params.get("config") if callspec else None
|
|
if not isinstance(product_config, dict):
|
|
return None
|
|
product = product_config["product_name"].upper()
|
|
if "test_combined_setup.py" in nodeid:
|
|
return f"{product}_SESSION_TOKEN"
|
|
if "/sessions/" in nodeid:
|
|
return f"{product}_SESSION"
|
|
if "/tokens/" in nodeid:
|
|
return f"{product}_TOKEN"
|
|
return f"{product}_SETUP"
|
|
|
|
|
|
def _get_run_prefix(config):
|
|
|
|
invocation_args = " ".join(str(arg) for arg in config.invocation_params.args).replace("\\", "/").lower()
|
|
|
|
if "tests/setup/sessions" in invocation_args or "test_session.py" in invocation_args:
|
|
return "PXIL_SESSION"
|
|
|
|
if "tests/setup/tokens" in invocation_args or "test_trading_token.py" in invocation_args:
|
|
return "PXIL_TOKEN"
|
|
|
|
if "tests/setup" in invocation_args or "test_combined_setup.py" in invocation_args:
|
|
return "PXIL_SETUP"
|
|
|
|
runtime_path = os.path.abspath(os.path.join("config", "runtime.json"))
|
|
if os.path.exists(runtime_path):
|
|
try:
|
|
with open(runtime_path, "r", encoding="utf-8") as runtime_file:
|
|
runtime_config = json.load(runtime_file)
|
|
|
|
execution = runtime_config.get("execution", {})
|
|
product_family = execution.get("product_family", "regression").upper()
|
|
flow_scope = execution.get("flow_scope", "run").upper()
|
|
return f"{product_family}_{flow_scope}"
|
|
except Exception:
|
|
pass
|
|
|
|
if "matching" in invocation_args:
|
|
return "REGRESSION_MATCHING"
|
|
|
|
if "bidding" in invocation_args:
|
|
return "REGRESSION_BIDDING"
|
|
|
|
if "reports" in invocation_args:
|
|
return "REGRESSION_REPORTS"
|
|
|
|
if "order_book" in invocation_args:
|
|
return "REGRESSION_ORDER_BOOK"
|
|
|
|
if "trade_book" in invocation_args:
|
|
return "REGRESSION_TRADE_BOOK"
|
|
|
|
return "PXIL_RUN"
|