From e60c93de64efed8c30cc1dd77d25824b043de262 Mon Sep 17 00:00:00 2001 From: Diptorup Deb Date: Tue, 23 Jun 2026 12:06:11 -0500 Subject: [PATCH] [ROCm] tests: remove ROCm code-coverage tests (relocated to aisw-ci-builder-tester) Per the verl dev lead, the extra ROCm code-coverage tests are not carried in the ROCm/verl fork. They now live in the CI repo (AMD-AIOSS/aisw-ci-builder-tester under verl/common/extra-tests) and are bind-mounted into the coverage tester at runtime, so the fork no longer accumulates AMD-only coverage test files. Removes the 7 coverage test files added by the earlier batches; the underlying ROCm source fixes (e.g. the get_device_uuid fallback) remain in verl. Co-authored-by: Cursor --- tests/utils/test_device_rocm.py | 98 ---- tests/utils/test_fsdp_utils_rocm.py | 467 ------------------ tests/utils/test_import_utils_rocm.py | 171 ------- tests/utils/test_seqlen_balancing_rocm.py | 107 ---- tests/utils/test_torch_functional_rocm.py | 264 ---------- .../rollout_vllm/test_rollout_utils_rocm.py | 242 --------- .../rollout_vllm/test_vllm_smoke_rocm.py | 189 ------- 7 files changed, 1538 deletions(-) delete mode 100644 tests/utils/test_device_rocm.py delete mode 100644 tests/utils/test_fsdp_utils_rocm.py delete mode 100644 tests/utils/test_import_utils_rocm.py delete mode 100644 tests/utils/test_seqlen_balancing_rocm.py delete mode 100644 tests/utils/test_torch_functional_rocm.py delete mode 100644 tests/workers/rollout/rollout_vllm/test_rollout_utils_rocm.py delete mode 100644 tests/workers/rollout/rollout_vllm/test_vllm_smoke_rocm.py diff --git a/tests/utils/test_device_rocm.py b/tests/utils/test_device_rocm.py deleted file mode 100644 index 8a96111e4da..00000000000 --- a/tests/utils/test_device_rocm.py +++ /dev/null @@ -1,98 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""Coverage for verl.utils.device on the ROCm (CUDA-compatible) path. - -verl maps AMD/HIP onto torch's "cuda" device namespace, so on a ROCm host these -accessors exercise the is_cuda_available branches of the device abstraction -- -the primary ROCm support surface. NPU-only helpers are out of scope (excluded in -.coveragerc.rocm). Falls back gracefully on a CPU-only host. -""" - -import types - -import pytest -import torch - -from verl.utils import device as dev - - -def test_availability_flags(): - assert isinstance(dev.is_cuda_available, bool) - assert isinstance(dev.is_npu_available, bool) - assert dev.is_torch_npu_available(check_device=False) in (True, False) - - -def test_device_name_and_module(): - name = dev.get_device_name() - assert name in ("cuda", "npu", "cpu") - mod = dev.get_torch_device() - assert mod is not None - if dev.is_cuda_available: - assert name == "cuda" - assert mod is torch.cuda - - -def test_resource_and_visible_devices_keyword(): - assert dev.get_resource_name() in ("GPU", "NPU") - kw = dev.get_visible_devices_keyword() - assert kw in ("CUDA_VISIBLE_DEVICES", "ASCEND_RT_VISIBLE_DEVICES") - if dev.is_cuda_available: - assert kw == "CUDA_VISIBLE_DEVICES" - - -def test_nccl_backend(): - backend = dev.get_nccl_backend() - assert backend in ("nccl", "hccl") - if dev.is_cuda_available: - assert backend == "nccl" - - -def test_device_id_and_capability(): - if not dev.is_cuda_available: - pytest.skip("no accelerator visible") - assert dev.get_device_id() >= 0 - major, minor = dev.get_device_capability(0) - assert major is not None and minor is not None - - -def test_capability_when_no_cuda(): - # The (None, None) path is only taken without CUDA; just assert the contract. - major, minor = dev.get_device_capability(0) - if not dev.is_cuda_available: - assert (major, minor) == (None, None) - - -def test_set_expandable_segments(): - # No-op on CPU; on ROCm/CUDA it forwards to the allocator. Must not raise. - dev.set_expandable_segments(True) - dev.set_expandable_segments(False) - - -def test_is_support_ipc(): - # On a GPU (ROCm/CUDA) host this is unconditionally True. - result = dev.is_support_ipc() - assert isinstance(result, bool) - if dev.is_cuda_available: - assert result is True - - -def test_auto_set_device_cuda_noop(): - # On a non-NPU host auto_set_device must leave a "cuda" trainer device alone. - cfg = types.SimpleNamespace(trainer=types.SimpleNamespace(device="cuda")) - dev.auto_set_device(cfg) - if not dev.is_npu_available: - assert cfg.trainer.device == "cuda" - # None / missing-attr configs are tolerated. - dev.auto_set_device(None) - dev.auto_set_device(types.SimpleNamespace()) diff --git a/tests/utils/test_fsdp_utils_rocm.py b/tests/utils/test_fsdp_utils_rocm.py deleted file mode 100644 index 3e0e002706c..00000000000 --- a/tests/utils/test_fsdp_utils_rocm.py +++ /dev/null @@ -1,467 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""ROCm code-coverage tests for verl.utils.fsdp_utils. - -Two groups: - -1. Pure-python helpers (no GPU / no torch.distributed): wrap-policy selection, - fsdp_version, optimizer offload/load, the meta-device init context manager, - and the shard-placement helper. These run on CPU and always execute. - -2. A SINGLE-GPU FSDP2 path (world_size=1, spawned via mp.spawn so distributed - state stays isolated from the rest of the curated suite). It wraps a tiny - Qwen2 model with fully_shard and exercises offload/load to CPU/GPU, the full - state-dict collection, fsdp2 grad-norm clipping, and reshard-after-forward. - Genuinely multi-rank FSDP1 sharding paths are out of scope here -- they - belong to the multi-GPU coverage tier. - -The point is COVERAGE of verl's own ROCm-relevant code, not numerical -correctness, so assertions are intentionally light. -""" - -import os -from types import SimpleNamespace - -import pytest -import torch -import torch.nn as nn - - -# --------------------------------------------------------------------------- -# Group 1 -- pure-python helpers (CPU, no distributed) -# --------------------------------------------------------------------------- - - -class _MockDecoderLayer(nn.Module): - def __init__(self, hidden=32): - super().__init__() - self.lin = nn.Linear(hidden, hidden) - - -class _MockCausalLM(nn.Module): - _no_split_modules = ["_MockDecoderLayer"] - - def __init__(self, hidden=32, layers=2): - super().__init__() - self.config = SimpleNamespace(tie_word_embeddings=False) - self.embed = nn.Embedding(16, hidden) - self.layers = nn.ModuleList([_MockDecoderLayer(hidden) for _ in range(layers)]) - self.lm_head = nn.Linear(hidden, 16, bias=False) - - -def test_get_fsdp_wrap_policy_disable(): - from verl.utils.fsdp_utils import get_fsdp_wrap_policy - - assert get_fsdp_wrap_policy(_MockCausalLM(), config={"disable": True}) is None - - -def test_get_fsdp_wrap_policy_min_num_params(): - from verl.utils.fsdp_utils import get_fsdp_wrap_policy - - policy = get_fsdp_wrap_policy(_MockCausalLM(), config={"min_num_params": 100}) - assert policy is not None - - -def test_get_fsdp_wrap_policy_transformer_cls(): - from verl.utils.fsdp_utils import get_fsdp_wrap_policy - - # default_transformer_cls_names_to_wrap comes from _no_split_modules - policy = get_fsdp_wrap_policy(_MockCausalLM(), config={}) - assert policy is not None - - -def test_get_fsdp_wrap_policy_lora(): - from verl.utils.fsdp_utils import get_fsdp_wrap_policy - - policy = get_fsdp_wrap_policy(_MockCausalLM(), config={}, is_lora=True) - assert policy is not None - - -def test_get_fsdp_wrap_policy_none_config(): - from verl.utils.fsdp_utils import get_fsdp_wrap_policy - - # config=None -> {} ; no min params and a model with _no_split_modules - assert get_fsdp_wrap_policy(_MockCausalLM(), config=None) is not None - - -def test_fsdp_version_plain_module_is_zero(): - from verl.utils.fsdp_utils import fsdp_version - - assert fsdp_version(nn.Linear(4, 4)) == 0 - - -def test_get_shard_placement_fn(): - from verl.utils.fsdp_utils import get_shard_placement_fn - - fn = get_shard_placement_fn(fsdp_size=4) - # a param whose dim 0 is divisible by 4 - placement = fn(torch.empty(8, 3)) - assert placement is not None - # a param whose only divisible dim is not 0 - placement2 = fn(torch.empty(3, 8)) - assert placement2 is not None - # nothing divisible -> falls back to Shard(0) - placement3 = fn(torch.empty(3, 3)) - assert placement3 is not None - - -def test_offload_and_load_fsdp_optimizer_cpu(): - from verl.utils.fsdp_utils import load_fsdp_optimizer, offload_fsdp_optimizer - - model = nn.Linear(8, 8) - opt = torch.optim.AdamW(model.parameters(), lr=1e-3) - # empty state -> early return path - offload_fsdp_optimizer(opt) - load_fsdp_optimizer(opt, device_id=torch.device("cpu")) - # build optimizer state via a step - loss = model(torch.randn(2, 8)).sum() - loss.backward() - opt.step() - offload_fsdp_optimizer(opt) - load_fsdp_optimizer(opt, device_id=torch.device("cpu")) - - -def test_meta_device_init_creates_meta_params(): - from verl.utils.fsdp_utils import meta_device_init - - with meta_device_init(): - m = nn.Linear(8, 8) - assert m.weight.is_meta - - -def test_normalize_peft_param_name(): - from verl.utils.fsdp_utils import normalize_peft_param_name - - params = { - "base_model.model.model.embed_tokens.weight": 1, - "base_model.model.model.layers.0.self_attn.q_proj.base_layer.weight": 2, - # LoRA delta keys must be stripped out - "base_model.model.model.layers.0.self_attn.q_proj.lora_A.default.weight": 3, - "base_model.model.model.layers.0.self_attn.q_proj.lora_B.default.weight": 4, - } - out = normalize_peft_param_name(params) - assert "model.embed_tokens.weight" in out - assert "model.layers.0.self_attn.q_proj.weight" in out - assert not any("lora_" in k for k in out) - - -def test_replace_lora_wrapper_passthrough(): - """Non-target keys pass through unchanged; target keys get base_layer suffix.""" - from verl.utils.fsdp_utils import replace_lora_wrapper - - try: - from peft import LoraConfig - - peft_config = LoraConfig(r=8, target_modules=["q_proj"], task_type="CAUSAL_LM") - except Exception as exc: # pragma: no cover - peft must be present in the tier - pytest.skip(f"peft unavailable: {exc}") - - # a non-weight/bias key returns unchanged - assert replace_lora_wrapper("model.norm", peft_config) == "model.norm" - # a stacked-param weight gets rewritten to base_layer - out = replace_lora_wrapper("model.layers.0.self_attn.q_proj.weight", peft_config) - assert out.endswith(".weight") - - -# --------------------------------------------------------------------------- -# Group 2 -- single-GPU FSDP2 path (world_size=1, isolated subprocess) -# --------------------------------------------------------------------------- - - -def _fsdp2_single_gpu_worker(rank, rendezvous_file): - import torch.distributed as dist - from torch.distributed import init_device_mesh - from transformers import Qwen2Config - from transformers import AutoModelForCausalLM - - from verl.utils.device import get_device_name, get_nccl_backend, get_torch_device - from verl.utils.fsdp_utils import ( - MixedPrecisionPolicy, - apply_fsdp2, - fsdp2_clip_grad_norm_, - fsdp2_load_full_state_dict, - fsdp2_sharded_load_from_cpu, - fsdp2_sharded_save_to_cpu, - fsdp_version, - get_fsdp_full_state_dict, - get_init_weight_context_manager, - load_fsdp2_model_to_gpu, - offload_fsdp2_model_to_cpu, - set_reshard_after_forward, - ) - - world_size = 1 - get_torch_device().set_device(rank) - dist.init_process_group( - backend=get_nccl_backend(), - init_method=f"file://{rendezvous_file}", - rank=rank, - world_size=world_size, - ) - try: - device_mesh = init_device_mesh(get_device_name(), mesh_shape=(world_size,), mesh_dim_names=("dp",)) - - # init-weight context manager (both branches) - _ = get_init_weight_context_manager(use_meta_tensor=True, mesh=device_mesh) - _ = get_init_weight_context_manager(use_meta_tensor=False) - - cfg = Qwen2Config( - num_hidden_layers=2, - num_attention_heads=2, - num_key_value_heads=2, - hidden_size=128, - intermediate_size=256, - vocab_size=512, - ) - with torch.device(get_device_name()): - model = AutoModelForCausalLM.from_config( - config=cfg, torch_dtype=torch.bfloat16, attn_implementation="eager" - ) - model = model.to(device=get_device_name()) - - mp_policy = MixedPrecisionPolicy( - param_dtype=torch.bfloat16, reduce_dtype=torch.float32, cast_forward_inputs=True - ) - apply_fsdp2(model, {"mesh": device_mesh, "mp_policy": mp_policy}, {}) - assert fsdp_version(model) == 2 - - # reshard-after-forward toggle (the runtime override helper) - try: - set_reshard_after_forward(model, True) - except Exception as exc: # torch < 2.8 may lack the internal API - print(f"set_reshard_after_forward non-fatal: {type(exc).__name__}: {exc}") - - # offload to CPU then load back to GPU - offload_fsdp2_model_to_cpu(model) - load_fsdp2_model_to_gpu(model) - - # full (gathered) state dict -- fsdp2 path - sd = get_fsdp_full_state_dict(model, offload_to_cpu=True, rank0_only=True) - assert isinstance(sd, dict) and len(sd) > 0 - - # a tiny fwd/bwd to populate grads, then fsdp2 grad-norm clip - input_ids = torch.randint(0, 512, (1, 8), device=get_device_name()) - out = model(input_ids=input_ids, labels=input_ids) - out.loss.backward() - total_norm = fsdp2_clip_grad_norm_(model.parameters(), max_norm=1.0) - assert total_norm is not None - - # broadcast a full state dict back into the sharded model (rank0 path) - fsdp2_load_full_state_dict(model, sd, device_mesh=device_mesh, cpu_offload=None) - - # sharded save (local DTensor shards -> CPU) then load back - cpu_state, global_spec = fsdp2_sharded_save_to_cpu(model) - assert isinstance(cpu_state, dict) and global_spec is not None - fsdp2_sharded_load_from_cpu(model, cpu_state, global_spec) - finally: - dist.barrier() - dist.destroy_process_group() - - -def test_fsdp2_single_gpu_paths(): - if not torch.cuda.is_available(): - pytest.skip("no GPU visible") - pytest.importorskip("transformers") - pytest.importorskip("accelerate") - - import tempfile - - import torch.multiprocessing as mp - - with tempfile.TemporaryDirectory() as tmp: - rendezvous_file = os.path.join(tmp, "rdzv_fsdp2_single") - mp.spawn(fn=_fsdp2_single_gpu_worker, args=(rendezvous_file,), nprocs=1, join=True) - - -def _fsdp1_single_gpu_worker(rank, rendezvous_file): - """FSDP1 (FullyShardedDataParallel) at world_size=1 -- exercises the v1 - offload/load/full-state-dict branches that the fsdp2 worker does not hit.""" - import torch.distributed as dist - from torch.distributed.fsdp import FullyShardedDataParallel as FSDP - - from verl.utils.device import get_device_name, get_nccl_backend, get_torch_device - from verl.utils.fsdp_utils import ( - fsdp_version, - get_fsdp_full_state_dict, - load_fsdp_model_to_gpu, - offload_fsdp_model_to_cpu, - ) - - get_torch_device().set_device(rank) - dist.init_process_group( - backend=get_nccl_backend(), init_method=f"file://{rendezvous_file}", rank=rank, world_size=1 - ) - try: - plain = nn.Sequential(nn.Linear(32, 32), nn.ReLU(), nn.Linear(32, 16)).to(get_device_name()) - model = FSDP(plain, device_id=get_torch_device().current_device()) - assert fsdp_version(model) == 1 - - # full (gathered) state dict via the FSDP1 FULL_STATE_DICT context - sd = get_fsdp_full_state_dict(model, offload_to_cpu=True, rank0_only=True) - assert isinstance(sd, dict) - - # FSDP1 offload-to-CPU / load-to-GPU branches (flat-param handle walk) - offload_fsdp_model_to_cpu(model) - load_fsdp_model_to_gpu(model) - finally: - dist.barrier() - dist.destroy_process_group() - - -def test_fsdp1_single_gpu_paths(): - if not torch.cuda.is_available(): - pytest.skip("no GPU visible") - import tempfile - - import torch.multiprocessing as mp - - with tempfile.TemporaryDirectory() as tmp: - rendezvous_file = os.path.join(tmp, "rdzv_fsdp1_single") - mp.spawn(fn=_fsdp1_single_gpu_worker, args=(rendezvous_file,), nprocs=1, join=True) - - -def _lora_fsdp2_worker(rank, rendezvous_file): - """LoRA + FSDP2 collect / merge / unmerge paths at world_size=1. - - These weight-gathering paths use FSDP.summon_full_params on fully_shard - modules and depend on the exact peft/torch versions in the tier, so each - step is best-effort: failures are logged, not raised, so we still collect - coverage for whatever executes without breaking the curated set. - """ - import torch.distributed as dist - from torch.distributed import init_device_mesh - from transformers import AutoModelForCausalLM, Qwen2Config - - from verl.utils.device import get_device_name, get_nccl_backend, get_torch_device - from verl.utils.fsdp_utils import MixedPrecisionPolicy, apply_fsdp2 - - get_torch_device().set_device(rank) - dist.init_process_group( - backend=get_nccl_backend(), init_method=f"file://{rendezvous_file}", rank=rank, world_size=1 - ) - - def _aux(label, fn): - try: - fn() - print(f"LORA {label}: ok") - except Exception as exc: # noqa: BLE001 - coverage, not correctness - print(f"LORA {label}: non-fatal {type(exc).__name__}: {exc}") - - try: - from peft import LoraConfig, get_peft_model - - device_mesh = init_device_mesh(get_device_name(), mesh_shape=(1,), mesh_dim_names=("fsdp",)) - cfg = Qwen2Config( - num_hidden_layers=2, - num_attention_heads=2, - num_key_value_heads=2, - hidden_size=64, - intermediate_size=128, - vocab_size=256, - ) - with torch.device(get_device_name()): - base = AutoModelForCausalLM.from_config(config=cfg, torch_dtype=torch.float32, attn_implementation="eager") - lora_cfg = LoraConfig(r=8, lora_alpha=16, target_modules=["q_proj", "v_proj"], task_type="CAUSAL_LM") - model = get_peft_model(base, lora_cfg).to(get_device_name()) - - mp_policy = MixedPrecisionPolicy(param_dtype=torch.float32, reduce_dtype=torch.float32) - apply_fsdp2( - model, - {"mesh": device_mesh, "mp_policy": mp_policy}, - {"wrap_policy": {"transformer_layer_cls_to_wrap": ["Qwen2DecoderLayer"]}}, - ) - - from verl.utils.fsdp_utils import ( - backup_base_model_weights, - collect_lora_params, - collect_merged_lora_params, - fsdp_merge_unmerge, - merged_lora_context, - restore_base_model_weights, - ) - - _aux("collect_lora_params(base_sync_done)", lambda: collect_lora_params(model, False, True)) - _aux("backup/restore_base", lambda: restore_base_model_weights(model, backup_base_model_weights(model))) - _aux("merge_unmerge", lambda: (fsdp_merge_unmerge(model, True), fsdp_merge_unmerge(model, False))) - _aux("collect_merged_lora_params", lambda: collect_merged_lora_params(model)) - - def _merged_ctx(): - with merged_lora_context(model, backup_adapters=True): - pass - with merged_lora_context(model, backup_adapters=False): - pass - - _aux("merged_lora_context", _merged_ctx) - finally: - dist.barrier() - dist.destroy_process_group() - - -def test_lora_fsdp2_paths(): - if not torch.cuda.is_available(): - pytest.skip("no GPU visible") - pytest.importorskip("transformers") - pytest.importorskip("peft") - - import tempfile - - import torch.multiprocessing as mp - - with tempfile.TemporaryDirectory() as tmp: - rendezvous_file = os.path.join(tmp, "rdzv_lora_fsdp2") - mp.spawn(fn=_lora_fsdp2_worker, args=(rendezvous_file,), nprocs=1, join=True) - - -def _parallel_safetensors_worker(rank, rendezvous_file, ckpt_dir): - import torch.distributed as dist - - from verl.utils.device import get_nccl_backend, get_torch_device - from verl.utils.fsdp_utils import parallel_load_safetensors - - get_torch_device().set_device(rank) - dist.init_process_group( - backend=get_nccl_backend(), init_method=f"file://{rendezvous_file}", rank=rank, world_size=1 - ) - try: - # single-file checkpoint (no index json) -> the model.safetensors branch - shard_states = parallel_load_safetensors(ckpt_dir) - assert isinstance(shard_states, dict) and len(shard_states) > 0 - finally: - dist.barrier() - dist.destroy_process_group() - - -def test_parallel_load_safetensors(): - if not torch.cuda.is_available(): - pytest.skip("no GPU visible") - pytest.importorskip("safetensors") - - import tempfile - - import torch.multiprocessing as mp - from safetensors.torch import save_file - - with tempfile.TemporaryDirectory() as ckpt_dir: - save_file( - {"a.weight": torch.randn(4, 4), "b.bias": torch.randn(4)}, - os.path.join(ckpt_dir, "model.safetensors"), - ) - rendezvous_file = os.path.join(ckpt_dir, "rdzv_safetensors") - mp.spawn(fn=_parallel_safetensors_worker, args=(rendezvous_file, ckpt_dir), nprocs=1, join=True) - - -if __name__ == "__main__": - import sys - - sys.exit(pytest.main([__file__, "-v"])) diff --git a/tests/utils/test_import_utils_rocm.py b/tests/utils/test_import_utils_rocm.py deleted file mode 100644 index 3407edb1c59..00000000000 --- a/tests/utils/test_import_utils_rocm.py +++ /dev/null @@ -1,171 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""ROCm code-coverage tests for verl.utils.import_utils. - -This module is pure-python (no GPU / no heavy deps), so it runs anywhere the -verl package imports. It complements tests/utils/test_import_utils_on_cpu.py -(which only exercises load_extern_object) by covering the availability probes, -load_module (pkg:// + file:// + module_name caching + error paths), the -`deprecated` decorator (function and class forms), load_class_from_fqn, and the -deprecated load_extern_type wrapper. -""" - -import os -import sys -import warnings - -import pytest - -from verl.utils.import_utils import ( - deprecated, - import_external_libs, - is_megatron_core_available, - is_msprobe_available, - is_nvtx_available, - is_sglang_available, - is_trl_available, - is_vllm_available, - load_class_from_fqn, - load_extern_type, - load_module, -) - -TEST_MODULE_PATH = os.path.join(os.path.dirname(__file__), "_test_module.py") - - -def test_availability_probes_return_bool(): - # We only assert the type; the truth value depends on the environment. - for probe in ( - is_megatron_core_available, - is_vllm_available, - is_sglang_available, - is_nvtx_available, - is_trl_available, - is_msprobe_available, - ): - assert isinstance(probe(), bool) - - -def test_import_external_libs_variants(): - # None -> no-op - assert import_external_libs(None) is None - # single string is wrapped into a list - import_external_libs("os") - # list of importable modules - import_external_libs(["os", "sys"]) - - -def test_load_module_empty_returns_none(): - assert load_module("") is None - - -def test_load_module_pkg_prefix(): - mod = load_module("pkg://verl.utils.import_utils") - assert mod is not None and hasattr(mod, "load_module") - - -def test_load_module_file_prefix_and_caching(): - name = "verl_cc_loaded_test_module" - sys.modules.pop(name, None) - try: - mod = load_module(f"file://{TEST_MODULE_PATH}", module_name=name) - assert mod is not None and hasattr(mod, "TestClass") - # module_name was registered in sys.modules - assert sys.modules.get(name) is mod - finally: - sys.modules.pop(name, None) - - -def test_load_module_missing_file_raises(): - with pytest.raises(FileNotFoundError): - load_module("/nonexistent/verl_cc/module.py") - - -def test_load_module_name_collision_raises(): - name = "verl_cc_collision_module" - sys.modules[name] = object() # a different object than what we will load - try: - with pytest.raises(RuntimeError): - load_module(TEST_MODULE_PATH, module_name=name) - finally: - sys.modules.pop(name, None) - - -def test_deprecated_function_warns_and_calls(): - @deprecated(replacement="new_func") - def old_func(a, b): - return a + b - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - assert old_func(2, 3) == 5 - assert any(issubclass(w.category, FutureWarning) for w in caught) - - -def test_deprecated_function_without_replacement(): - @deprecated() - def old_func(): - return "ok" - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - assert old_func() == "ok" - assert any(issubclass(w.category, FutureWarning) for w in caught) - - -def test_deprecated_class_warns_on_init(): - @deprecated(replacement="NewClass") - class OldClass: - def __init__(self, value): - self.value = value - - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - obj = OldClass(42) - assert obj.value == 42 - assert any(issubclass(w.category, FutureWarning) for w in caught) - - -def test_load_class_from_fqn_success(): - cls = load_class_from_fqn("collections.OrderedDict") - from collections import OrderedDict - - assert cls is OrderedDict - - -def test_load_class_from_fqn_no_dot_raises(): - with pytest.raises(ValueError): - load_class_from_fqn("NoDotHere") - - -def test_load_class_from_fqn_bad_module_raises(): - with pytest.raises(ImportError): - load_class_from_fqn("verl_cc_no_such_module.SomeClass") - - -def test_load_class_from_fqn_missing_attr_raises(): - with pytest.raises(AttributeError): - load_class_from_fqn("collections.NoSuchClass") - - -def test_load_extern_type_deprecated_wrapper(): - with warnings.catch_warnings(record=True) as caught: - warnings.simplefilter("always") - TestClass = load_extern_type(TEST_MODULE_PATH, "TestClass") - assert TestClass is not None and TestClass.__name__ == "TestClass" - assert any(issubclass(w.category, FutureWarning) for w in caught) - - -if __name__ == "__main__": - sys.exit(pytest.main([__file__, "-v"])) diff --git a/tests/utils/test_seqlen_balancing_rocm.py b/tests/utils/test_seqlen_balancing_rocm.py deleted file mode 100644 index 36d00a1a2e2..00000000000 --- a/tests/utils/test_seqlen_balancing_rocm.py +++ /dev/null @@ -1,107 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""ROCm code-coverage tests for verl.utils.seqlen_balancing pure helpers. - -Complements tests/utils/test_seqlen_balancing.py (which focuses on -rearrange_micro_batches / Karmarkar-Karp on the default single-sample path) -by covering the greedy partitioner, the imbalance-logging metric helper, -the equal_size Karmarkar path, and the force_group_size / no-dynamic-balance -branches of rearrange_micro_batches. All pure-python / CPU. -""" - -import torch - -from verl import DataProto -from verl.utils.seqlen_balancing import ( - get_seqlen_balanced_partitions, - greedy_partition, - log_seqlen_unbalance, - rearrange_micro_batches, -) - - -def _covers_all_indices(partitions, n): - seen = set() - for p in partitions: - seen.update(p) - return seen == set(range(n)) - - -def test_greedy_partition_unequal(): - seqlen = [10, 20, 30, 40, 50, 60] - parts = greedy_partition(seqlen, k_partitions=3, equal_size=False) - assert len(parts) == 3 - assert _covers_all_indices(parts, len(seqlen)) - - -def test_greedy_partition_equal_size(): - seqlen = [10, 20, 30, 40, 50, 60] - parts = greedy_partition(seqlen, k_partitions=3, equal_size=True) - assert len(parts) == 3 - # equal_size -> each partition has the same number of items - assert all(len(p) * 3 == len(seqlen) for p in parts) - assert _covers_all_indices(parts, len(seqlen)) - - -def test_karmarkar_karp_equal_size(): - seqlen = [5, 1, 8, 3, 9, 2, 7, 4] - parts = get_seqlen_balanced_partitions(seqlen, k_partitions=2, equal_size=True) - assert len(parts) == 2 - assert all(len(p) * 2 == len(seqlen) for p in parts) - assert _covers_all_indices(parts, len(seqlen)) - - -def test_log_seqlen_unbalance_metrics(): - seqlen = [10, 20, 30, 40] - partitions = [[0, 3], [1, 2]] # balanced: 50 / 50 - metrics = log_seqlen_unbalance(seqlen, partitions, prefix="seq") - for suffix in ("min", "max", "minmax_diff", "balanced_min", "balanced_max", "mean"): - assert f"seq/{suffix}" in metrics - assert metrics["seq/balanced_min"] == 50 - assert metrics["seq/balanced_max"] == 50 - - -def _make_batch(batch_size=8, seqlen=16): - input_ids = torch.randint(low=1, high=50, size=(batch_size, seqlen)) - attention_mask = torch.ones(batch_size, seqlen, dtype=torch.long) - # vary effective lengths so partitioning is non-trivial - for i in range(batch_size): - attention_mask[i, seqlen - (i % 4) :] = 0 - data = {"input_ids": input_ids, "attention_mask": attention_mask} - return DataProto.from_single_dict(data).batch - - -def test_rearrange_micro_batches_force_group_size(): - batch = _make_batch(batch_size=8, seqlen=16) - micro_batches, idx = rearrange_micro_batches(batch, max_token_len=64, force_group_size=2) - # every group of 2 consecutive samples must stay together in one micro-batch - for partition in idx: - partition = sorted(partition) - for j in range(0, len(partition), 2): - assert partition[j] // 2 == partition[j + 1] // 2 - assert sum(len(p) for p in idx) == 8 - - -def test_rearrange_micro_batches_no_dynamic_balance(): - batch = _make_batch(batch_size=8, seqlen=16) - micro_batches, idx = rearrange_micro_batches(batch, max_token_len=64, use_dynamic_bsz_balance=False) - assert sum(len(p) for p in idx) == 8 - - -if __name__ == "__main__": - import sys - - import pytest - - sys.exit(pytest.main([__file__, "-v"])) diff --git a/tests/utils/test_torch_functional_rocm.py b/tests/utils/test_torch_functional_rocm.py deleted file mode 100644 index c919ab28f03..00000000000 --- a/tests/utils/test_torch_functional_rocm.py +++ /dev/null @@ -1,264 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""Device-side coverage for verl.utils.torch_functional (code-coverage tier). - -These exercise the pure tensor / scheduler helpers in torch_functional on the -active accelerator device (HIP on ROCm), including the flash-attn cross-entropy -path used by logprobs_from_logits. They are single-process and need only one -GPU, so they are safe in the standalone ROCm coverage gate. They fall back to -CPU when no accelerator is visible. -""" - -import math - -import pytest -import torch -import torch.nn as nn - -from verl.utils import torch_functional as tf - - -def _device(): - if torch.cuda.is_available(): - return torch.device("cuda:0") - return torch.device("cpu") - - -DEV = _device() - - -def test_gather_from_labels(): - data = torch.tensor([[0.1, 0.2, 0.3], [0.4, 0.5, 0.6]], device=DEV) - label = torch.tensor([2, 0], device=DEV) - out = tf.gather_from_labels(data, label) - torch.testing.assert_close(out, torch.tensor([0.3, 0.4], device=DEV)) - - -@pytest.mark.parametrize("dtype", [torch.float32, torch.bfloat16]) -def test_logprobs_from_logits_paths(dtype): - torch.manual_seed(0) - logits = torch.randn(4, 7, 11, device=DEV, dtype=dtype) - labels = torch.randint(0, 11, (4, 7), device=DEV) - - # public entry (flash-attn path on GPU, v2 path on CPU) - lp = tf.logprobs_from_logits(logits, labels) - assert lp.shape == (4, 7) - - # naive + v2 reference paths - naive = tf.logprobs_from_logits_naive(logits.float(), labels) - v2 = tf.logprobs_from_logits_v2(logits.float(), labels) - torch.testing.assert_close(naive, v2, atol=1e-4, rtol=1e-4) - - -def test_clip_by_value(): - x = torch.tensor([-2.0, 0.0, 5.0], device=DEV) - lo = torch.tensor([-1.0, -1.0, -1.0], device=DEV) - hi = torch.tensor([1.0, 1.0, 1.0], device=DEV) - out = tf.clip_by_value(x, lo, hi) - torch.testing.assert_close(out, torch.tensor([-1.0, 0.0, 1.0], device=DEV)) - - -def test_entropy_from_logits_and_chunking(): - torch.manual_seed(1) - logits = torch.randn(5, 13, device=DEV) - ent = tf.entropy_from_logits(logits) - ent_chunked = tf.entropy_from_logits_with_chunking(logits, chunk_size=2) - assert ent.shape == (5,) - torch.testing.assert_close(ent, ent_chunked, atol=1e-4, rtol=1e-4) - # entropy is non-negative - assert torch.all(ent >= -1e-5) - - -def test_masked_sum_mean_var_whiten(): - values = torch.tensor([1.0, 2.0, 3.0, 4.0], device=DEV) - mask = torch.tensor([1.0, 1.0, 0.0, 1.0], device=DEV) - assert tf.masked_sum(values, mask).item() == pytest.approx(7.0) - assert tf.masked_mean(values, mask).item() == pytest.approx(7.0 / 3.0, abs=1e-4) - - var = tf.masked_var(values, mask) - assert var.item() > 0 - whit = tf.masked_whiten(values, mask) - assert whit.shape == values.shape - # shift_mean=False re-adds the mean - whit_nomean = tf.masked_whiten(values, mask, shift_mean=False) - assert whit_nomean.shape == values.shape - - -def test_masked_var_error_paths(): - values = torch.tensor([1.0, 2.0], device=DEV) - with pytest.raises(ValueError): - tf.masked_var(values, torch.tensor([0.0, 0.0], device=DEV)) - with pytest.raises(ValueError): - tf.masked_var(values, torch.tensor([1.0, 0.0], device=DEV)) - - -@pytest.mark.parametrize("eos", [1, [1, 2]]) -def test_get_response_mask(eos): - response = torch.tensor( - [[20, 10, 34, 1, 0, 0, 0], [78, 0, 76, 2, 1, 0, 0]], - device=DEV, - ) - mask = tf.get_response_mask(response, eos_token=eos) - assert mask.shape == response.shape - # every row keeps at least its first token - assert torch.all(mask[:, 0] == 1) - - -def test_pad_helpers(): - padded = tf.pad_2d_list_to_length([[1, 2], [3]], pad_token_id=0, max_length=4) - assert padded.tolist() == [[1, 2, 0, 0], [3, 0, 0, 0]] - - t = torch.ones(2, 3, device=DEV) - right = tf.pad_sequence_to_length(t, max_seq_len=5, pad_token_id=0) - assert right.shape == (2, 5) and right[:, 3:].sum() == 0 - left = tf.pad_sequence_to_length(t, max_seq_len=5, pad_token_id=0, left_pad=True) - assert left.shape == (2, 5) and left[:, :2].sum() == 0 - # no-op when already long enough - assert tf.pad_sequence_to_length(t, max_seq_len=2, pad_token_id=0).shape == (2, 3) - - -@pytest.mark.parametrize("truncation", ["left", "right", "middle"]) -def test_postprocess_data_truncation(truncation): - ids = torch.arange(2 * 8, device=DEV).reshape(2, 8) - mask = torch.ones(2, 8, device=DEV, dtype=torch.long) - out_ids, out_mask = tf.postprocess_data(ids, mask, max_length=4, pad_token_id=0, truncation=truncation) - assert out_ids.shape == (2, 4) and out_mask.shape == (2, 4) - - -def test_postprocess_data_pad_and_error(): - ids = torch.arange(2 * 3, device=DEV).reshape(2, 3) - mask = torch.ones(2, 3, device=DEV, dtype=torch.long) - out_ids, out_mask = tf.postprocess_data(ids, mask, max_length=6, pad_token_id=0, left_pad=True) - assert out_ids.shape == (2, 6) - with pytest.raises(NotImplementedError): - tf.postprocess_data(ids, mask, max_length=2, pad_token_id=0, truncation="error") - - -def test_remove_pad_token(): - ids = torch.tensor([[5, 6, 7], [8, 9, 0]], device=DEV) - mask = torch.tensor([[1, 1, 1], [1, 1, 0]], device=DEV) - out = tf.remove_pad_token(ids, mask) - assert out == [[5, 6, 7], [9, 0]] - - -def test_log_probs_from_logits_response(): - torch.manual_seed(2) - input_ids = torch.randint(0, 11, (2, 9), device=DEV) - logits = torch.randn(2, 9, 11, device=DEV) - out = tf.log_probs_from_logits_response(input_ids, logits, response_length=4) - assert out.shape == (2, 4) - - -def test_post_process_logits(): - logits = torch.randn(2, 5, device=DEV) - out = tf.post_process_logits(None, logits.clone(), temperature=2.0, top_k=0, top_p=1.0) - assert out.shape == logits.shape - - -def test_calculate_sum_pi_squared(): - logits = torch.randn(3, 8, device=DEV) - out = tf.calculate_sum_pi_squared_from_logits(logits) - expected = torch.softmax(logits, dim=-1).pow(2).sum(dim=-1) - torch.testing.assert_close(out, expected, atol=1e-5, rtol=1e-5) - - -def test_attention_mask_helpers(): - bsz, tgt = 2, 5 - causal = tf._make_causal_mask((bsz, tgt), torch.float32, DEV) - assert causal.shape == (bsz, 1, tgt, tgt) - attn = torch.ones(bsz, tgt, device=DEV) - expanded = tf._expand_mask(attn, torch.float32, tgt_len=tgt) - assert expanded.shape == (bsz, 1, tgt, tgt) - embeds = torch.zeros(bsz, tgt, 4, device=DEV) - combined = tf.prepare_decoder_attention_mask(attn, (bsz, tgt), embeds) - assert combined.shape == (bsz, 1, tgt, tgt) - - -def test_get_unpad_data(): - attn = torch.tensor([[1, 1, 0], [1, 1, 1]], device=DEV) - indices, cu_seqlens, max_seqlen = tf.get_unpad_data(attn) - assert max_seqlen == 3 - assert cu_seqlens.tolist() == [0, 2, 5] - assert indices.numel() == 5 - - -@pytest.mark.parametrize( - "factory", - [ - lambda opt: tf.get_cosine_schedule_with_warmup(opt, num_warmup_steps=2, num_training_steps=10), - lambda opt: tf.get_cosine_schedule_with_warmup( - opt, num_warmup_steps=2, num_training_steps=10, min_lr_ratio=0.1, zero_indexed_step=False - ), - lambda opt: tf.get_constant_schedule_with_warmup(opt, num_warmup_steps=3), - lambda opt: tf.get_wsd_schedule_with_warmup(opt, num_warmup_steps=2, num_training_steps=10, stable_ratio=0.5), - ], -) -def test_lr_schedules(factory): - model = nn.Linear(2, 2) - opt = torch.optim.SGD(model.parameters(), lr=0.1) - sched = factory(opt) - lrs = [] - for _ in range(12): - opt.step() - sched.step() - lrs.append(opt.param_groups[0]["lr"]) - assert all(math.isfinite(lr) and lr >= 0 for lr in lrs) - - -def test_check_device_is_available(): - if torch.cuda.is_available(): - with tf.check_device_is_available(): - pass - else: - pytest.skip("no accelerator visible") - - -def test_compute_grad_norm(): - model = nn.Linear(3, 1) - x = torch.randn(4, 3) - model(x).sum().backward() - gn = tf.compute_grad_norm(model) - assert gn >= 0 - - -def test_split_dict_tensor_into_batches(): - from tensordict import TensorDict - - td = TensorDict({"a": torch.arange(8).view(8, 1), "b": torch.randn(8, 3)}, batch_size=[8]) - chunks = tf.split_dict_tensor_into_batches(td, batch_size=4) - assert len(chunks) == 2 - assert all(c.batch_size[0] == 4 for c in chunks) - - -def test_use_original_torch_compile_no_mindspeed(): - # mindspeed is not installed in the ROCm tier; the contextmanager must fall - # back to a plain yield via its except branch. - with tf.use_original_torch_compile(): - pass - - -def test_tokenize_and_postprocess_data(): - # Uses the tokenizer already cached for the rollout smoke test. Skips cleanly - # if the model/tokenizer is not present in the offline HF cache. - transformers = pytest.importorskip("transformers") - try: - tokenizer = transformers.AutoTokenizer.from_pretrained("Qwen/Qwen2.5-0.5B-Instruct") - except Exception as exc: # offline / not cached - pytest.skip(f"tokenizer unavailable: {exc}") - pad_id = tokenizer.pad_token_id if tokenizer.pad_token_id is not None else 0 - input_ids, attention_mask = tf.tokenize_and_postprocess_data( - "hello world from rocm coverage", tokenizer, max_length=16, pad_token_id=pad_id, left_pad=True - ) - assert input_ids.shape[-1] == 16 - assert attention_mask.shape == input_ids.shape diff --git a/tests/workers/rollout/rollout_vllm/test_rollout_utils_rocm.py b/tests/workers/rollout/rollout_vllm/test_rollout_utils_rocm.py deleted file mode 100644 index 2019f4bcd89..00000000000 --- a/tests/workers/rollout/rollout_vllm/test_rollout_utils_rocm.py +++ /dev/null @@ -1,242 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""ROCm code-coverage tests for verl.workers.rollout.vllm_rollout.utils. - -Targets the pure-python helpers that do NOT need a live vLLM worker / ZMQ -transport: CLI-arg building, prompt-logprob extraction, LoRA rank rounding, -the ROCm device-uuid fallback, the compute-logits monkeypatch, and the -signal-suppression context manager. The heavy IPC weight-update path -(update_weights_from_ipc) is exercised separately by the rollout smoke test. - -Importing this module needs vllm present (the rollout utils import from -vllm.outputs), which the rocm/vllm coverage tier provides; the whole module -skips cleanly otherwise. -""" - -import signal -import threading -from types import SimpleNamespace - -import pytest - -pytest.importorskip("vllm") - -from verl.workers.rollout.vllm_rollout.utils import ( # noqa: E402 - build_cli_args_from_config, - extract_prompt_logprobs, - get_device_uuid, - get_vllm_max_lora_rank, - monkey_patch_compute_logits, - SuppressSignalInThread, -) - - -def test_build_cli_args_from_config_all_branches(): - cfg = { - "skip_none": None, # dropped - "flag_true": True, # -> --flag-true - "flag_false": False, # dropped - "empty_list": [], # dropped - "sizes": [1, 2, 4], # expanded - "json_dict": {"a": 1}, # json serialized - "scalar": "hello", # --scalar hello - "number": 7, - } - args = build_cli_args_from_config(cfg) - assert "--skip_none" not in args - assert "--flag_true" in args - assert "--flag_false" not in args - assert "--empty_list" not in args - # list expansion - i = args.index("--sizes") - assert args[i + 1 : i + 4] == ["1", "2", "4"] - # dict -> json - j = args.index("--json_dict") - assert args[j + 1] == '{"a": 1}' - # scalar / number - assert args[args.index("--scalar") + 1] == "hello" - assert args[args.index("--number") + 1] == "7" - - -def test_get_vllm_max_lora_rank_rounds_up(): - # rounds up to the nearest allowed rank - assert get_vllm_max_lora_rank(4) >= 4 - assert get_vllm_max_lora_rank(8) >= 8 - - -def test_get_vllm_max_lora_rank_invalid(): - with pytest.raises(AssertionError): - get_vllm_max_lora_rank(0) - with pytest.raises(ValueError): - get_vllm_max_lora_rank(100000) - - -def _lp(logprob, rank): - return SimpleNamespace(logprob=logprob, rank=rank) - - -def test_extract_prompt_logprobs_none_is_noop(): - result = {} - extract_prompt_logprobs(SimpleNamespace(prompt_logprobs=None), None, result) - assert result == {} - - -def test_extract_prompt_logprobs_zero(): - # num_prompt_logprobs == 0 -> single top token per position - out = SimpleNamespace(prompt_logprobs=[None, {"5": _lp(-0.1, 1)}, {"7": _lp(-0.2, 1)}]) - result = {} - extract_prompt_logprobs(out, 0, result) - assert "prompt_ids" in result and "prompt_logprobs" in result - assert result["prompt_ids"][0] == [5] - - -def test_extract_prompt_logprobs_topk(): - out = SimpleNamespace( - prompt_logprobs=[ - None, - {"5": _lp(-0.1, 1), "9": _lp(-0.5, 2)}, - {"7": _lp(-0.2, 1), "3": _lp(-0.9, 2)}, - ] - ) - result = {} - extract_prompt_logprobs(out, 2, result) - assert result["prompt_ids"][0] == [5, 9] - - -def test_get_device_uuid_returns_string(): - # On ROCm the vLLM platform raises NotImplementedError and the fallback - # derives a stable id from the visible-device env. Assert it produces a - # non-empty identifier rather than a specific value (varies by build). - import os - - prev = os.environ.get("HIP_VISIBLE_DEVICES") - os.environ["HIP_VISIBLE_DEVICES"] = "0,1" - try: - uuid = get_device_uuid(0) - finally: - if prev is None: - os.environ.pop("HIP_VISIBLE_DEVICES", None) - else: - os.environ["HIP_VISIBLE_DEVICES"] = prev - assert isinstance(uuid, str) and len(uuid) > 0 - - -def test_monkey_patch_compute_logits_masks_oov(): - import torch - - class _FakeModel: - def compute_logits(self, *args, **kwargs): - return torch.zeros(1, 8) - - model = _FakeModel() - monkey_patch_compute_logits(model, vocab_size=4) - logits = model.compute_logits() - # tokens >= vocab_size masked to -inf - assert torch.isinf(logits[..., 4:]).all() - assert not torch.isinf(logits[..., :4]).any() - - -def test_extract_prompt_logprobs_topk_skips_out_of_rank(): - # An entry whose rank exceeds num_prompt_logprobs (the sampled token not in - # the top-k) must be skipped via `continue`. - out = SimpleNamespace( - prompt_logprobs=[ - None, - {"5": _lp(-0.1, 1), "9": _lp(-0.5, 2), "2": _lp(-3.0, 3)}, - ] - ) - result = {} - extract_prompt_logprobs(out, 2, result) - # only the two in-rank tokens are kept; the rank-3 entry is dropped - assert result["prompt_ids"][0] == [5, 9] - - -def test_set_death_signal_non_linux_is_noop(monkeypatch): - import platform - - from verl.workers.rollout.vllm_rollout.utils import set_death_signal - - monkeypatch.setattr(platform, "system", lambda: "Darwin") - # Should return immediately without touching libc. - set_death_signal() - - -def _call_set_death_signal_child(): - # Runs in a forked child so the PR_SET_PDEATHSIG / potential self-SIGKILL - # (when ppid == 1) can never take down the pytest process itself. - from verl.workers.rollout.vllm_rollout.utils import set_death_signal - - set_death_signal() - - -def test_set_death_signal_linux_runs(): - import platform - - if platform.system() != "Linux": - pytest.skip("Linux-only path") - import multiprocessing as mp - - # IMPORTANT: never call set_death_signal() in-process. Under the coverage - # container the entrypoint is PID 1, so the test process's ppid is 1 and - # set_death_signal() would os.kill(self, SIGKILL). Isolate it in a child - # (whose ppid is this process, not 1) so the prctl path is covered safely. - ctx = mp.get_context("fork") - p = ctx.Process(target=_call_set_death_signal_child) - p.start() - p.join(timeout=30) - assert p.exitcode == 0 - - -def test_get_device_uuid_torch_props_fallback(): - # With no visible-device env var set, the ROCm fallback drops past the env - # loop into the torch device-properties / final-id path. - import os - - saved = {k: os.environ.pop(k, None) for k in ("HIP_VISIBLE_DEVICES", "ROCR_VISIBLE_DEVICES", "CUDA_VISIBLE_DEVICES")} - try: - uuid = get_device_uuid(0) - assert isinstance(uuid, str) and uuid.startswith("ROCm-") - finally: - for k, v in saved.items(): - if v is not None: - os.environ[k] = v - - -def test_suppress_signal_in_thread(): - with SuppressSignalInThread(): - # registering a signal handler from a non-main thread is a no-op - done = {} - - def worker(): - try: - signal.signal(signal.SIGUSR1, signal.SIG_IGN) - done["ok"] = True - except Exception as exc: # pragma: no cover - done["err"] = exc - - t = threading.Thread(target=worker) - t.start() - t.join() - assert done.get("ok") is True - # signal.signal is restored after the context: registering from the main - # thread works normally again. - prev = signal.getsignal(signal.SIGUSR1) - signal.signal(signal.SIGUSR1, signal.SIG_IGN) - signal.signal(signal.SIGUSR1, prev) - - -if __name__ == "__main__": - import sys - - sys.exit(pytest.main([__file__, "-v"])) diff --git a/tests/workers/rollout/rollout_vllm/test_vllm_smoke_rocm.py b/tests/workers/rollout/rollout_vllm/test_vllm_smoke_rocm.py deleted file mode 100644 index fbd3caf3224..00000000000 --- a/tests/workers/rollout/rollout_vllm/test_vllm_smoke_rocm.py +++ /dev/null @@ -1,189 +0,0 @@ -# Copyright 2026 AMD -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. -"""ROCm vLLM rollout smoke test (code-coverage tier). - -Single-GPU, tiny-model (Qwen2.5-0.5B-Instruct), single short generation through -verl's standalone vLLM rollout server. The point is COVERAGE of the real ROCm -rollout path (vllm_async_server / vllm_rollout adapter / rollout utils), not a -correctness/perf assertion -- so it only checks that generation produced tokens. - -Skips cleanly (rather than failing) when the prerequisites for a meaningful run -are absent (no GPU, vllm missing, or the model cannot be fetched), so it is safe -in the curated coverage set. -""" - -import asyncio -import os - -import pytest - -MODEL_ID = os.environ.get("VERL_SMOKE_MODEL", "Qwen/Qwen2.5-0.5B-Instruct") - - -def _require(cond, reason): - if not cond: - pytest.skip(reason) - - -def test_vllm_rollout_smoke_rocm(): - # ---- Preconditions (skip, don't fail, if the tier can't run this) -------- - try: - import torch - except Exception: - pytest.skip("torch not importable") - _require(torch.cuda.is_available(), "no GPU visible") - _require(pytest.importorskip("vllm") is not None, "vllm not installed") - - # ---- Fetch the tiny model (local snapshot) ------------------------------- - try: - from huggingface_hub import snapshot_download - - model_path = snapshot_download(MODEL_ID) - except Exception as exc: # network/gated/etc. - pytest.skip(f"could not fetch {MODEL_ID}: {exc}") - - import ray - - # The vLLM rollout manages device visibility itself, so its ray workers (incl. - # the colocated CheckpointEngineWorker) require RAY_EXPERIMENTAL_NOSET_*_VISIBLE_DEVICES - # to be set; otherwise ray rewrites *_VISIBLE_DEVICES under the worker and the - # rollout worker init hits an unimplemented device path. A coverage harness that - # also runs CPU-only single_controller tests may clear these flags globally, so - # re-assert them for THIS rollout via the cluster runtime_env. - rollout_env = { - "VLLM_USE_V1": "1", - "TOKENIZERS_PARALLELISM": "false", - "RAY_EXPERIMENTAL_NOSET_HIP_VISIBLE_DEVICES": "1", - "RAY_EXPERIMENTAL_NOSET_ROCR_VISIBLE_DEVICES": "1", - "RAY_EXPERIMENTAL_NOSET_CUDA_VISIBLE_DEVICES": "1", - } - # Start from a clean cluster: when this test runs after other ray-using tests - # in the same session, a stale cluster would be reused and the runtime_env - # below silently ignored, breaking rollout worker init. - if ray.is_initialized(): - ray.shutdown() - ray.init( - runtime_env={"env_vars": rollout_env}, - ignore_reinit_error=True, - ) - try: - from hydra import compose, initialize_config_dir - - from verl.utils.tokenizer import normalize_token_ids - - config_dir = os.path.abspath("verl/trainer/config") - if not os.path.exists(config_dir): - config_dir = os.path.abspath("verl/verl/trainer/config") - with initialize_config_dir(config_dir=config_dir, version_base=None): - config = compose(config_name="ppo_trainer") - - config.trainer.n_gpus_per_node = 1 - config.trainer.nnodes = 1 - config.actor_rollout_ref.model.path = model_path - config.actor_rollout_ref.rollout.name = "vllm" - config.actor_rollout_ref.rollout.mode = "async" - config.actor_rollout_ref.rollout.tensor_model_parallel_size = 1 - config.actor_rollout_ref.rollout.prompt_length = 128 - config.actor_rollout_ref.rollout.response_length = 32 - # Keep it light: skip CUDA-graph capture, modest mem. - for k, v in (("enforce_eager", True), ("gpu_memory_utilization", 0.5), ("free_cache_engine", True)): - if k in config.actor_rollout_ref.rollout: - config.actor_rollout_ref.rollout[k] = v - - from verl.workers.rollout.replica import get_rollout_replica_class - - rollout_server_class = get_rollout_replica_class("vllm") - server = rollout_server_class( - replica_rank=0, - config=config.actor_rollout_ref.rollout, - model_config=config.actor_rollout_ref.model, - gpus_per_node=1, - ) - asyncio.run(server.init_standalone()) - server_handle = server._server_handle - - from transformers import AutoTokenizer - - tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True) - prompt_ids = normalize_token_ids( - tokenizer.apply_chat_template( - [{"role": "user", "content": "Say hello in one short sentence."}], - add_generation_prompt=True, - tokenize=True, - ) - ) - - output = ray.get( - server_handle.generate.remote( - request_id="smoke_0", - prompt_ids=prompt_ids, - sampling_params={"temperature": 0.0, "top_p": 1.0, "logprobs": False}, - image_data=None, - ), - timeout=240.0, - ) - assert output is not None and len(output.token_ids) > 0, "rollout produced no tokens" - print("SMOKE OK:", tokenizer.decode(output.token_ids)[:120]) - - # ---- Extra coverage of the async server lifecycle --------------------- - # These exercise more of vllm_async_server (sampling variants, KV-cache - # reset, abort/resume, sleep/wake offload, profiler hooks) on the real - # ROCm stack. They are best-effort: the smoke gate is the generation - # above, so any op that a given vllm/ROCm build does not support is - # logged and skipped rather than failing the curated coverage set. - def _aux(label, fn): - try: - fn() - print(f"SMOKE {label}: ok") - except Exception as exc: # noqa: BLE001 - coverage, not correctness - print(f"SMOKE {label}: non-fatal {type(exc).__name__}: {exc}") - - # (1) A second generation with a different sampling profile (stochastic, - # top-k/top-p, logprobs on) to cover the non-greedy params path. - def _gen2(): - out = ray.get( - server_handle.generate.remote( - request_id="smoke_1", - prompt_ids=prompt_ids, - sampling_params={ - "temperature": 0.7, - "top_p": 0.95, - "top_k": 50, - "logprobs": True, - "max_tokens": 16, - }, - image_data=None, - ), - timeout=240.0, - ) - assert out is not None - - _aux("gen2", _gen2) - # (2) KV-cache reset. - _aux("clear_kv_cache", lambda: asyncio.run(server.clear_kv_cache())) - # (3) Partial-rollout abort then resume. - _aux("abort_all_requests", lambda: asyncio.run(server.abort_all_requests())) - _aux("resume_generation", lambda: asyncio.run(server.resume_generation())) - # (4) Sleep (offload weights/KV to host) then wake. - _aux("sleep", lambda: asyncio.run(server.sleep())) - _aux("wake_up", lambda: asyncio.run(server.wake_up())) - # (5) Profiler start/stop hooks. - _aux("start_profile", lambda: asyncio.run(server.start_profile())) - _aux("stop_profile", lambda: asyncio.run(server.stop_profile())) - finally: - ray.shutdown() - - -if __name__ == "__main__": - test_vllm_rollout_smoke_rocm()