Автоматизация: pytest plugin для flaky tests

import json
import pytest
 
def pytest_addoption(parser):
    parser.addoption("--slow-threshold", action="store", type=float, default=0.5)
    parser.addoption("--flaky-cache", action="store", default=None, help="путь к json со списком flaky")
 
def pytest_configure(config):
    config.addinivalue_line("markers", "contract: контрактные тесты API")
    config.addinivalue_line("markers", "flaky: нестабильный тест, требует исправления")
    config._slow_tests = []
    config._suspected_flaky = set()
    if cache := config.getoption("--flaky-cache"):
        try:
            with open(cache) as f:
                config._suspected_flaky = set(json.load(f))
        except FileNotFoundError:
            config._suspected_flaky = set()
 
@pytest.hookimpl(tryfirst=True)
def pytest_runtest_setup(item):
    if item.nodeid in getattr(item.config, "_suspected_flaky", set()):
        item.add_marker(pytest.mark.xfail(reason="known flaky"))
 
def pytest_runtest_logreport(report):
    if report.when != "call":
        return
    cfg = report.config
    threshold = cfg.getoption("--slow-threshold")
    if report.duration > threshold:
        cfg._slow_tests.append((report.nodeid, report.duration))
    if report.failed and "flaky" in report.keywords:
        cfg._suspected_flaky.add(report.nodeid)
 
@pytest.hookimpl(trylast=True)
def pytest_sessionfinish(session, exitstatus):
    slow = sorted(session.config._slow_tests, key=lambda t: t[1], reverse=True)
    reporter = session.config.pluginmanager.get_plugin("terminalreporter")
    if slow:
        reporter.write_line("\n[slow-tests] топ медленных:", red=False, bold=True)
        for nodeid, duration in slow[:10]:
            reporter.write_line(f"{nodeid} - {duration:.3f}s")
    cache_path = session.config.getoption("--flaky-cache")
    if cache_path:
        with open(cache_path, "w") as f:
            json.dump(sorted(session.config._suspected_flaky), f, indent=2)

pytest -p tests.plugins.stability --slow-threshold=0.3 --flaky-cache=artifacts/flaky.json

@pytest.fixture(autouse=True)
def mark_contract_tests(request, metrics_client):
    if "contract" in request.keywords:
        metrics_client.increment("tests.contract")