conftest.py

# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

"""Block dynamo.vllm/sglang/triton from shadowing the installed packages.

Pytest collection puts components/src/dynamo on sys.path, which makes
`import vllm` resolve to dynamo.vllm. Spawned subprocesses (EngineCore,
sglang scheduler) inherit that and crash on `from vllm.v1 ...`. The same
applies to dynamo.triton vs. the triton compiler package torch imports.
"""

from __future__ import annotations

import importlib
import os
import sys
from pathlib import Path

import pytest

from tests.marker_categories import REQUIRED_CATEGORIES

_NO_DEFAULT_MARKERS_ENV = "DYNAMO_PYTEST_NO_DEFAULT_MARKERS"
_SUITE_MARKERS = REQUIRED_CATEGORIES["Lifecycle"]
_MACHINE_MARKERS = REQUIRED_CATEGORIES["Hardware"]

# Trees with dedicated CI jobs or runnable demos outside the runtime suites.
# Helm chart tests run in helm-chart-tests, which installs the Helm CLI.
# Their files are still collected, so a developer can run one explicitly, but
# they must not be defaulted into a lifecycle: an unmarked demo helper such as
# examples/backends/sglang/test_sglang_profile.py assumes a server the harness
# never starts, and defaulting it into pre_merge makes the CPU job run it.
# Anything under these roots that carries its own markers is unaffected.
# ``skills`` is a symlink to ``.agents/skills``, so resolve() lands on
# ``.agents``; both names are listed because either can reach collection.
_UNMANAGED_PATHS = (("examples",), ("skills",), (".agents",), ("deploy", "helm"))
_REPO_ROOT = Path(__file__).resolve().parent

# Seed sys.modules with the venv copies before pytest collection runs.
# Best-effort: an engine that is present but not importable (an editable install
# with an unreadable source tree raises PermissionError) must not abort collection.
for _name in ("vllm", "sglang", "triton"):
    try:
        importlib.import_module(_name)
    # ImportError is a missing engine; OSError is one whose sources are present
    # but unreadable. A failure of any other kind comes from inside an engine
    # that did start importing, and swallowing it here would run the suite
    # against a half-initialized engine, so it propagates.
    except (ImportError, OSError) as _exc:
        if isinstance(_exc, ModuleNotFoundError) and _exc.name == _name:
            continue  # the engine is simply not installed; nothing to report
        # Anything else means the install is broken, and one line here names
        # the cause rather than leaving a bare collection error downstream.
        print(
            f"conftest: {_name} is installed but could not be imported, so "
            f"tests that import it will fail: {type(_exc).__name__}: {_exc}",
            file=sys.stderr,
        )

# Suppress ImportPathMismatchError when pytest later loads dynamo.vllm
# under the bare name "vllm".
os.environ.setdefault("PY_IGNORE_IMPORTMISMATCH", "1")

_BAD_DYNAMO_PATH = str(
    Path(__file__).resolve().parent / "components" / "src" / "dynamo"
)


def _strip_bad_path() -> None:
    while _BAD_DYNAMO_PATH in sys.path:
        sys.path.remove(_BAD_DYNAMO_PATH)


# Strip the bad path before multiprocessing.spawn freezes sys.path for the
# child — catches re-insertions that happen during fixture/test execution.
try:
    import multiprocessing.spawn as _mps

    _orig_get_preparation_data = _mps.get_preparation_data

    def _patched_get_preparation_data(name):
        _strip_bad_path()
        return _orig_get_preparation_data(name)

    _mps.get_preparation_data = _patched_get_preparation_data
except Exception:
    pass


def pytest_runtest_setup(item):
    _strip_bad_path()


@pytest.hookimpl(wrapper=True, trylast=True)
def pytest_runtest_protocol(item, nextitem):
    """Tear down the parent's setup stack after a forked item.

    pytest-forked runs setup and teardown only in the child process
    (https://github.com/pytest-dev/pytest-forked, ``forked_run_report``), so
    the parent never calls ``teardown_exact(nextitem)`` for a forked item.
    Collectors the parent set up for earlier non-forked items, such as the
    ``Package`` of a test directory whose last tests are forked, then stay on
    the stack, and the first item of the next package fails setup with
    "previous item was not torn down properly". Doing the teardown pytest
    would have done keeps the parent's stack in step with ``nextitem``.

    A parent-side finalizer failure is reported as a teardown error for the
    item, as pytest's own teardown phase would, rather than escaping the hook
    as an INTERNALERROR. The child already reported a passing teardown, so
    nothing more is reported when the parent teardown succeeds.
    """
    result = yield
    if item.config.pluginmanager.has_plugin("pytest_forked") and (
        item.config.getoption("forked", default=False)
        or item.get_closest_marker("forked")
    ):
        # Match pytest's teardown when the session will not run the next item.
        if item.session.shouldfail or item.session.shouldstop:
            nextitem = None
        reraise: tuple[type[BaseException], ...] = (pytest.exit.Exception,)
        if not item.config.getoption("usepdb", False):
            reraise += (KeyboardInterrupt,)
        call = pytest.CallInfo.from_call(
            lambda: item.session._setupstate.teardown_exact(nextitem),
            when="teardown",
            reraise=reraise,
        )
        if call.excinfo is not None:
            # Built directly: makereport hook wrappers such as
            # pytest-rerunfailures expect per-item state that their protocol
            # hook sets, and pytest-forked's protocol hook skips it.
            report = pytest.TestReport.from_item_and_call(item, call)
            item.ihook.pytest_runtest_logreport(report=report)
            if not call.excinfo.errisinstance(pytest.skip.Exception):
                item.ihook.pytest_exception_interact(
                    node=item, call=call, report=report
                )
    return result


def _is_unmanaged(item) -> bool:
    """True when runtime CI must not supply lifecycle or hardware defaults."""
    try:
        path = Path(str(item.path)).resolve()
    except (AttributeError, OSError, ValueError):
        return False
    try:
        parts = path.relative_to(_REPO_ROOT).parts
    except ValueError:
        return False
    return any(parts[: len(prefix)] == prefix for prefix in _UNMANAGED_PATHS)


def _tag_xfail_for_datadog(item) -> None:
    """Expose ``@pytest.mark.xfail`` to Datadog Test Optimization.

    Datadog's test status is three-valued — pass / fail / skip. pytest reports
    an xfailed test as ``skipped`` and an xpassed one as ``passed``, and
    ddtrace forwards that raw outcome without looking at ``wasxfail``, so both
    are indistinguishable from an ordinary skip or pass once uploaded. Allure
    reads ``wasxfail`` and keeps first-class ``xfailed`` / ``xpassed``, which is
    why the OpenSearch-backed reports can show an xfail rate and Datadog cannot.

    The outcome is not knowable at collection time, but the *marker* is, and
    that is enough: combined with the status Datadog already records,

        @test.is_xfail:true AND @test.status:skip  -> xfailed
        @test.is_xfail:true AND @test.status:pass  -> xpassed

    ``dd_tags`` is ddtrace's supported per-test tag marker. It is read during
    test discovery (``_get_test_custom_tags`` in ddtrace's pytest plugin), so
    it has to be applied at collection — tagging from a report hook is too
    late.

    No-ops when ddtrace is absent: ``dd_tags`` is only registered by its
    plugin, so applying it without would fail collection as an unknown marker.
    """
    marker = item.get_closest_marker("xfail")
    if marker is None or not item.config.pluginmanager.hasplugin("ddtrace"):
        return
    # Conditions are positional args or ``condition=``; no condition means
    # unconditional. When every condition is falsy pytest runs the test as a
    # normal one, so tagging it would make a plain pass look like an xpass.
    conditions = marker.args or (
        (marker.kwargs["condition"],) if "condition" in marker.kwargs else ()
    )
    if conditions and not any(conditions):
        return
    item.add_marker(
        pytest.mark.dd_tags(
            **{
                "test.is_xfail": "true",
                # A strict xfail that passes is reported as a failure, so the
                # two cannot be collapsed downstream.
                "test.xfail_strict": str(
                    bool(
                        marker.kwargs.get("strict", item.config.getini("xfail_strict"))
                    )
                ).lower(),
            }
        )
    )


def pytest_itemcollected(item):
    """Apply CI defaults to tests missing lifecycle or hardware markers.

    This hook lives in the repository-root conftest so it applies to every
    collected test tree, including ``tests/``, ``components/src``, and
    ``aisimulate/tests``. It runs before pytest's marker filter.

    Anything defaulted also gets ``defaulted``, which routes it to the CPU
    parallel job. Without that marker every defaulted test matches
    ``not parallel`` and lands in the sequential job, which already carries the
    fault-tolerance suite in a single process -- the imports alone cost ~90 MB
    resident for the rest of the run, and the job was OOM-killed (exit 137).

    Trees in ``_UNMANAGED_PATHS`` are skipped entirely; they keep the
    pre-default behaviour of being deselected unless they mark themselves.

    ``DYNAMO_PYTEST_NO_DEFAULT_MARKERS=1`` disables the defaults so the marker
    report can inspect authored markers only.
    """
    _tag_xfail_for_datadog(item)
    if os.environ.get(_NO_DEFAULT_MARKERS_ENV) == "1":
        return
    if _is_unmanaged(item):
        return
    defaulted = False
    if not any(item.get_closest_marker(marker) for marker in _SUITE_MARKERS):
        item.add_marker(pytest.mark.pre_merge)
        defaulted = True
    if not any(item.get_closest_marker(marker) for marker in _MACHINE_MARKERS):
        item.add_marker(pytest.mark.gpu_0)
        defaulted = True
    if defaulted:
        item.add_marker(pytest.mark.defaulted)