diff --git a/.buildkite/common/ci_source_file_dependencies.yml b/.buildkite/common/ci_source_file_dependencies.yml index f6784c440..e4a54670a 100644 --- a/.buildkite/common/ci_source_file_dependencies.yml +++ b/.buildkite/common/ci_source_file_dependencies.yml @@ -236,6 +236,14 @@ source_file_dependencies: - *minicpmo_4_5 - tests/e2e/accuracy/minicpmo_4_5/ + omni_duplex_eval_ci: + - *minicpmo_4_5 + - *minicpmo_4_5_duplex + - tests/e2e/accuracy/omni_duplex_eval/** + - tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py + - vllm_omni/benchmarks/duplex/** + - vllm_omni/entrypoints/cli/benchmark/omni_duplex_eval.py + omni_minicpmo_4_5_perf: - *minicpmo_4_5 - tests/dfx/perf/scripts/run_benchmark.py diff --git a/.buildkite/cuda/test-nightly.yml b/.buildkite/cuda/test-nightly.yml index c69dcb146..fb9977d4d 100644 --- a/.buildkite/cuda/test-nightly.yml +++ b/.buildkite/cuda/test-nightly.yml @@ -70,6 +70,22 @@ steps: - export SEED_TTS_EVAL_DEVICE=cuda:1 - pytest -s -v tests/e2e/accuracy/minicpmo_4_5/test_minicpmo_4_5.py -m "full_model and H100 and B200 and cuda and (cards_1 or cards_2)" --run-level full_model + - label: ":full_moon: Omni · Omni-DuplexEval CI Guard" + source_file_dependencies: omni_duplex_eval_ci + timeout_in_minutes: 240 + env: + VLLM_DUPLEX_EVAL_JUDGE_URL: "http://127.0.0.1:8001" + commands: + - | + set +e + pytest -s -v tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci.py \ + -m "full_model and H100 and B200 and cuda and cards_2" \ + --run-level full_model \ + --basetemp /tmp/omni_duplex_ci + EXIT=$$? + buildkite-agent artifact upload "tests/e2e/accuracy/omni_duplex_eval/results/*.json" + exit $$EXIT + - label: ":full_moon: Omni · Perf Test · No Async Chunk" source_file_dependencies: omni_qwen3_omni_perf key: nightly-omni-performance-no-async-chunk diff --git a/.buildkite/npu/test-npu-nightly.yml b/.buildkite/npu/test-npu-nightly.yml index 601a7bf4b..5e242be72 100644 --- a/.buildkite/npu/test-npu-nightly.yml +++ b/.buildkite/npu/test-npu-nightly.yml @@ -85,6 +85,25 @@ steps: EXIT=$$? buildkite-agent artifact upload "tests/e2e/accuracy/minicpmo_4_5/results/*.json" exit $$EXIT + - label: ":full_moon: Omni · Omni-DuplexEval CI Guard" + source_file_dependencies: omni_duplex_eval_ci + timeout_in_minutes: 240 + mirror_hardwares: a3_npu_2 + env: + VLLM_WORKER_MULTIPROC_METHOD: spawn + VLLM_ALLOW_LONG_MAX_MODEL_LEN: "1" + HF_TOKEN: "${HF_TOKEN}" + VLLM_DUPLEX_EVAL_JUDGE_URL: "http://127.0.0.1:8001" + commands: + - | + set +e + pytest -s -v tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci.py \ + -m "full_model and npu and A3 and cards_2" \ + --run-level full_model \ + --basetemp /tmp/omni_duplex_ci + EXIT=$$? + buildkite-agent artifact upload "tests/e2e/accuracy/omni_duplex_eval/results/*.json" + exit $$EXIT - label: ":full_moon: Omni · MiniCPM-o 4.5 · Perf Test" source_file_dependencies: omni_minicpmo_4_5_perf key: nightly-omni-performance-minicpmo-4-5-npu diff --git a/.gitignore b/.gitignore index 542bc5035..8a0792829 100644 --- a/.gitignore +++ b/.gitignore @@ -245,6 +245,8 @@ datasets/ !recipes/cosmos3/negative_prompt.json !tests/dfx/perf/tests/*.json !apps/ComfyUI-vLLM-Omni/example_workflows/*.json +# Omni-DuplexEval CI config JSON (e.g. accuracy/omni_duplex_eval) +!tests/e2e/accuracy/omni_duplex_eval/*.json *.jsonl *.parquet diff --git a/tests/e2e/accuracy/omni_duplex_eval/README.md b/tests/e2e/accuracy/omni_duplex_eval/README.md new file mode 100644 index 000000000..d37e9ee02 --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/README.md @@ -0,0 +1,135 @@ +# Omni-DuplexEval CI Guard + +Nightly end-to-end scoring regression for [Hothan/Omni-DuplexEval] +(RTD 6 subtasks + PR 3 subtasks), driven by `vllm bench omni-duplex-eval` +(generate → evaluate → summarize). + +## Files + +| File | Purpose | +| ------ | --------- | +| `test_omni_duplex_eval_ci.py` | Main pytest: hardcoded thresholds + 6 assertions | +| `test_omni_duplex_eval_ci_mock.py` | CPU mock integration tests (no GPU / model needed) | +| `omni_duplex_eval_ci_config.json` | Dataset / revision / sample IDs / judge | +| `conftest.py` | `judge_server` fixture (endpoint resolution only) | +| `README.md` | This document | +| `results/.gitignore` | Track empty results dir for Buildkite artifact upload | + +## Run locally + +Judge server must be pre-started; this test never launches it: + +```bash +# CUDA (H100 / B200), or NPU (A3) +VLLM_DUPLEX_EVAL_JUDGE_URL=http://127.0.0.1:8001 \ + pytest -s -v tests/e2e/accuracy/omni_duplex_eval/ \ + -m "full_model and cuda and (H100 or B200) and cards_1" +``` + +The `omni_server` fixture boots MiniCPM-o-4_5 with the community duplex +params (`tests/e2e/online_serving/helpers/minicpmo_4_5_duplex.py`). +`judge.base_url_env` (default `VLLM_DUPLEX_EVAL_JUDGE_URL`) overrides +`judge.base_url` from the JSON config. + +## Results output + +After Phase 3 (summarize), the full `summarize_scores()` dict is persisted +to `results/summary.json` (same pattern as PR#6817 perf tests) and printed +to stdout: + +- `results/` is tracked in git via `results/.gitignore` (empty dir, nothing + committed inside). +- On Buildkite nightly, the step uploads `results/*.json` as artifacts + (`buildkite-agent artifact upload`) even when assertions fail, so a + failing run's scores remain downloadable. + +## Judge server + +Judge is a pre-started, OpenAI-compatible endpoint (never launched by this +test). It can run on a separate machine / GPU from the `omni_server`. + +Current judge: **Qwen2.5-VL-7B-Instruct** (`judge.model` in the JSON +config). The 7B model is chosen for Phase A to run on a single GPU (≈16 GB +FP16, or ≈5 GB with INT4 quantization); PR 0/1 classification and temporal +0–3 scoring are robust at this size. The 0.00–3.00 content scoring is the +weakest link (fine-grained counting / color / spatial checks) — if the +guard proves insensitive to regressions, upgrade the judge (e.g. +72B on FP8 H100 or an external API) and **re-run the baseline** to reset +`_MIN_*` thresholds. Changing the judge never affects the P0 assertions +(`protocol_pin`, sample count, `by_task` keys) or the CPU mock tests. + +Example (any OpenAI-compatible vLLM server): + +```bash +python -m vllm.entrypoints.openai.api_server \ + --model Qwen/Qwen2.5-VL-7B-Instruct --port 8001 --max-model-len 65536 \ + --allowed-local-media-path /tmp/omni_duplex_ci +``` + +If the judge is started on a different host or the media temporary directory +changes, adjust `--allowed-local-media-path` to match the path used by +`vllm bench omni-duplex-eval evaluate` for materialized videos. + +## CPU mock tests (no GPU / model / judge needed) + +`test_omni_duplex_eval_ci_mock.py` drives the *real* +`test_omni_duplex_eval_ci()` guard on CPU by stubbing the heavy I/O: + +- `_run_cli` is replaced by an argv-recording spy (never executes). +- `resolve_ref_audio` is stubbed to a local path. +- Fake score files are pre-written under `tmp_path/scores//` from + `sample_ids_per_split` (minus `exclude_ids`), so `summarize_scores()` + and the Phase-4 assertions run for real. + +It validates the CLI argument assembly (generate + evaluate), the +score-file-count check, `protocol_pin`, the RTD/PR `by_task` keys, and +all threshold comparisons — including the failure paths. Runs in CI on +every PR (`core_model` + `cpu` marks): + +```bash +pytest -v tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci_mock.py +``` + +## Thresholds (record mode → tighten after first nightly run) + +`sample_ids_per_split` contains real dataset IDs (verified against the +Hothan/Omni-DuplexEval splits; PR_correction / PR_event_reminder IDs are +inferred from the split row order and must be re-verified once the dataset +is reachable). `exclude_ids` has been removed. + +All `_MIN_*` thresholds are currently **0.0 (record mode)**: the nightly run +produces real scores and persists `results/summary.json` without failing the +gate. Tighten after the first nightly run: + +1. Record `summarize_scores()` output (mean content / temporal / success). +2. Set `_MIN_*` thresholds = `mean - n * std` (start `n=2`) in + `test_omni_duplex_eval_ci.py`. As a starting point, the paper's + MiniCPM-o 4.5 scores (100-point scale) map to ≈1.15 content / ≈2.40 + temporal / ≈0.20 PR on the code scales (see "Reference paper" below); + ~50% of those is a reasonable lower bound. +3. If needed, exclude known bad samples (generate hang / judge failure) via + `exclude_ids`. + +Thresholds are bound to the judge model: **switching the judge requires +re-running steps 1–2**. If the 7B judge shows high variance (small models +are more prompt-sensitive), widen `n` (e.g. `n=3`) or increase the sample +size before tightening. + +## Assertions (in order) + +1. `protocol_pin == PROTOCOL_PIN` (track upstream protocol changes) +2. `summary["samples"]` == number of selected IDs +3. RTD `by_task` keys == 6 RTD subtasks +4. PR `by_task` keys == 3 PR subtasks +5. RTD mean content / temporal scores >= thresholds +6. PR `mean_all_success` >= threshold + +## Reference paper & scoring scale + +Reference paper: **Omni-DuplexEval** (arXiv:2605.17360). The paper reports +a 100-point scale, while the evaluation code +(`vllm_omni/benchmarks/duplex/omni_duplex_eval_metrics.py`) uses a 3-point +content scale (0.00–3.00), a 0–3 integer temporal scale, and a 0/1 PR +success flag. The CI thresholds are defined against the **code's** scales +(the source of truth for `summarize_scores()` output), not the paper's +100-point scale. diff --git a/tests/e2e/accuracy/omni_duplex_eval/__init__.py b/tests/e2e/accuracy/omni_duplex_eval/__init__.py new file mode 100644 index 000000000..0ef6415ba --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/__init__.py @@ -0,0 +1,3 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""CI guard for Omni-DuplexEval end-to-end scoring regression.""" diff --git a/tests/e2e/accuracy/omni_duplex_eval/conftest.py b/tests/e2e/accuracy/omni_duplex_eval/conftest.py new file mode 100644 index 000000000..27e5a5444 --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/conftest.py @@ -0,0 +1,156 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Module-scoped judge server fixture for omni-duplex-eval CI guard.""" + +from __future__ import annotations + +import json +import os +import subprocess +import time +from collections.abc import Generator +from pathlib import Path + +import pytest +import requests +import torch + +_CONFIG_PATH = Path(__file__).parent / "omni_duplex_eval_ci_config.json" + + +def _load_judge_model() -> str: + """Read judge model from CI config, fall back to default.""" + try: + config = json.loads(_CONFIG_PATH.read_text()) + return config.get("judge", {}).get("model", "Qwen/Qwen2.5-VL-7B-Instruct") + except Exception: + return "Qwen/Qwen2.5-VL-7B-Instruct" + + +def _parse_visible_devices() -> tuple[str, list[str]]: + """Parse CUDA or Ascend NPU visible devices from environment.""" + if torch.cuda.is_available(): + env_key = "CUDA_VISIBLE_DEVICES" + elif hasattr(torch, "npu") and torch.npu.is_available(): + env_key = "ASCEND_RT_VISIBLE_DEVICES" + else: + return "", [] + + raw = os.environ.get(env_key, "").strip() + if not raw: + # No env restriction — physical device IDs are 0, 1, 2... + count = torch.accelerator.device_count() if hasattr(torch, "accelerator") else 0 + return env_key, [str(i) for i in range(count)] + + devices = [d.strip() for d in raw.split(",") if d.strip()] + return env_key, devices + + +def _omni_server_device_env() -> dict[str, str] | None: + """Env overrides pinning the Omni server to device[1] (split from judge). + + Returns ``None`` when fewer than two devices are allocated (nothing to + split, e.g. on CPU-only hosts where ``_parse_visible_devices`` yields an + empty list). Pure function: never mutates the pytest process environment. + """ + env_key, devices = _parse_visible_devices() + if len(devices) >= 2: + return {env_key: devices[1]} + return None + + +@pytest.fixture(scope="module") +def judge_server( + tmp_path_factory: pytest.TempPathFactory, +) -> Generator[str, None, None]: + """Launch the Qwen2.5-VL-7B judge server, wait for health, yield base URL. + + If ``VLLM_DUPLEX_EVAL_JUDGE_URL`` is already set *and* the endpoint + responds to ``/health``, reuse it (developer convenience). + + Otherwise, start the judge as a subprocess: + - The judge is pinned to the first device of the job's allocation + (``devices[0]``), so it never selects a device outside the job's set. + - The omni server (managed by the ``omni_server`` fixture) is pinned to + device[1] via ``OmniServerParams.env_dict`` (see + ``_omni_server_device_env``), leaving the pytest process environment + untouched. + + The fixture terminates the judge on teardown. + """ + port = int(os.environ.get("VLLM_DUPLEX_EVAL_JUDGE_PORT", "8001")) + base_url = f"http://127.0.0.1:{port}" + + # Reuse a running judge if URL is pre-set and healthy. + preset_url = os.environ.get("VLLM_DUPLEX_EVAL_JUDGE_URL") + if preset_url: + try: + resp = requests.get(f"{preset_url}/health", timeout=5) + if resp.status_code == 200: + yield preset_url + return + except Exception: + pass + + # Determine judge device from the job's device allocation. + _, devices = _parse_visible_devices() + judge_dev = devices[0] if devices else "0" + + # Build device-isolated environment. + env = os.environ.copy() + if torch.cuda.is_available(): + env["CUDA_VISIBLE_DEVICES"] = judge_dev + env["ASCEND_RT_VISIBLE_DEVICES"] = "" # clear NPU visibility + elif hasattr(torch, "npu") and torch.npu.is_available(): + env["ASCEND_RT_VISIBLE_DEVICES"] = judge_dev + env["CUDA_VISIBLE_DEVICES"] = "" # clear CUDA visibility + + # Derive media allowlist from pytest's actual temp root. + media_path = str(tmp_path_factory.getbasetemp()) + os.makedirs(media_path, exist_ok=True) + + cmd = [ + "python", + "-m", + "vllm.entrypoints.openai.api_server", + "--model", + _load_judge_model(), + "--port", + str(port), + "--max-model-len", + "65536", + "--trust-remote-code", + "--dtype", + "bfloat16", + "--gpu-memory-utilization", + "0.85", + "--allowed-local-media-path", + media_path, + ] + + proc = subprocess.Popen(cmd, env=env) + + # Health-check loop: up to 60 iterations × 5 s = 5 min. + for _ in range(60): + try: + resp = requests.get(f"{base_url}/health", timeout=2) + if resp.status_code == 200: + break + except Exception: + pass + time.sleep(5) + else: + proc.kill() + proc.wait() + raise RuntimeError(f"Judge server failed to become healthy at {base_url} within 300 s") + + yield base_url + + # Teardown. + proc.terminate() + try: + proc.wait(timeout=30) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait() diff --git a/tests/e2e/accuracy/omni_duplex_eval/omni_duplex_eval_ci_config.json b/tests/e2e/accuracy/omni_duplex_eval/omni_duplex_eval_ci_config.json new file mode 100644 index 000000000..a3e5b6243 --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/omni_duplex_eval_ci_config.json @@ -0,0 +1,99 @@ +{ + "dataset": "Hothan/Omni-DuplexEval", + "dataset_revision": null, + "protocol_pin": "ca3c122b4d4bf67afd6b18ea5e724b4561bdde48", + "sample_ids_per_split": { + "RTD_world_knowledge": [ + "360", + "361", + "362", + "363", + "364" + ], + "RTD_counting": [ + "410", + "411", + "412", + "413", + "414" + ], + "RTD_fine_grained_movement": [ + "460", + "461", + "462", + "463", + "464" + ], + "RTD_interaction_relation": [ + "510", + "511", + "512", + "513", + "514" + ], + "RTD_OCR": [ + "560", + "561", + "562", + "563", + "564" + ], + "RTD_Omni": [ + "610", + "611", + "612", + "613", + "614" + ], + "PR_correction": [ + "0", + "1", + "2", + "3", + "4", + "5", + "6", + "7", + "8", + "9", + "10", + "11", + "12" + ], + "PR_event_reminder": [ + "133", + "134", + "135", + "136", + "137", + "138", + "139", + "140", + "141", + "142", + "143", + "144", + "145" + ], + "PR_post_event_reminder": [ + "261", + "262", + "263", + "264", + "265", + "266", + "267", + "268", + "269", + "270" + ] + }, + "judge": { + "model": "Qwen/Qwen2.5-VL-7B-Instruct", + "base_url": "http://127.0.0.1:8001", + "base_url_env": "VLLM_DUPLEX_EVAL_JUDGE_URL", + "video_mode": "video_url", + "fps": 2, + "eval_workers": 4 + } +} diff --git a/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci.py b/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci.py new file mode 100644 index 000000000..29f106025 --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci.py @@ -0,0 +1,217 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""CI guard for Omni-DuplexEval scoring regression. + +Pipeline: generate -> evaluate -> summarize -> assert. +""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import pytest + +from tests.e2e.accuracy.omni_duplex_eval import conftest as _omni_conftest +from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import ( + SERVER_PARAMS as DUPLEX_TEST_PARAMS, +) +from tests.e2e.online_serving.helpers.minicpmo_4_5_duplex import ( + realtime_url, + resolve_ref_audio, +) +from tests.helpers.mark import hardware_test +from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import summarize_scores +from vllm_omni.benchmarks.duplex.omni_duplex_eval_metrics import PROTOCOL_PIN + +# Thresholds (hardcoded, community accuracy test pattern). +# baseline commit: 873e9bff7c545c5cda79fdb93a867f09e443b61d +# +# RECORD MODE: all thresholds are 0.0 until the first nightly run produces +# real scores. After that, tighten to ~50% of the measured baseline (paper +# MiniCPM-o 4.5 on the 100-point scale maps to ~1.15 content / ~2.40 +# temporal / ~0.20 PR on the code scales; see README "Reference paper"). +_MIN_RTD_MEAN_CONTENT_SCORE = 0.0 # 3-point scale (0.00-3.00) +_MIN_RTD_MEAN_TEMPORAL_SCORE = 0.0 # 0-3 scale +_MIN_PR_MEAN_ALL_SUCCESS = 0.0 # 0-1 scale + +_CONFIG_PATH = Path(__file__).resolve().parent / "omni_duplex_eval_ci_config.json" + + +def _load_ci_config() -> dict: + return json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) + + +pytestmark = [pytest.mark.full_model, pytest.mark.omni] + +# Dynamically inject the sub-process device isolation env_dict to avoid mutating +# the pytest parent process's os.environ (which causes leaks to other multi-GPU +# tests in the same session). ``_omni_server_device_env()`` returns ``None`` on +# CPU-only hosts (no device split), which matches the default ``env_dict=None``. +_DUPLEX_SERVER_PARAMS = [ + pytest.param( + p.values[0]._replace(env_dict=_omni_conftest._omni_server_device_env()), + id=p.id, + ) + for p in DUPLEX_TEST_PARAMS +] + + +_RESULT_DIR = Path(__file__).resolve().parent / "results" + + +@hardware_test(res={"cuda": ["H100", "B200"], "npu": "A3"}, num_cards=2) +@pytest.mark.parametrize("omni_server", _DUPLEX_SERVER_PARAMS, indirect=True) +def test_omni_duplex_eval_ci(omni_server, tmp_path: Path, judge_server: str) -> None: + """CI guard: generate -> evaluate -> summarize -> assert scores.""" + config = _load_ci_config() + response_root = tmp_path / "responses" + score_root = tmp_path / "scores" + response_root.mkdir(parents=True, exist_ok=True) + score_root.mkdir(parents=True, exist_ok=True) + + all_ids = sorted(set(sid for sids in config["sample_ids_per_split"].values() for sid in sids)) + exclude = set(config.get("exclude_ids", [])) + sample_ids = [sid for sid in all_ids if sid not in exclude] + + # Validate dataset_revision (reserved for future pinning; not yet supported). + dataset_revision = config.get("dataset_revision") + if dataset_revision: + raise NotImplementedError(f"dataset_revision pinning is not yet supported (got {dataset_revision!r}).") + + # Phase 1: Generate + dataset_ref = config["dataset"] + _run_cli( + [ + "vllm", + "bench", + "omni-duplex-eval", + "--omni", + "generate", + "--dataset", + dataset_ref, + "--split", + "all", + "--ids", + *sample_ids, + "--url", + realtime_url(omni_server), + "--model", + omni_server.model, + "--ref-audio", + str(resolve_ref_audio()), + "--response-root", + str(response_root), + ] + ) + + # Phase 2: Evaluate + judge_cfg = config["judge"] + _run_cli( + [ + "vllm", + "bench", + "omni-duplex-eval", + "--omni", + "evaluate", + "--dataset", + dataset_ref, + "--split", + "all", + "--ids", + *sample_ids, + "--response-root", + str(response_root), + "--score-root", + str(score_root), + "--judge-model", + judge_cfg["model"], + "--judge-base-url", + judge_server, + "--judge-video-mode", + judge_cfg.get("video_mode", "video_url"), + "--judge-fps", + str(judge_cfg.get("fps", 2)), + "--eval-workers", + str(judge_cfg.get("eval_workers", 4)), + ] + ) + + # Sanity check: config pin must match the runtime source-of-truth constant. + config_pin = config.get("protocol_pin") + assert config_pin == PROTOCOL_PIN, ( + f"config protocol_pin {config_pin!r} does not match source of truth {PROTOCOL_PIN!r}" + ) + + score_files = [p for p in Path(score_root).rglob("*.json") if not p.name.endswith("_summary.json")] + assert len(score_files) == len(sample_ids), f"expected {len(sample_ids)} score files, found {len(score_files)}" + + # Validate each score file is consistent with the configured protocol pin. + mismatches = [] + for sf in score_files: + row = json.loads(sf.read_text(encoding="utf-8")) + row_pin = row.get("protocol_pin") + if row_pin != config_pin: + mismatches.append(f"{sf.name}: {row_pin!r} != {config_pin!r}") + assert not mismatches, "protocol_pin mismatch in score files:\n" + "\n".join(mismatches) + + # Phase 3: Summarize + persist result JSON (Buildkite artifact upload pattern) + summary = summarize_scores(score_root) + _RESULT_DIR.mkdir(parents=True, exist_ok=True) + result_path = _RESULT_DIR / "summary.json" + result_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") + print(f"\n[Omni-DuplexEval] summary saved to {result_path}") + print(json.dumps(summary, ensure_ascii=False, indent=2)) + + # Phase 4: Assertions + assert summary["protocol_pin"] == PROTOCOL_PIN, f"protocol_pin changed: {summary['protocol_pin']} != {PROTOCOL_PIN}" + assert summary["samples"] == len(sample_ids), f"expected {len(sample_ids)} samples, got {summary['samples']}" + + rtd = summary.get("rtd", {}) + pr = summary.get("pr", {}) + assert rtd.get("by_task", {}).keys() >= { + "RTD_world_knowledge", + "RTD_counting", + "RTD_fine_grained_movement", + "RTD_interaction_relation", + "RTD_OCR", + "RTD_Omni", + }, f"RTD by_task keys mismatch: {rtd.get('by_task', {}).keys()}" + assert pr.get("by_task", {}).keys() >= { + "correction", + "proactive_reminder", + "post_event_reminder", + }, f"PR by_task keys mismatch: {pr.get('by_task', {}).keys()}" + + assert rtd.get("mean_content_score", 0.0) >= _MIN_RTD_MEAN_CONTENT_SCORE, ( + f"RTD content score {rtd.get('mean_content_score'):.4f} < {_MIN_RTD_MEAN_CONTENT_SCORE}" + ) + assert rtd.get("mean_avg_temporal_score", 0.0) >= _MIN_RTD_MEAN_TEMPORAL_SCORE, ( + f"RTD temporal score {rtd.get('mean_avg_temporal_score'):.4f} < {_MIN_RTD_MEAN_TEMPORAL_SCORE}" + ) + assert pr.get("mean_all_success", 0.0) >= _MIN_PR_MEAN_ALL_SUCCESS, ( + f"PR success rate {pr.get('mean_all_success'):.4f} < {_MIN_PR_MEAN_ALL_SUCCESS}" + ) + + +_CLI_TIMEOUT = 5400 + + +def _run_cli(argv: list[str]) -> None: + """Run a CLI command and assert success within timeout. + + stdout/stderr are printed in real-time (not captured) so CI logs + show full generate / evaluate progress — same pattern as PR#6817 + perf tests (``-s -v`` + ``run_benchmark()``). + """ + label = " ".join(argv[:3]) + print(f"\n[Omni-DuplexEval] Running: {' '.join(argv)}\n", flush=True) + try: + result = subprocess.run(argv, capture_output=False, timeout=_CLI_TIMEOUT) + except subprocess.TimeoutExpired: + raise AssertionError(f"[{label}] timed out after {_CLI_TIMEOUT}s: {' '.join(argv)}") from None + # Tolerate -6 (SIGABRT from torch_npu exit heap corruption in vllm-ascend + # v0.29.0-a3 image); the guard assertions below verify artifact integrity. + assert result.returncode in (0, -6), f"failed with unexpected exit code {result.returncode}" + print(f"\n[Omni-DuplexEval] Done: {label}\n", flush=True) diff --git a/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci_mock.py b/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci_mock.py new file mode 100644 index 000000000..ccfea367a --- /dev/null +++ b/tests/e2e/accuracy/omni_duplex_eval/test_omni_duplex_eval_ci_mock.py @@ -0,0 +1,614 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""CPU-based mock tests for the Omni-DuplexEval CI guard. + +Runs the *real* ``test_omni_duplex_eval_ci()`` guard function without GPU / +real model by stubbing the heavy I/O: + + - ``_run_cli`` is replaced by a spy that records argv (verifies the exact + ``vllm bench omni-duplex-eval`` argument assembly) and never executes. + - ``resolve_ref_audio`` is stubbed to a local path. + - Fake score files are pre-written under ``tmp_path/scores//`` using + the sample IDs from ``omni_duplex_eval_ci_config.json`` (minus excludes), + so ``summarize_scores()`` and the Phase-4 assertions run for real. + +This validates parameter construction, the score-file-count assertion, the +``protocol_pin`` check, the RTD/PR ``by_task`` key checks, and the threshold +comparisons — all against the genuine guard logic. + +Also includes unit tests for the ``judge_server`` fixture endpoint resolution, +``_run_cli`` (success / non-zero / timeout), and ``summarize_scores`` shape. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass +from pathlib import Path + +import pytest + +from tests.e2e.accuracy.omni_duplex_eval import conftest +from vllm_omni.benchmarks.duplex.omni_duplex_eval_eval import summarize_scores +from vllm_omni.benchmarks.duplex.omni_duplex_eval_metrics import PROTOCOL_PIN + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +_CONFIG_PATH = Path(__file__).resolve().parent / "omni_duplex_eval_ci_config.json" + + +def _load_config() -> dict: + return json.loads(_CONFIG_PATH.read_text(encoding="utf-8")) + + +def _sample_ids(config: dict) -> list[str]: + """Mirror the ID assembly in ``test_omni_duplex_eval_ci()``.""" + all_ids = sorted(set(sid for sids in config["sample_ids_per_split"].values() for sid in sids)) + exclude = set(config.get("exclude_ids", [])) + return [sid for sid in all_ids if sid not in exclude] + + +def _rtd_task_type(split: str) -> str: + return split.replace("RTD_", "").lower() + + +def _pr_task_type(split: str) -> str: + if "correction" in split: + return "correction" + if "post_event" in split: + return "post_event_reminder" + return "proactive_reminder" + + +# --------------------------------------------------------------------------- +# Fake score file writers (shaped like evaluate_sample() output) +# --------------------------------------------------------------------------- + + +def _fake_rtd_score( + sample_id: str, split: str, *, content_score: float = 2.8, temporal_score: int = 3, protocol_pin: str = PROTOCOL_PIN +) -> dict: + return { + "id": sample_id, + "family": "rtd", + "task_type": _rtd_task_type(split), + "protocol_pin": protocol_pin, + "temporal": { + "sentences": [ + { + "sentence": "The object is moving left.", + "start": 0.0, + "end": 2.0, + "sentence_start": 0.0, + "sentence_end": 2.0, + "sentence_duration": 2.0, + "window_start": 0.0, + "window_end": 1.0, + "error": None, + "temporal_score": temporal_score, + "is_relevant": 1, + } + ], + "summary": { + "avg_temporal_score": float(temporal_score), + "total_sentences": 1, + "evaluated_sentences": 1, + "error_count": 0, + "relevant_sentences_count": 1, + "irrelevant_sentences_count": 0, + "score_distribution": { + "3_points": 1 if temporal_score == 3 else 0, + "2_points": 1 if temporal_score == 2 else 0, + "1_point": 0, + "0_points": 0, + }, + "relevance_stats": { + "first_relevant_time": 0.0, + "total_relevant_duration": 2.0, + "total_irrelevant_duration": 0.0, + "irrelevant_duration_ratio": 0.0, + }, + }, + }, + "content": {"content_score": content_score, "content_reasoning": "Looks correct."}, + } + + +def _fake_pr_score(sample_id: str, split: str, *, all_success: bool = True, protocol_pin: str = PROTOCOL_PIN) -> dict: + return { + "id": sample_id, + "family": "pr", + "task_type": _pr_task_type(split), + "protocol_pin": protocol_pin, + "pr": { + "events": [ + { + "response_segment": "I see the object moving.", + "success_score": 1 if all_success else 0, + "reasoning": "Correct.", + } + ], + "all_success": all_success, + "total_score": 1 if all_success else 0, + }, + } + + +def _write_scores_from_config( + score_root: Path, + config: dict, + *, + content_score: float = 2.8, + temporal_score: int = 3, + pr_success: bool = True, + protocol_pin: str = PROTOCOL_PIN, +) -> int: + """Write one fake score file per selected sample ID into the split dir. + + Returns the number of files written (== expected ``summary["samples"]``). + """ + exclude = set(config.get("exclude_ids", [])) + count = 0 + for split, ids in config["sample_ids_per_split"].items(): + subdir = score_root / split + subdir.mkdir(parents=True, exist_ok=True) + for sid in ids: + if sid in exclude: + continue + if split.startswith("RTD"): + row = _fake_rtd_score( + sid, split, content_score=content_score, temporal_score=temporal_score, protocol_pin=protocol_pin + ) + else: + row = _fake_pr_score(sid, split, all_success=pr_success, protocol_pin=protocol_pin) + (subdir / f"{sid}.json").write_text(json.dumps(row), encoding="utf-8") + count += 1 + return count + + +# --------------------------------------------------------------------------- +# Driving the REAL guard test function +# --------------------------------------------------------------------------- + + +@dataclass +class _FakeServer: + """Minimal typed stand-in for the ``omni_server`` fixture. + + Only the read-only attributes the guard touches (host / port / model) are + modeled; keeps the CPU-only mock free of the real OmniServer dependency. + """ + + host: str + port: int + model: str + + +def _call_real_guard( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, judge_url: str = "http://127.0.0.1:8001" +) -> list[list[str]]: + """Call ``test_omni_duplex_eval_ci()`` with stubbed heavy I/O. + + Returns the argv lists captured by the ``_run_cli`` spy. + """ + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + fake_server = _FakeServer(host="127.0.0.1", port=9999, model="mock-model") + calls: list[list[str]] = [] + + def spy_run_cli(argv: list[str]) -> None: + calls.append(argv) + + monkeypatch.setattr(guard, "_run_cli", spy_run_cli) + monkeypatch.setattr(guard, "resolve_ref_audio", lambda: tmp_path / "ref.wav") + monkeypatch.setattr(guard, "_RESULT_DIR", tmp_path) # write summary.json to tmp_path, not real results/ + guard.test_omni_duplex_eval_ci(fake_server, tmp_path, judge_url) + return calls + + +def test_real_guard_full_flow_passes(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard passes end-to-end with good scores and stubbed CLI.""" + config = _load_config() + expected = len(_sample_ids(config)) + written = _write_scores_from_config(tmp_path / "scores", config) + assert written == expected, f"test setup: expected {expected} score files, wrote {written}" + _call_real_guard(monkeypatch, tmp_path) # must not raise + + +def test_real_guard_assembles_generate_and_evaluate_argv(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Verify the exact CLI argument assembly for both phases.""" + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config) + calls = _call_real_guard(monkeypatch, tmp_path) + + assert len(calls) == 2, f"expected generate + evaluate, got {len(calls)} calls" + generate, evaluate = calls + assert generate[:5] == ["vllm", "bench", "omni-duplex-eval", "--omni", "generate"] + assert evaluate[:5] == ["vllm", "bench", "omni-duplex-eval", "--omni", "evaluate"] + + # generate: dataset ref, split, ids, url, model, ref-audio, response-root + assert generate[generate.index("--dataset") + 1] == config["dataset"] + assert generate[generate.index("--split") + 1] == "all" + gen_ids = generate[generate.index("--ids") + 1 : generate.index("--url")] + assert gen_ids == _sample_ids(config) + assert generate[generate.index("--url") + 1] == "ws://127.0.0.1:9999/v1/realtime?duplex=1" + assert generate[generate.index("--model") + 1] == "mock-model" + assert Path(generate[generate.index("--ref-audio") + 1]).name == "ref.wav" + assert generate[generate.index("--response-root") + 1] == str(tmp_path / "responses") + + # evaluate: dataset ref, ids, roots, judge config + assert evaluate[evaluate.index("--dataset") + 1] == config["dataset"] + eval_ids = evaluate[evaluate.index("--ids") + 1 : evaluate.index("--response-root")] + assert eval_ids == _sample_ids(config) + assert evaluate[evaluate.index("--response-root") + 1] == str(tmp_path / "responses") + assert evaluate[evaluate.index("--score-root") + 1] == str(tmp_path / "scores") + assert evaluate[evaluate.index("--judge-model") + 1] == config["judge"]["model"] + assert evaluate[evaluate.index("--judge-base-url") + 1] == "http://127.0.0.1:8001" + assert evaluate[evaluate.index("--judge-video-mode") + 1] == config["judge"].get("video_mode", "video_url") + assert int(evaluate[evaluate.index("--judge-fps") + 1]) == config["judge"].get("fps", 2) + assert int(evaluate[evaluate.index("--eval-workers") + 1]) == config["judge"].get("eval_workers", 4) + + +def test_real_guard_rejects_unsupported_dataset_revision(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Non-null ``dataset_revision`` is currently unsupported and must raise.""" + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config) + + patched = dict(config) + patched["dataset_revision"] = "abcd1234" + monkeypatch.setattr(guard, "_load_ci_config", lambda: patched) + + with pytest.raises(NotImplementedError, match="dataset_revision pinning is not yet supported"): + _call_real_guard(monkeypatch, tmp_path) + + +def test_real_guard_fails_on_low_rtd_content(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard raises when RTD content score is below threshold. + + The production thresholds are 0.0 (record mode). We temporarily inject + a non-zero threshold to verify the assertion logic still catches low scores. + """ + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + monkeypatch.setattr(guard, "_MIN_RTD_MEAN_CONTENT_SCORE", 2.0) + monkeypatch.setattr(guard, "_MIN_RTD_MEAN_TEMPORAL_SCORE", 0.0) + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config, content_score=1.0, temporal_score=3) + with pytest.raises(AssertionError, match="RTD content score"): + _call_real_guard(monkeypatch, tmp_path) + + +def test_real_guard_fails_on_low_pr_success(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard raises when PR success rate is below threshold. + + The production thresholds are 0.0 (record mode). We temporarily inject + a non-zero threshold to verify the assertion logic still catches low scores. + """ + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + monkeypatch.setattr(guard, "_MIN_PR_MEAN_ALL_SUCCESS", 0.5) + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config, pr_success=False) + with pytest.raises(AssertionError, match="PR success rate"): + _call_real_guard(monkeypatch, tmp_path) + + +def test_real_guard_fails_on_missing_score_files(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard raises when score file count does not match sample count.""" + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config) + # Add an extra file to break the count assertion + extra_dir = tmp_path / "scores" / "RTD_world_knowledge" + extra_dir.mkdir(parents=True, exist_ok=True) + (extra_dir / "extra.json").write_text("{}", encoding="utf-8") + with pytest.raises(AssertionError, match="expected .* score files"): + _call_real_guard(monkeypatch, tmp_path) + + +def test_real_guard_fails_on_stale_score_file_protocol_pin(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard raises when an individual score file's protocol_pin does not + match the configured pin (simulates stale files from a previous protocol version). + """ + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config) + + # Force the config pin back to the real value so the config-vs-constant + # check passes; the per-file check should still fail because one file + # contains a stale pin. + monkeypatch.setattr(guard, "PROTOCOL_PIN", PROTOCOL_PIN) + monkeypatch.setattr(guard, "_load_ci_config", lambda: config) + + # Overwrite one score file with a stale pin + rtd_dir = tmp_path / "scores" / "RTD_world_knowledge" + stale = json.loads((rtd_dir / "360.json").read_text(encoding="utf-8")) + stale["protocol_pin"] = "deadbeef" + (rtd_dir / "360.json").write_text(json.dumps(stale), encoding="utf-8") + + with pytest.raises(AssertionError, match="protocol_pin mismatch"): + _call_real_guard(monkeypatch, tmp_path) + + +def test_real_guard_fails_when_config_protocol_pin_drifts(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + """Real guard raises when the JSON config's protocol_pin drifts from the + source-of-truth constant in the Python code.""" + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + config = _load_config() + _write_scores_from_config(tmp_path / "scores", config) + + patched = dict(config) + patched["protocol_pin"] = "deadbeef" + monkeypatch.setattr(guard, "_load_ci_config", lambda: patched) + + with pytest.raises(AssertionError, match="does not match source of truth"): + _call_real_guard(monkeypatch, tmp_path) + + +# --------------------------------------------------------------------------- +# judge_server fixture endpoint resolution logic (mirrors conftest.py) +# --------------------------------------------------------------------------- + + +def _resolve_judge_url() -> str: + """Mirror of the ``judge_server`` fixture body in ``conftest.py``.""" + import os + + config = _load_config() + judge_cfg = config["judge"] + env_name = judge_cfg.get("base_url_env") + base_url = os.environ.get(env_name) if env_name else None + if base_url is not None: + return base_url + return str(judge_cfg.get("base_url", "http://127.0.0.1:8001")) + + +def test_judge_server_uses_env_override(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("VLLM_DUPLEX_EVAL_JUDGE_URL", "http://judge-env:8000") + assert _resolve_judge_url() == "http://judge-env:8000" + + +def test_judge_server_falls_back_to_config(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("VLLM_DUPLEX_EVAL_JUDGE_URL", raising=False) + config = _load_config() + assert _resolve_judge_url() == config["judge"]["base_url"] + + +# --------------------------------------------------------------------------- +# _run_cli unit tests +# --------------------------------------------------------------------------- + + +def test_run_cli_success() -> None: + from tests.e2e.accuracy.omni_duplex_eval.test_omni_duplex_eval_ci import _run_cli + + _run_cli(["echo", "ok"]) + + +def test_run_cli_nonzero_exit() -> None: + from tests.e2e.accuracy.omni_duplex_eval.test_omni_duplex_eval_ci import _run_cli + + with pytest.raises(AssertionError, match="failed"): + _run_cli(["false"]) + + +def test_run_cli_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + from tests.e2e.accuracy.omni_duplex_eval import test_omni_duplex_eval_ci as guard + + monkeypatch.setattr(guard, "_CLI_TIMEOUT", 1) + with pytest.raises(AssertionError, match="timed out"): + guard._run_cli(["sleep", "5"]) + + +# --------------------------------------------------------------------------- +# summarize_scores shape / boundary tests +# --------------------------------------------------------------------------- + + +@pytest.fixture +def score_root(tmp_path: Path) -> Path: + root = tmp_path / "scores" + root.mkdir(parents=True, exist_ok=True) + return root + + +def test_summarize_scores_from_config(score_root: Path) -> None: + """Full config-derived score set produces the expected summary keys.""" + config = _load_config() + _write_scores_from_config(score_root, config) + summary = summarize_scores(score_root) + assert summary["protocol_pin"] == PROTOCOL_PIN + assert summary["samples"] == len(_sample_ids(config)) + assert summary["rtd"]["by_task"].keys() == { + "RTD_world_knowledge", + "RTD_counting", + "RTD_fine_grained_movement", + "RTD_interaction_relation", + "RTD_OCR", + "RTD_Omni", + } + assert summary["pr"]["by_task"].keys() == { + "correction", + "proactive_reminder", + "post_event_reminder", + } + + +def test_summarize_scores_empty(score_root: Path) -> None: + """Empty score directory returns minimal summary.""" + summary = summarize_scores(score_root) + assert summary["protocol_pin"] == PROTOCOL_PIN + assert summary["samples"] == 0 + assert "rtd" not in summary + assert "pr" not in summary + + +# --------------------------------------------------------------------------- +# _parse_visible_devices / _omni_server_device_env unit tests +# +# These run the *real* device-allocation logic from conftest.py on a CPU host +# by faking the torch device probes, covering both unset and pre-populated +# visibility variables (reviewer-requested coverage). +# +# The key regression property: ``_omni_server_device_env()`` returns +# per-subprocess env overrides for the Omni server without ever mutating the +# pytest parent process's environment (a session-scoped mutation previously +# leaked a narrowed ``CUDA_VISIBLE_DEVICES``/``ASCEND_RT_VISIBLE_DEVICES`` to +# unrelated multi-GPU tests in the same pytest session). +# --------------------------------------------------------------------------- + + +class _FakeAccelerator: + """Minimal ``torch.accelerator`` stand-in for CPU mock tests.""" + + def __init__(self, count: int) -> None: + self._count = count + + def device_count(self) -> int: + return self._count + + +class _FakeNpu: + """Minimal ``torch.npu`` stand-in for CPU mock tests.""" + + def __init__(self, available: bool) -> None: + self._available = available + + def is_available(self) -> bool: + return self._available + + +def _patch_torch_cuda( + monkeypatch: pytest.MonkeyPatch, + *, + available: bool, + count: int = 0, +) -> None: + """Route the conftest probes to the CUDA branch.""" + monkeypatch.setattr(conftest.torch.cuda, "is_available", lambda: available) + # ``torch.accelerator`` may not exist on older torch builds. + monkeypatch.setattr(conftest.torch, "accelerator", _FakeAccelerator(count), raising=False) + # Ensure the NPU branch is never selected while testing the CUDA path. + monkeypatch.setattr(conftest.torch, "npu", _FakeNpu(False), raising=False) + + +def _patch_torch_npu( + monkeypatch: pytest.MonkeyPatch, + *, + available: bool, + count: int = 0, +) -> None: + """Route the conftest probes to the NPU branch.""" + monkeypatch.setattr(conftest.torch, "npu", _FakeNpu(available), raising=False) + monkeypatch.setattr(conftest.torch, "accelerator", _FakeAccelerator(count), raising=False) + # Ensure the CUDA branch is never selected while testing the NPU path. + monkeypatch.setattr(conftest.torch.cuda, "is_available", lambda: False) + + +def test_parse_visible_devices_cuda_preset(monkeypatch: pytest.MonkeyPatch) -> None: + """CUDA pre-populated ``2,3``: Omni env -> device 3, judge -> device 2.""" + _patch_torch_cuda(monkeypatch, available=True) + monkeypatch.delenv("ASCEND_RT_VISIBLE_DEVICES", raising=False) + monkeypatch.setenv("CUDA_VISIBLE_DEVICES", "2,3") + + env_key, devices = conftest._parse_visible_devices() + assert env_key == "CUDA_VISIBLE_DEVICES" + assert devices == ["2", "3"] + + # Per-subprocess override for the Omni server (parent env untouched). + assert conftest._omni_server_device_env() == {"CUDA_VISIBLE_DEVICES": "3"} + # Judge derives its device from the job allocation, never outside it. + assert devices[0] == "2" + # Regression: the pytest parent process env is not mutated. + assert os.environ["CUDA_VISIBLE_DEVICES"] == "2,3" + + +def test_parse_visible_devices_cuda_unset(monkeypatch: pytest.MonkeyPatch) -> None: + """CUDA unset with 2 physical GPUs: IDs fall back to 0,1.""" + _patch_torch_cuda(monkeypatch, available=True, count=2) + monkeypatch.delenv("ASCEND_RT_VISIBLE_DEVICES", raising=False) + monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False) + + env_key, devices = conftest._parse_visible_devices() + assert env_key == "CUDA_VISIBLE_DEVICES" + assert devices == ["0", "1"] + + assert conftest._omni_server_device_env() == {"CUDA_VISIBLE_DEVICES": "1"} + assert devices[0] == "0" + # Regression: unset stays unset in the parent process. + assert "CUDA_VISIBLE_DEVICES" not in os.environ + + +def test_parse_visible_devices_npu_preset(monkeypatch: pytest.MonkeyPatch) -> None: + """NPU pre-populated ``0,1``: Omni env -> device 1, judge -> device 0.""" + _patch_torch_npu(monkeypatch, available=True) + monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False) + monkeypatch.setenv("ASCEND_RT_VISIBLE_DEVICES", "0,1") + + env_key, devices = conftest._parse_visible_devices() + assert env_key == "ASCEND_RT_VISIBLE_DEVICES" + assert devices == ["0", "1"] + + assert conftest._omni_server_device_env() == {"ASCEND_RT_VISIBLE_DEVICES": "1"} + assert devices[0] == "0" + # Regression: the pytest parent process env is not mutated. + assert os.environ["ASCEND_RT_VISIBLE_DEVICES"] == "0,1" + + +def test_parse_visible_devices_npu_unset(monkeypatch: pytest.MonkeyPatch) -> None: + """NPU unset with 2 physical NPUs: IDs fall back to 0,1.""" + _patch_torch_npu(monkeypatch, available=True, count=2) + monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False) + monkeypatch.delenv("ASCEND_RT_VISIBLE_DEVICES", raising=False) + + env_key, devices = conftest._parse_visible_devices() + assert env_key == "ASCEND_RT_VISIBLE_DEVICES" + assert devices == ["0", "1"] + + assert conftest._omni_server_device_env() == {"ASCEND_RT_VISIBLE_DEVICES": "1"} + assert devices[0] == "0" + # Regression: unset stays unset in the parent process. + assert "ASCEND_RT_VISIBLE_DEVICES" not in os.environ + + +def test_omni_server_device_env_single_device_returns_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Single device (or CPU-only) allocation: no split, no env override.""" + _patch_torch_cuda(monkeypatch, available=True, count=1) + monkeypatch.delenv("ASCEND_RT_VISIBLE_DEVICES", raising=False) + monkeypatch.delenv("CUDA_VISIBLE_DEVICES", raising=False) + + env_key, devices = conftest._parse_visible_devices() + assert env_key == "CUDA_VISIBLE_DEVICES" + assert devices == ["0"] + assert conftest._omni_server_device_env() is None + assert "CUDA_VISIBLE_DEVICES" not in os.environ + + +def test_omni_server_device_env_leaves_parent_visibility_unchanged( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression: configuring/tearing down the split never narrows the parent. + + A session-scoped mutation used to permanently change the pytest process + environment; this test asserts the parent visibility variables are byte-for- + byte identical before and after repeated device-split configuration. + """ + _patch_torch_cuda(monkeypatch, available=True) + monkeypatch.delenv("ASCEND_RT_VISIBLE_DEVICES", raising=False) + monkeypatch.setenv("CUDA_VISIBLE_DEVICES", "2,3") + + before = os.environ.copy() + for _ in range(3): + env_dict = conftest._omni_server_device_env() + assert env_dict == {"CUDA_VISIBLE_DEVICES": "3"} + # Simulate the server teardown: nothing to undo because the parent + # environment was never touched in the first place. + after = os.environ.copy() + assert after == before + assert os.environ["CUDA_VISIBLE_DEVICES"] == "2,3" diff --git a/tests/helpers/fixtures/config.py b/tests/helpers/fixtures/config.py index 2f46004a6..bf9f22219 100644 --- a/tests/helpers/fixtures/config.py +++ b/tests/helpers/fixtures/config.py @@ -18,9 +18,14 @@ def default_env(): keys = ("VLLM_WORKER_MULTIPROC_METHOD", "VLLM_TARGET_DEVICE") previous = {key: os.environ.get(key) for key in keys} os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = previous["VLLM_WORKER_MULTIPROC_METHOD"] or "spawn" - os.environ["VLLM_TARGET_DEVICE"] = previous["VLLM_TARGET_DEVICE"] or ( - "cuda" if torch.cuda.is_available() and torch.accelerator.device_count() > 0 else "cpu" - ) + if previous["VLLM_TARGET_DEVICE"]: + pass # already set, keep it + elif torch.cuda.is_available() and torch.accelerator.device_count() > 0: + os.environ["VLLM_TARGET_DEVICE"] = "cuda" + elif hasattr(torch, "npu") and torch.npu.is_available(): + os.environ["VLLM_TARGET_DEVICE"] = "npu" + else: + os.environ["VLLM_TARGET_DEVICE"] = "cpu" yield for key, value in previous.items(): if value is None: @@ -39,8 +44,12 @@ def default_vllm_config(): desynchronize vLLM init vs request preprocessing (e.g. renderer state). """ # Use CPU device if no GPU is available (e.g., in CI environments) - has_gpu = torch.cuda.is_available() and torch.accelerator.device_count() > 0 - device = "cuda" if has_gpu else "cpu" + if torch.cuda.is_available() and torch.accelerator.device_count() > 0: + device = "cuda" + elif hasattr(torch, "npu") and torch.npu.is_available(): + device = "npu" + else: + device = "cpu" device_config = DeviceConfig(device=device) with set_current_vllm_config(VllmConfig(device_config=device_config)):