plugin.py

"""VIP pytest plugin.

Registered via the ``pytest11`` entry point so it activates automatically
when the ``vip`` package is installed.

Responsibilities:
- Register custom markers.
- Add CLI options (``--vip-config``, ``--vip-extensions``, ``--vip-report``,
  ``--interactive-auth``).
- Deselect (exclude) tests whose product is not configured.
- Auto-skip tests whose product version doesn't meet ``min_version``.
- Ensure prerequisites run before other tests.
- Collect extension directories.
- Write a JSON results file for the Quarto report.
- Handle interactive OIDC authentication for external identity providers.
"""

from __future__ import annotations

import json
import re
import sys
import threading
import time
import warnings
from collections.abc import Generator
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import pytest

from vip.config import VIPConfig, load_config
from vip.version import ProductVersion

# ---------------------------------------------------------------------------
# Stash keys
# ---------------------------------------------------------------------------

_vip_config_key = pytest.StashKey[VIPConfig]()
_ext_dirs_key = pytest.StashKey[list[str]]()
_results_key = pytest.StashKey[list[dict[str, Any]]]()
_auth_session_key = pytest.StashKey[Any]()
_auth_mode_key = pytest.StashKey[str]()
_version_na_key = pytest.StashKey[bool]()

# Module-level reference to the active pytest.Config, set in pytest_configure.
# Safe because pytester runs in a subprocess (fresh import each time).
_active_config: pytest.Config | None = None

# Color of the result line currently being rendered, set in
# pytest_runtest_logreport (tryfirst, so it lands before the terminal reporter
# renders the same report).  Consumed by the progress-indicator recolor wrapper
# installed on the terminal reporter.  ``None`` means "fall back to pytest's
# default color".  Module-level is safe: this only matters in a single,
# non-xdist process (see _install_progress_recolor).
_current_line_color: str | None = None

# Mapping from pytest marker name to product config key.
_PRODUCT_MARKERS = {
    "connect": "connect",
    "workbench": "workbench",
    "package_manager": "package_manager",
}

# ---------------------------------------------------------------------------
# Plugin hooks
# ---------------------------------------------------------------------------


def pytest_addoption(parser: pytest.Parser) -> None:
    group = parser.getgroup("vip", "Verified Installation of Posit")
    group.addoption(
        "--vip-config",
        default=None,
        help="Path to vip.toml configuration file.",
    )
    group.addoption(
        "--vip-extensions",
        action="append",
        default=[],
        help="Additional directories containing custom VIP test cases (repeatable).",
    )
    group.addoption(
        "--vip-report",
        default="report/results.json",
        help="Write a JSON results file at this path for Quarto report generation."
        " Set to empty string to disable. (default: report/results.json)",
    )
    group.addoption(
        "--interactive-auth",
        action="store_true",
        default=False,
        help="Launch a browser for manual OIDC login before running tests.",
    )
    group.addoption(
        "--headless-auth",
        action="store_true",
        default=False,
        help="Automate login in a headless browser (OIDC/SAML/OAuth2 requires [auth] idp).",
    )
    group.addoption(
        "--no-auth",
        action="store_true",
        default=False,
        help="Skip all tests that require authentication credentials (Connect and Workbench).",
    )
    group.addoption(
        "--api-auth",
        action="store_true",
        default=False,
        help="Run only API-key-authenticated tests; skip tests that require browser credentials.",
    )
    group.addoption(
        "--vip-verbose",
        action="store_true",
        default=False,
        help="Show full pytest tracebacks instead of concise error messages.",
    )


def pytest_configure(config: pytest.Config) -> None:
    global _active_config
    _active_config = config

    # Register the canonical warning filters in the plugin so they apply
    # regardless of the pytest rootdir, including when vip is installed into
    # an unrelated project.
    for line in (
        # gherkin-official 29.0.0 passes maxsplit positionally to re.split;
        # fixed upstream but pulled transitively via pytest-bdd.
        "ignore:'maxsplit' is passed as positional argument:DeprecationWarning:gherkin",
        # TLS downgrade tests deliberately use deprecated TLS 1.0/1.1 versions.
        "ignore:ssl.TLSVersion.TLSv1:DeprecationWarning",
        "ignore:ssl.TLSVersion.TLSv1_1:DeprecationWarning",
        # pytest-bdd scenario functions return fixture values; not a real issue.
        "ignore::pytest.PytestReturnNotNoneWarning",
        # pytest-bdd 8.1.0 injects fixtures via _register_fixture(nodeid=...) and
        # FixtureDef(baseid=...), which pytest >= 9.1 deprecates in favor of node=.
        # Framework-level and only fixable when pytest-bdd updates; nothing VIP can
        # change. A category filter is required because the warning surfaces from two
        # modules (pytest_bdd and _pytest.fixtures), so module/message scoping misses one.
        "ignore::pytest.PytestRemovedIn10Warning",
        # gevent monkey-patching happens after ssl is imported by other plugins;
        # unavoidable without patching at process start.
        "ignore:Monkey-patching ssl",
    ):
        config.addinivalue_line("filterwarnings", line)

    # Register markers
    config.addinivalue_line("markers", "connect: tests for Posit Connect")
    config.addinivalue_line("markers", "workbench: tests for Posit Workbench")
    config.addinivalue_line("markers", "package_manager: tests for Posit Package Manager")
    config.addinivalue_line("markers", "prerequisites: prerequisite checks")
    config.addinivalue_line("markers", "cross_product: cross-product / admin tests")
    config.addinivalue_line(
        "markers",
        "performance: performance validation tests (opt-in; excluded by default)",
    )
    config.addinivalue_line("markers", "security: security validation tests")
    config.addinivalue_line(
        "markers",
        "config_hygiene: checks of VIP's own configuration (opt-in; excluded by default)",
    )
    config.addinivalue_line(
        "markers",
        "slow: detailed/long-running checks; excluded by --basic",
    )
    config.addinivalue_line(
        "markers",
        "min_version(product, version): skip when product is below the specified version",
    )
    config.addinivalue_line(
        "markers",
        "if_applicable: skip when the related feature is not configured",
    )
    config.addinivalue_line(
        "markers",
        "api_auth: test requires only an API key, not browser credentials",
    )
    config.addinivalue_line("markers", "rstudio: Workbench RStudio IDE-launch scenario")
    config.addinivalue_line("markers", "vscode: Workbench VS Code IDE-launch scenario")
    config.addinivalue_line("markers", "jupyter: Workbench JupyterLab IDE-launch scenario")
    config.addinivalue_line("markers", "positron: Workbench Positron IDE-launch scenario")

    # In concise mode, suppress the "short test summary info" section — the
    # inline concise error messages make it redundant.
    if not config.getoption("--vip-verbose", default=False):
        config.option.reportchars = ""

    # Show skip/xfail reasons in full on the verbose (``-v``) test line instead
    # of ellipsizing them to the terminal width. pytest only prints the
    # untrimmed reason at *test-case* verbosity >= 2 (see _pytest/terminal.py).
    # Bumping that fine-grained level — rather than the global ``-v`` count —
    # leaves failure tracebacks and assertion reprs at the user's chosen
    # verbosity, and a skip reason never carries a traceback, so this only ever
    # lengthens a one-line reason. We act only when the user passed exactly
    # ``-v`` (resolved test-case verbosity 1): at level 0 the reporter is in
    # dot mode and bumping would force per-test lines on; above 1 it is already
    # full. An explicit ``verbosity_test_cases`` in the user's config (anything
    # other than the "auto" default) is respected.
    try:
        if (
            config.getini("verbosity_test_cases") == "auto"
            and config.get_verbosity(pytest.Config.VERBOSITY_TEST_CASES) == 1
        ):
            config._inicache["verbosity_test_cases"] = "2"
    except (ValueError, AttributeError):
        pass

    # Load VIP config and stash it for fixtures / collection hooks.
    vip_cfg = load_config(config.getoption("--vip-config"))
    config.stash[_vip_config_key] = vip_cfg

    # Initialize per-session results list (avoids module-level global).
    config.stash[_results_key] = []

    # Merge extension dirs from config file and CLI.
    ext_dirs: list[str] = list(vip_cfg.extension_dirs)
    ext_dirs.extend(config.getoption("--vip-extensions") or [])
    config.stash[_ext_dirs_key] = ext_dirs

    # Handle interactive auth — login via browser, then close before tests.
    # With pytest-xdist the browser auth happens once in the controller;
    # credentials are forwarded to workers via pytest_configure_node.
    config.stash[_auth_session_key] = None

    if hasattr(config, "workerinput"):
        # xdist worker — restore auth data shared by the controller (if any).
        # Workers must never re-run the controller-only browser-auth branches
        # below: pytest_configure fires on every worker, so doing so would, for
        # example, re-emit the "no auth-requiring products" warning once per
        # worker, flooding the output.
        if config.workerinput.get("vip_interactive_auth"):
            _restore_worker_auth(config, vip_cfg)
        elif config.workerinput.get("vip_auth_mode"):
            # No browser session was forwarded (no auth-requiring product), but
            # still mirror the controller's auth mode so the auth_mode fixture
            # matches a non-xdist run.
            config.stash[_auth_mode_key] = config.workerinput["vip_auth_mode"]
    elif config.getoption("--interactive-auth"):
        config.stash[_auth_mode_key] = "interactive"
        connect_url = vip_cfg.connect.url if vip_cfg.connect.is_configured else None
        wb_url = vip_cfg.workbench.url if vip_cfg.workbench.is_configured else None

        if not connect_url and not wb_url:
            # No auth-requiring product is configured (e.g. only Package Manager).
            # Skip the browser flow entirely rather than erroring out.
            warnings.warn(
                "VIP: --interactive-auth was requested but no auth-requiring products "
                "(Connect, Workbench) are configured; skipping browser authentication.",
                stacklevel=1,
            )
        else:
            from vip.auth import start_interactive_auth

            cache_path = Path(config.rootpath) / ".vip-auth-cache.json"
            session = start_interactive_auth(
                connect_url=connect_url,
                workbench_url=wb_url,
                cache_path=cache_path,
                insecure=vip_cfg.insecure,
                ca_bundle=vip_cfg.ca_bundle,
            )
            config.stash[_auth_session_key] = session
            if session.api_key:
                vip_cfg.connect.api_key = session.api_key
            # Auth may have rewritten the Connect URL (sub-path dashboard +
            # root API).  Sync so the test clients hit the same base mint did.
            if session._connect_url and connect_url and session._connect_url != connect_url:
                vip_cfg.connect.url = session._connect_url
            if not session.api_key and connect_url:
                warnings.warn(
                    "VIP: --interactive-auth could not mint an API key. "
                    "API-based tests will likely fail. "
                    "Try again or set VIP_CONNECT_API_KEY to fix.",
                    stacklevel=1,
                )
    elif config.getoption("--headless-auth"):
        config.stash[_auth_mode_key] = "headless"
        connect_url = vip_cfg.connect.url if vip_cfg.connect.is_configured else None
        wb_url = vip_cfg.workbench.url if vip_cfg.workbench.is_configured else None

        if not connect_url and not wb_url:
            # No auth-requiring product is configured (e.g. only Package Manager).
            # Skip the browser flow entirely rather than erroring out.
            warnings.warn(
                "VIP: --headless-auth was requested but no auth-requiring products "
                "(Connect, Workbench) are configured; skipping browser authentication.",
                stacklevel=1,
            )
        else:
            from vip.auth import AuthConfigError, start_headless_auth

            cache_path = Path(config.rootpath) / ".vip-auth-cache.json"
            try:
                session = start_headless_auth(
                    connect_url=connect_url,
                    workbench_url=wb_url,
                    idp=vip_cfg.auth.idp,
                    provider=vip_cfg.auth.provider,
                    username=vip_cfg.auth.username,
                    password=vip_cfg.auth.password,
                    cache_path=cache_path,
                    verbose=config.getoption("--vip-verbose", default=False),
                    insecure=vip_cfg.insecure,
                    ca_bundle=vip_cfg.ca_bundle,
                )
            except AuthConfigError as exc:
                raise pytest.UsageError(str(exc)) from None
            config.stash[_auth_session_key] = session
            if session.api_key:
                vip_cfg.connect.api_key = session.api_key
            # Auth may have rewritten the Connect URL (sub-path dashboard +
            # root API).  Sync so the test clients hit the same base mint did.
            if session._connect_url and connect_url and session._connect_url != connect_url:
                vip_cfg.connect.url = session._connect_url
            if not session.api_key and connect_url:
                warnings.warn(
                    "VIP: --headless-auth could not mint an API key. "
                    "API-based tests will likely fail. "
                    "Try again or set VIP_CONNECT_API_KEY to fix.",
                    stacklevel=1,
                )


def _restore_worker_auth(config: pytest.Config, vip_cfg: VIPConfig) -> None:
    """Reconstruct an auth session in an xdist worker from controller data."""
    from vip.auth import InteractiveAuthSession

    wi = config.workerinput  # type: ignore[attr-defined]  # xdist injects this
    api_key = wi.get("vip_api_key") or None
    storage_state = wi.get("vip_storage_state", "")
    connect_url = wi.get("vip_connect_url", "") or ""

    if api_key:
        vip_cfg.connect.api_key = api_key
    # Mirror the controller's URL rewrite (split sub-path dashboard + root
    # API).  Without this, workers build ConnectClient from the original
    # sub-path URL and 404 against /__api__/.
    if connect_url and vip_cfg.connect.is_configured and connect_url != vip_cfg.connect.url:
        vip_cfg.connect.url = connect_url

    session = InteractiveAuthSession(
        storage_state_path=Path(storage_state) if storage_state else Path("/dev/null"),
        api_key=api_key,
        key_name=wi.get("vip_key_name", ""),
        workbench_auth_error=wi.get("vip_workbench_auth_error") or None,
        _connect_url=connect_url,
        _workbench_url=wi.get("vip_workbench_url", "") or "",
        _tmpdir="",  # Workers don't own the temp dir; controller cleans up.
    )
    config.stash[_auth_session_key] = session
    mode = wi.get("vip_auth_mode")
    if mode:
        config.stash[_auth_mode_key] = mode


def pytest_configure_node(node) -> None:
    """xdist controller hook: share interactive-auth credentials with workers."""
    # Forward the auth mode even when no browser session was established (e.g.
    # only Package Manager configured, so the flow was skipped). Workers don't
    # re-run the controller-only auth branch, so without this their auth_mode
    # fixture would resolve to "none" while a non-xdist run reports the mode.
    mode = node.config.stash.get(_auth_mode_key, "")
    if mode:
        node.workerinput["vip_auth_mode"] = mode
    auth = node.config.stash.get(_auth_session_key, None)
    if auth is not None:
        node.workerinput["vip_interactive_auth"] = True
        node.workerinput["vip_api_key"] = auth.api_key or ""
        node.workerinput["vip_storage_state"] = str(auth.storage_state_path)
        node.workerinput["vip_key_name"] = auth.key_name
        node.workerinput["vip_connect_url"] = auth._connect_url
        node.workerinput["vip_workbench_url"] = auth._workbench_url
        node.workerinput["vip_workbench_auth_error"] = auth.workbench_auth_error or ""


def _outcome_color(report: pytest.TestReport) -> str | None:
    """Return the pytest markup color for a single result line's outcome.

    Mirrors pytest's own per-letter markup (see
    ``TerminalReporter.pytest_runtest_logreport``): green for a plain pass,
    yellow for xfail/xpass/skip, red for a failure or error.  Returns ``None``
    for the setup/teardown phases and any outcome we don't recolor, so the
    caller falls back to pytest's default (cumulative) color.

    Only the ``call`` phase — plus a ``setup`` that skipped, which is how an
    ordinary skip surfaces — carries a line the user sees, so other phases are
    ignored to avoid recoloring a percentage that belongs to a passing test.
    """
    if report.when not in ("call", "setup"):
        return None
    if report.when == "setup" and not report.skipped:
        return None
    if report.failed:
        return "red"
    if report.skipped or hasattr(report, "wasxfail"):
        return "yellow"
    if report.passed:
        return "green"
    return None


def _install_progress_recolor(config: pytest.Config) -> None:
    """Color each line's trailing progress indicator by that line's outcome.

    pytest colors the ``[ 42%]`` progress indicator with the session's
    *cumulative* "main color": once any test fails, ``_get_main_color`` returns
    red and every subsequent line's indicator is red too — making a run look far
    more broken than it is.  We want only the failing line's indicator red.

    ``_write_progress_information_filling_space`` reads the color from
    ``self._get_main_color()``.  We wrap it to temporarily swap in the color of
    the line just reported (captured in ``pytest_runtest_logreport``), then
    restore the real one so the end-of-session summary still uses pytest's
    cumulative color.

    Skipped under xdist: the controller renders progress on its own
    ``pytest_runtest_logreport`` path (the ``running_xdist`` branch), which uses
    a fixed cyan indicator and never calls the wrapped method, so there is
    nothing to recolor and the module-level ``_current_line_color`` would race
    across workers besides.
    """
    if config.getoption("--vip-verbose", default=False):
        return
    if hasattr(config, "workerinput"):
        return
    tr = config.pluginmanager.get_plugin("terminalreporter")
    if tr is None:
        return

    original_fill = tr._write_progress_information_filling_space
    original_get_main_color = tr._get_main_color

    def recolored_fill(*args: Any, **kwargs: Any) -> Any:
        if _current_line_color is None:
            return original_fill(*args, **kwargs)
        # Swap _get_main_color to report this line's color, then restore it so
        # nothing else (summary line, past-edge writes) is affected.
        known = original_get_main_color()[1]
        tr._get_main_color = lambda: (_current_line_color, known)  # type: ignore[method-assign]
        try:
            return original_fill(*args, **kwargs)
        finally:
            tr._get_main_color = original_get_main_color  # type: ignore[method-assign]

    tr._write_progress_information_filling_space = recolored_fill  # type: ignore[method-assign]


# The installed test package root.  pytest node paths are "/"-normalized
# (see _pytest.terminal._locationline), so the marker uses a forward slash on
# every platform.
_TEST_PKG_MARKER = "vip_tests/"


def _shorten_location_line(line: str) -> str:
    """Drop the install-path prefix from a pytest ``-v`` location line.

    ``vip verify`` runs pytest with ``-v``, so every result line is prefixed
    with the node's path.  When VIP is installed as a tool that path is the
    long site-packages location, e.g.::

        ../../../.local/share/uv/tools/posit-vip/lib/python3.13/site-packages/vip_tests/connect/test_auth.py::test_connect_login_ui

    Keeping only the portion after the ``vip_tests/`` package root puts the
    meaningful part first::

        connect/test_auth.py::test_connect_login_ui

    Lines without the marker (e.g. user extension tests collected from an
    arbitrary directory) are returned unchanged.  Uses ``rfind`` so a marker
    substring appearing earlier in an install path never wins over the real
    package root.
    """
    idx = line.rfind(_TEST_PKG_MARKER)
    if idx == -1:
        return line
    return line[idx + len(_TEST_PKG_MARKER) :]


def _install_location_shortener(config: pytest.Config) -> None:
    """Wrap the terminal reporter's ``_locationline`` to truncate node paths.

    Only active in concise (non-``--vip-verbose``) mode.  There is no public
    hook for the per-test location string, so we wrap the bound method on the
    registered reporter instance.  Under xdist this runs on the controller,
    which is where result lines are rendered, so worker output is covered too.
    """
    if config.getoption("--vip-verbose", default=False):
        return
    tr = config.pluginmanager.get_plugin("terminalreporter")
    if tr is None:
        return
    original = tr._locationline

    def shortened(*args: Any, **kwargs: Any) -> str:
        return _shorten_location_line(original(*args, **kwargs))

    tr._locationline = shortened  # type: ignore[method-assign]


def pytest_sessionstart(session: pytest.Session) -> None:
    """Add extension directories to sys.path so their conftest / modules
    are importable, register them for collection, recolor progress indicators
    per-line, and shorten node paths."""
    # Wrap the terminal reporter here rather than in pytest_configure: the
    # builtin terminal plugin registers "terminalreporter" in its own
    # pytest_configure, which pluggy may call after ours.  By sessionstart the
    # reporter is guaranteed to exist.
    _install_progress_recolor(session.config)
    _install_location_shortener(session.config)

    ext_dirs = session.config.stash.get(_ext_dirs_key, [])
    for d in ext_dirs:
        p = Path(d).resolve()
        if p.is_dir():
            str_p = str(p)
            if str_p not in sys.path:
                sys.path.insert(0, str_p)
            # Tell pytest to collect from this directory as well.
            session.config.args.append(str_p)


def pytest_collection_modifyitems(
    config: pytest.Config,
    items: list[pytest.Item],
) -> None:
    """Deselect tests whose product is not configured, skip tests whose
    version requirement is not met, and ensure prerequisites run first."""
    vip_cfg: VIPConfig = config.stash[_vip_config_key]
    no_auth = config.getoption("--no-auth", default=False)
    api_auth = config.getoption("--api-auth", default=False)

    # Sort so prerequisites run before everything else, and assign xdist
    # groups so that each product's tests land on a dedicated worker.
    # Tests for unconfigured products are deselected (excluded entirely)
    # rather than skipped, so they don't appear in the report.
    prerequisites: list[pytest.Item] = []
    rest: list[pytest.Item] = []
    deselected: list[pytest.Item] = []
    for item in items:
        if _should_deselect_for_product(item, vip_cfg):
            deselected.append(item)
            continue
        if no_auth and _requires_auth(item):
            deselected.append(item)
            continue
        if api_auth and _requires_auth(item) and not _is_api_auth_only(item):
            deselected.append(item)
            continue
        _maybe_skip_for_version(item, vip_cfg)
        _assign_xdist_group(item)
        _stash_scenario_metadata(item)
        if item.get_closest_marker("prerequisites"):
            prerequisites.append(item)
        else:
            rest.append(item)
    items[:] = prerequisites + rest
    if deselected:
        config.hook.pytest_deselected(items=deselected)


# Directories whose tests get a dedicated xdist worker.
_PRODUCT_DIRS = {"connect", "workbench", "package_manager"}


def _assign_xdist_group(item: pytest.Item) -> None:
    """Assign an ``xdist_group`` marker so each product runs on its own worker.

    Respects any existing ``xdist_group`` marker (set via conftest.py
    ``pytestmark`` or per-test decorators).  Otherwise, tests under
    ``tests/connect/``, ``tests/workbench/``, or ``tests/package_manager/``
    are grouped by directory name, and everything else (prerequisites,
    cross_product, performance, security) lands in a shared ``general`` group.
    """
    if item.get_closest_marker("xdist_group") is not None:
        return
    fspath = getattr(item, "path", None) or Path()
    dir_name = fspath.parent.name
    group = dir_name if dir_name in _PRODUCT_DIRS else "general"
    item.add_marker(pytest.mark.xdist_group(group))


# Mapping from "Given" step name prefix to product config key.
# Steps like "Connect is configured in vip.toml" gate a scenario on
# a product being configured; when it isn't, the test should be
# deselected rather than skipped so it doesn't clutter the output.
_GIVEN_PRODUCT_STEPS = {
    "Connect is configured": "connect",
    "Workbench is configured": "workbench",
    "Package Manager is configured": "package_manager",
}

# Display name → config key for parameterized "<product>" placeholders.
_PRODUCT_DISPLAY_NAMES = {
    "Connect": "connect",
    "Workbench": "workbench",
    "Package Manager": "package_manager",
}


def _get_bdd_param_product(item: pytest.Item) -> str | None:
    """Extract the product display name from a pytest-bdd parameterized item.

    pytest-bdd stores Scenario Outline examples in ``callspec.params`` as
    ``{'_pytest_bdd_example': {'product': 'Connect', ...}}``.
    """
    callspec = getattr(item, "callspec", None)
    if callspec is None:
        return None
    example = callspec.params.get("_pytest_bdd_example")
    if isinstance(example, dict):
        return example.get("product")
    return None


def _should_deselect_for_product(item: pytest.Item, cfg: VIPConfig) -> bool:
    """Return True if *item* should be deselected because its product is not configured."""
    # Check explicit product markers (@connect, @workbench, @package_manager).
    for marker_name, product_key in _PRODUCT_MARKERS.items():
        marker = item.get_closest_marker(marker_name)
        if marker is not None:
            pc = cfg.product_config(product_key)
            if not pc.is_configured:
                return True

    # Check BDD scenario "Given" steps for product-configuration guards.
    fn = getattr(item, "obj", None)
    scenario_obj = getattr(fn, "__scenario__", None) if fn else None
    if scenario_obj is not None:
        for step in getattr(scenario_obj, "steps", []):
            if step.type != "given":
                continue
            # Direct match: "Connect is configured in vip.toml"
            for prefix, product_key in _GIVEN_PRODUCT_STEPS.items():
                if step.name.startswith(prefix):
                    pc = cfg.product_config(product_key)
                    if not pc.is_configured:
                        return True
            # Parameterized match: "<product> is configured in vip.toml"
            # pytest-bdd stores Scenario Outline examples in callspec.params
            # as {'_pytest_bdd_example': {'product': 'Connect', ...}}.
            if step.name.startswith("<") and "is configured" in step.name:
                product_name = _get_bdd_param_product(item)
                if product_name:
                    resolved_key = _PRODUCT_DISPLAY_NAMES.get(product_name)
                    if resolved_key:
                        pc = cfg.product_config(resolved_key)
                        if not pc.is_configured:
                            return True

    return False


# Products that require username/password credentials.
_AUTH_PRODUCTS = {"connect", "workbench"}


def _requires_auth(item: pytest.Item) -> bool:
    """Return True if *item* requires authentication credentials."""
    # Explicit product markers.
    for marker_name in _AUTH_PRODUCTS:
        if item.get_closest_marker(marker_name) is not None:
            return True

    # BDD "Given" steps that reference an auth-required product.
    fn = getattr(item, "obj", None)
    scenario_obj = getattr(fn, "__scenario__", None) if fn else None
    if scenario_obj is not None:
        for step in getattr(scenario_obj, "steps", []):
            if step.type != "given":
                continue
            for prefix, product_key in _GIVEN_PRODUCT_STEPS.items():
                if product_key in _AUTH_PRODUCTS and step.name.startswith(prefix):
                    return True

    return False


def _is_api_auth_only(item: pytest.Item) -> bool:
    """Return True if *item* is marked as requiring only API-key authentication."""
    return item.get_closest_marker("api_auth") is not None


def require_connect_api_key(vip_cfg: VIPConfig) -> None:
    """Fail loudly if Connect is configured but no API key is available.

    Without this guard the ``connect_client`` fixture returns a client with an
    empty ``Authorization`` header, which silently succeeds for unauthenticated
    endpoints (``/server_settings``) but produces opaque 401/403 failures deep
    inside individual scenarios — each one rendered through pytest-bdd's
    ``call_fixture_func`` frame, hiding the real cause (auth setup failed).

    Calling this from the fixture turns every Connect API test's first error
    into a single, actionable root-cause message pointing at the missing key
    and the upstream mint diagnostics.

    Does nothing when Connect is not configured at all (no URL set) — those
    tests are deselected upstream.
    """
    if not vip_cfg.connect.is_configured:
        return
    if vip_cfg.connect.api_key:
        return
    pytest.fail(
        "Connect API key is not configured — API-based tests cannot run.\n"
        "  - Set VIP_CONNECT_API_KEY in the environment, or\n"
        "  - Set connect.api_key in vip.toml, or\n"
        "  - Use --headless-auth or --interactive-auth to mint one automatically.\n"
        "If you already used --headless-auth, scroll up to the 'Mint diagnostic' "
        "lines for why minting failed.",
        pytrace=False,
    )


def _maybe_skip_for_version(item: pytest.Item, cfg: VIPConfig) -> None:
    """Skip *item* when its ``min_version`` marker requirement is not met.

    When the deployed or required version cannot be parsed as a
    ``ProductVersion`` (including the "version unknown" case, ``pc.version
    is None``), the test is skipped and flagged as N/A-by-version rather
    than run optimistically or skipped indistinguishably from an ordinary
    skip. This surfaces version-detection gaps in the report instead of
    hiding them behind a possibly-spurious pass or a generic "skipped".
    """
    marker = item.get_closest_marker("min_version")
    if marker is None:
        return

    product = marker.kwargs.get("product") or (marker.args[0] if marker.args else None)
    version = marker.kwargs.get("version") or (marker.args[1] if len(marker.args) > 1 else None)
    if not product or not version:
        return

    try:
        pc = cfg.product_config(product)
    except ValueError:
        return

    if pc.version is None:
        _skip_version_unknown(item, product, version, reason="version unknown")
        return

    try:
        deployed = ProductVersion(pc.version)
    except ValueError:
        _skip_version_unknown(
            item, product, version, reason=f"deployed version {pc.version!r} is unparseable"
        )
        return

    try:
        required = ProductVersion(version)
    except ValueError:
        _skip_version_unknown(
            item, product, version, reason=f"required version {version!r} is unparseable"
        )
        return

    if deployed < required:
        item.add_marker(
            pytest.mark.skip(reason=f"{product} version {pc.version} < required {version}")
        )


def _skip_version_unknown(item: pytest.Item, product: str, version: str, *, reason: str) -> None:
    """Skip *item* and flag it as N/A-by-version, warning about the gap."""
    item.stash[_version_na_key] = True
    item.add_marker(
        pytest.mark.skip(
            reason=f"VIP: version unknown for {product} — cannot evaluate "
            f"min_version(product={product!r}, version={version!r}) ({reason})"
        )
    )
    warnings.warn(
        f"VIP: cannot evaluate min_version(product={product!r}, version={version!r}) "
        f"for {item.nodeid} — {reason}. Skipping and marking N/A instead of running "
        "optimistically.",
        stacklevel=1,
    )


# ---------------------------------------------------------------------------
# JSON results for Quarto report
# ---------------------------------------------------------------------------


_scenario_stash_key = pytest.StashKey[dict[str, str | None]]()


def _stash_scenario_metadata(item: pytest.Item) -> None:
    """Extract pytest-bdd scenario metadata and stash it on the item."""
    scenario_title = None
    feature_description = None

    # pytest-bdd stores scenario info as __scenario__ on the wrapper function.
    fn = getattr(item, "obj", None)
    scenario_obj = getattr(fn, "__scenario__", None) if fn else None
    if scenario_obj is not None:
        scenario_title = getattr(scenario_obj, "name", None)
        feature_obj = getattr(scenario_obj, "feature", None)
        if feature_obj is not None:
            feature_description = getattr(feature_obj, "description", None)

    item.stash[_scenario_stash_key] = {
        "scenario_title": scenario_title,
        "feature_description": feature_description,
    }


def _extract_exception_info(longrepr: str) -> tuple[str, str]:
    """Extract (exception_type, message) from a longrepr string.

    Handles four common formats:
    - pytest's ``E   ExcType: message`` lines in tracebacks
    - pytest's bare ``E   assert ...`` lines (assertion rewriting, no type prefix)
    - pytest's bare ``E   ExcType`` lines (exception with no message)
    - plain ``ExcType: message`` strings (e.g. from failures.json)

    Returns ``("UnknownError", <truncated string>)`` if parsing fails.
    """
    # Look for pytest's "E   ExcType: message" line format (message may be empty).
    # Multi-line assertion messages produce continuation "E   ..." lines that
    # we join together so the concise output keeps the full details.
    m = re.search(
        r"^E\s+([\w.]+(?:Error|Exception|Timeout|Refused)?):\s*(.*)",
        longrepr,
        re.MULTILINE,
    )
    if m:
        msg_lines = [m.group(2).strip()]
        # Gather only the contiguous block of E-lines that follow immediately.
        for line in longrepr[m.end() :].lstrip("\n").splitlines():
            cont = re.match(r"^E\s{3,}(.+)", line)
            if not cont:
                break
            msg_lines.append(cont.group(1).strip())
        return m.group(1), "\n".join(line for line in msg_lines if line)

    # Bare assertion from pytest's assertion rewriting: "E   assert 403 == 200"
    m = re.search(r"^E\s+(assert\s+.+)", longrepr, re.MULTILINE)
    if m:
        return "AssertionError", m.group(1).strip()

    # Bare exception type with no message: "E   ValueError" (no colon)
    m = re.search(
        r"^E\s+([\w.]+(?:Error|Exception|Timeout|Refused)?)\s*$",
        longrepr,
        re.MULTILINE,
    )
    if m:
        return m.group(1), ""

    # Fall back to "ExcType: message" at the start of the string.
    m = re.match(r"([\w.]+(?:Error|Exception|Timeout|Refused)?):\s*(.+)", longrepr.strip())
    if m:
        return m.group(1), m.group(2).strip()

    return "UnknownError", longrepr.strip()[:200]


def _format_concise_error(
    nodeid: str,
    exc_type: str,
    exc_message: str,
) -> str:
    """Format a concise one-liner error message for terminal and report display.

    AssertionError is treated as an expected test failure — the message is shown
    directly. All other exception types are prefixed with "an unexpected error
    occurred" to signal infrastructure or code issues.
    """
    test_name = nodeid.split("::")[-1] if "::" in nodeid else nodeid

    is_assertion = exc_type == "AssertionError" or exc_type.endswith(".AssertionError")

    if not exc_message:
        if is_assertion:
            return f"{test_name}: {exc_type}"
        return f"{test_name}: an unexpected error occurred: {exc_type}"

    if is_assertion:
        # Custom assertion messages are user-actionable — show them directly.
        # Bare assertions (e.g. "assert 403 == 200") still need the type prefix.
        if exc_message.lstrip().startswith("assert "):
            return f"{test_name}: AssertionError: {exc_message}"
        return f"{test_name}: {exc_message}"

    return f"{test_name}: an unexpected error occurred: {exc_type}: {exc_message}"


# ---------------------------------------------------------------------------
# Long-running test heartbeat
# ---------------------------------------------------------------------------

_HEARTBEAT_INTERVAL = 30  # seconds between "still running" messages


class _Heartbeat:
    """Print periodic elapsed-time messages while a test is running."""

    def __init__(self, writer, interval: int = _HEARTBEAT_INTERVAL):
        self._writer = writer
        self._interval = interval
        self._start: float = 0
        self._stop_event = threading.Event()
        self._thread: threading.Thread | None = None

    def start(self) -> None:
        self._start = time.monotonic()
        self._stop_event.clear()
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def stop(self) -> None:
        # Join without a timeout: callers (notably before locust/gevent import)
        # rely on no heartbeat thread being alive, and _run exits promptly once
        # the stop event is set.
        self._stop_event.set()
        if self._thread is not None:
            self._thread.join()
            self._thread = None

    def _run(self) -> None:
        while not self._stop_event.wait(self._interval):
            elapsed = int(time.monotonic() - self._start)
            self._writer(f"  ... still running ({elapsed}s)")


# Exposed so that code importing gevent/locust (which calls monkey.patch_all)
# can stop the heartbeat thread first to avoid a deadlock.
_current_heartbeat: _Heartbeat | None = None


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item: pytest.Item, nextitem) -> Generator[None, None, None]:  # noqa: ARG001
    """Print periodic heartbeat messages while a test is running."""
    global _current_heartbeat
    heartbeat: _Heartbeat | None = None
    if _active_config is not None:
        tr = _active_config.pluginmanager.get_plugin("terminalreporter")
        if tr is not None:
            heartbeat = _Heartbeat(tr.write_line)
            heartbeat.start()
            _current_heartbeat = heartbeat
    try:
        yield
    finally:
        if heartbeat is not None:
            heartbeat.stop()
            _current_heartbeat = None


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call):  # noqa: ARG001
    """Attach VIP metadata to the report so it survives xdist serialization.

    Under xdist, ``TestReport.__dict__`` is serialized and sent from worker
    to controller.  Custom attributes set here travel with the report, making
    them available in ``pytest_runtest_logreport`` on the controller where
    result collection happens.
    """
    outcome = yield
    report: pytest.TestReport = outcome.get_result()
    if _active_config is None:
        return
    if report.when == "call" or (report.when == "setup" and report.skipped):
        markers: list[str] = []
        try:
            markers = [m.name for m in item.iter_markers()]
        except Exception:
            pass

        item_stash = getattr(item, "stash", None)
        scenario_meta: dict[str, str | None] = {}
        na_version = False
        if item_stash is not None:
            scenario_meta = item_stash.get(_scenario_stash_key, {})
            na_version = item_stash.get(_version_na_key, False)

        # These attributes survive xdist worker→controller serialization.
        report.vip_markers = markers  # type: ignore[attr-defined]
        report.vip_scenario_title = scenario_meta.get("scenario_title")  # type: ignore[attr-defined]
        report.vip_feature_description = scenario_meta.get("feature_description")  # type: ignore[attr-defined]
        report.vip_na_version = na_version  # type: ignore[attr-defined]


@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_logstart(
    nodeid: str,  # noqa: ARG001
    location: tuple[str, int | None, str],  # noqa: ARG001
) -> Generator[None, None, None]:
    """Suppress the built-in terminal reporter's pre-test location line under xdist.

    Under real xdist parallelism, the controller relays each worker's
    ``pytest_runtest_logstart`` event (via ``dsession.worker_logstart``) the
    moment that worker *starts* a test, independent of when any other
    worker's test *finishes*. pytest's built-in ``TerminalReporter`` assumes
    a single serial stream: at ``-v`` it writes the location at logstart and
    later appends the outcome to that *same* line at logreport (matched via
    a shared ``currentfspath`` attribute). With several workers running
    concurrently, multiple tests' logstart events land before any of their
    logreport events, so ``currentfspath`` gets overwritten by other tests'
    locations before the original one's outcome arrives -- producing a
    location-only line up front and a disconnected result line later.

    We already delete ``report.node`` below so ``pytest_runtest_logreport``
    takes its non-xdist branch and writes location + outcome together, in
    one line, in a single call (see that hook's docstring). That only
    produces one line if nothing has already claimed the location line
    first. So here, only when the xdist controller is actually distributing
    (``dsession`` registered, and this isn't a worker), we suppress the
    logstart write for the duration of the wrapped call.

    The built-in reporter gates that write on ``showlongtestinfo``, which
    reads the fine-grained ``verbosity_test_cases`` *ini* value rather than
    the plain ``-v`` count whenever that ini value isn't ``"auto"`` -- and
    ``pytest_configure`` above already pins it (via ``config._inicache``) to
    force skip reasons to print in full. So bumping ``config.option.verbose``
    here would have no effect; we instead override that same ini-cache entry
    to ``"0"``, restoring it immediately after so the later
    ``pytest_runtest_logreport`` call (which reads the same cached value to
    decide whether to show the outcome line at all) is unaffected. Nothing
    else runs in between, since hookwrapper dispatch is synchronous.
    """
    config = _active_config
    if (
        config is None
        or hasattr(config, "workerinput")
        or not config.pluginmanager.hasplugin("dsession")
    ):
        yield
        return

    original = config._inicache.get("verbosity_test_cases")
    config._inicache["verbosity_test_cases"] = "0"
    try:
        yield
    finally:
        if original is None:
            config._inicache.pop("verbosity_test_cases", None)
        else:
            config._inicache["verbosity_test_cases"] = original


@pytest.hookimpl(tryfirst=True)
def pytest_runtest_logreport(report: pytest.TestReport) -> None:
    """Collect results for JSON report and apply concise terminal display.

    Under xdist, ``pytest_runtest_makereport`` runs on workers but the
    controller never sees the worker's stash.  This hook fires on the
    controller for every forwarded result (via ``dsession.worker_testreport``),
    making it the right place to aggregate results.  The ``workerinput``
    guard skips processing on workers so results are collected exactly once.
    """
    # Strip the xdist worker attribute so pytest's built-in TerminalReporter
    # does not prefix each verbose line with ``[gw<N>]`` and instead writes
    # the location and outcome together in one line (see
    # ``pytest_runtest_logstart`` above for why that requires suppressing
    # the pre-test location line too).  Runs before the terminal reporter
    # thanks to ``tryfirst=True``.
    if hasattr(report, "node"):
        del report.node

    if _active_config is None:
        return

    # On xdist workers, skip all processing — the controller handles it.
    # _active_config is a pytest.Config; xdist sets workerinput on worker configs.
    if hasattr(_active_config, "workerinput"):
        return

    # Capture this line's color for the progress-indicator recolor wrapper.
    # tryfirst ensures this runs before the terminal reporter renders the same
    # report, so the trailing ``[ x%]`` picks up this line's own outcome.
    global _current_line_color
    _current_line_color = _outcome_color(report)

    # --- Result collection (for JSON report) ---
    if report.when == "call" or (report.when == "setup" and report.skipped):
        results = _active_config.stash.get(_results_key, None)
        if results is not None:
            longrepr_str = str(report.longrepr) if report.longrepr else None
            concise_error = None
            if report.outcome == "failed" and longrepr_str:
                exc_type, exc_message = _extract_exception_info(longrepr_str)
                concise_error = _format_concise_error(report.nodeid, exc_type, exc_message)

            results.append(
                {
                    "nodeid": report.nodeid,
                    "outcome": report.outcome,
                    "duration": report.duration,
                    "longrepr": longrepr_str,
                    "concise_error": concise_error,
                    "markers": list(getattr(report, "vip_markers", ())),
                    "scenario_title": getattr(report, "vip_scenario_title", None),
                    "feature_description": getattr(report, "vip_feature_description", None),
                    "na_version": getattr(report, "vip_na_version", False),
                }
            )

    # --- Concise terminal display ---
    if _active_config.getoption("--vip-verbose", default=False):
        return
    if report.outcome not in ("failed", "error"):
        return
    if not report.longrepr:
        return

    longrepr_str = str(report.longrepr)
    exc_type, exc_message = _extract_exception_info(longrepr_str)
    report.longrepr = _format_concise_error(report.nodeid, exc_type, exc_message.replace("\n", " "))


def pytest_sessionfinish(session: pytest.Session, exitstatus: int) -> None:
    # xdist workers skip all session-end cleanup (controller handles it).
    is_worker = hasattr(session.config, "workerinput")

    if not is_worker:
        # Clean up interactive auth session (delete API key, remove temp files)
        auth_session = session.config.stash.get(_auth_session_key, None)
        if auth_session is not None:
            auth_session.cleanup()

    report_path = session.config.getoption("--vip-report")
    if not report_path or is_worker:
        return

    cfg: VIPConfig = session.config.stash[_vip_config_key]
    results = session.config.stash.get(_results_key, [])

    # Include product metadata for the report.
    # Derive product list from _PRODUCT_MARKERS so new products are picked up automatically.
    products: dict[str, dict[str, Any]] = {}
    for name in _PRODUCT_MARKERS.values():
        pc = cfg.product_config(name)
        products[name] = {
            "enabled": pc.enabled,
            "url": pc.url,
            "version": pc.version,
            "configured": pc.is_configured,
        }

    payload = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "deployment_name": cfg.deployment_name,
        "exit_status": exitstatus,
        "products": products,
        "results": results,
    }

    try:
        p = Path(report_path)
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps(payload, indent=2))
    except OSError as exc:
        warnings.warn(f"VIP: could not write report to {report_path}: {exc}", stacklevel=1)
        return

    # Write failures.json alongside results.json so report rendering is idempotent.
    failures = [r for r in results if r.get("outcome") == "failed"]
    if failures:
        failures_payload = {
            "deployment": cfg.deployment_name,
            "generated_at": payload["generated_at"],
            "failures": [
                {
                    "test": r["nodeid"],
                    "scenario": r.get("scenario_title"),
                    "feature": r.get("feature_description"),
                    "error_summary": r.get("concise_error") or (r.get("longrepr") or "")[:500],
                }
                for r in failures
            ],
        }
        failures_path = p.parent / "failures.json"
        try:
            failures_path.write_text(json.dumps(failures_payload, indent=2) + "\n")
        except OSError as exc:
            warnings.warn(
                f"VIP: could not write failures report to {failures_path}: {exc}", stacklevel=1
            )