From e06d53a172eefb9a5b756d94351fdb85455346d2 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Sat, 23 May 2026 18:10:15 +0200 Subject: [PATCH 1/5] tests/dragon_ci: standalone pre-release contract suite for Dragon MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pin every Dragon API surface our V3 backend and telemetry adapter depend on, so the Dragon team can detect contract changes before they ship. The suite has no Rhapsody dependency — only dragon, pytest, pytest-asyncio, pytest-timeout, cloudpickle, and the stdlib. Coverage: - Batch (ctor, num_workers/num_managers, function/process/job, fence, close/join/terminate, results_ddict 5-tuple shape on success/failure) - ProcessTemplate (cwd/env/policy/args/kwargs/argdata) - Policy (Distribution, Placement, gpu_affinity) - System (nnodes, hostname_policies) - ProcessGroup, native Queue/Event, DragonUserCodeError - DDict core ops - dragon.telemetry.collector + AccVendor - async-coroutine wrapping via asyncio.run shim - worker failure-mode detection boundary Three known Dragon bugs are pinned as deliberately-failing @pytest.mark.timeout tests, one per file (the leaked test thread prevents combining them): - test_ddict_unknown_key_blocks: DDict[unknown_key] blocks instead of raising KeyError when wait_for_keys=True - test_sigkill_in_worker_hangs: abnormal worker termination is undetected; task.get() blocks forever; timeout= kwarg is ignored on the blocking DDict read - test_unpickleable_function_hangs: same silent-hang failure mode for functions from modules unreachable via the worker's PYTHONPATH Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/dragon_ci/README.md | 89 ++++++++++ tests/dragon_ci/_dragon_ci_helpers.py | 77 +++++++++ tests/dragon_ci/_offpath/marker_helper.py | 7 + tests/dragon_ci/conftest.py | 159 ++++++++++++++++++ tests/dragon_ci/pytest.ini | 14 ++ tests/dragon_ci/test_async_in_batch.py | 38 +++++ tests/dragon_ci/test_batch_function.py | 85 ++++++++++ tests/dragon_ci/test_batch_job.py | 53 ++++++ tests/dragon_ci/test_batch_lifecycle.py | 62 +++++++ tests/dragon_ci/test_batch_process.py | 84 +++++++++ tests/dragon_ci/test_ddict.py | 90 ++++++++++ .../test_ddict_unknown_key_blocks.py | 39 +++++ tests/dragon_ci/test_machine_system.py | 46 +++++ tests/dragon_ci/test_policy.py | 57 +++++++ tests/dragon_ci/test_process_group.py | 69 ++++++++ tests/dragon_ci/test_process_template.py | 80 +++++++++ tests/dragon_ci/test_queue_event.py | 74 ++++++++ .../dragon_ci/test_sigkill_in_worker_hangs.py | 27 +++ tests/dragon_ci/test_telemetry_collector.py | 63 +++++++ .../test_unpickleable_function_hangs.py | 42 +++++ tests/dragon_ci/test_worker_failure_modes.py | 44 +++++ 21 files changed, 1299 insertions(+) create mode 100644 tests/dragon_ci/README.md create mode 100644 tests/dragon_ci/_dragon_ci_helpers.py create mode 100644 tests/dragon_ci/_offpath/marker_helper.py create mode 100644 tests/dragon_ci/conftest.py create mode 100644 tests/dragon_ci/pytest.ini create mode 100644 tests/dragon_ci/test_async_in_batch.py create mode 100644 tests/dragon_ci/test_batch_function.py create mode 100644 tests/dragon_ci/test_batch_job.py create mode 100644 tests/dragon_ci/test_batch_lifecycle.py create mode 100644 tests/dragon_ci/test_batch_process.py create mode 100644 tests/dragon_ci/test_ddict.py create mode 100644 tests/dragon_ci/test_ddict_unknown_key_blocks.py create mode 100644 tests/dragon_ci/test_machine_system.py create mode 100644 tests/dragon_ci/test_policy.py create mode 100644 tests/dragon_ci/test_process_group.py create mode 100644 tests/dragon_ci/test_process_template.py create mode 100644 tests/dragon_ci/test_queue_event.py create mode 100644 tests/dragon_ci/test_sigkill_in_worker_hangs.py create mode 100644 tests/dragon_ci/test_telemetry_collector.py create mode 100644 tests/dragon_ci/test_unpickleable_function_hangs.py create mode 100644 tests/dragon_ci/test_worker_failure_modes.py diff --git a/tests/dragon_ci/README.md b/tests/dragon_ci/README.md new file mode 100644 index 0000000..47ca491 --- /dev/null +++ b/tests/dragon_ci/README.md @@ -0,0 +1,89 @@ +# Dragon CI tests + +Standalone pre-release contract tests for the Dragon runtime. The suite has +**no dependency on Rhapsody** — only `dragon`, `pytest`, `pytest-asyncio`, +`pytest-timeout`, `cloudpickle`, and the standard library. Its purpose is to +fail loudly when a Dragon release breaks any API surface that Rhapsody (or +similar downstream stacks built on the V3 `Batch` workflow) depends on. + +## What's covered + +Each test pins one observable contract: symbol name, callable signature, +attribute presence, or a short behavioral round-trip. Failures should point +the Dragon team at the specific contract that changed. + +**Happy-path contract files** (expected to pass): + +| File | Subsystem | +|---------------------------------|------------------------------------------------------| +| `test_batch_lifecycle.py` | `Batch` ctor kwargs, attrs, close/join/terminate | +| `test_batch_function.py` | `Batch.function()` + `results_ddict` 5-tuple shape | +| `test_batch_process.py` | `Batch.process()` + stdio-capture pitfalls | +| `test_batch_job.py` | `Batch.job()` (wire + opt-in real MPI launch) | +| `test_process_template.py` | `ProcessTemplate` ctor, attrs, `Popen` consts | +| `test_policy.py` | `Policy` ctor, `Distribution`/`Placement` enums | +| `test_machine_system.py` | `System().nnodes`, `hostname_policies` | +| `test_process_group.py` | native `ProcessGroup` lifecycle | +| `test_queue_event.py` | `native.queue.Queue` + `native.event.Event` | +| `test_ddict.py` | `DDict` core ops | +| `test_telemetry_collector.py` | `dragon.telemetry.collector` + `AccVendor` | +| `test_async_in_batch.py` | async-coroutine wrapping pattern (V3) | +| `test_worker_failure_modes.py` | caught-exception and `sys.exit` reporting paths | + +**Known-Dragon-bug pins** (expected to FAIL via pytest-timeout until Dragon fixes them): + +| File | Bug pinned | +|--------------------------------------------|-----------------------------------------------------------| +| `test_ddict_unknown_key_blocks.py` | `DDict[unknown_key]` blocks instead of raising `KeyError` when `wait_for_keys=True` | +| `test_sigkill_in_worker_hangs.py` | SIGKILL'd worker is undetected; `task.get()` blocks forever; `timeout=` kwarg is ignored on the blocking DDict read | +| `test_unpickleable_function_hangs.py` | Worker dies during cloudpickle unpickling without diagnostic — same silent-hang as SIGKILL | + +These three live in their own files because `pytest-timeout`'s `method="thread"` +leaks the wedged test thread, which holds Dragon resources and breaks any +subsequent test in the same pytest invocation. One bug per file keeps things +clean. + +## Running + +The suite must run under the Dragon launcher because Dragon's multiprocessing +backend is required even for in-process tests. + +``` +dragon python -m pytest -c tests/dragon_ci/pytest.ini \ + --rootdir=tests/dragon_ci tests/dragon_ci/ -v +``` + +The `-c` and `--rootdir` flags keep pytest from picking up the parent +`tests/conftest.py` (which imports Rhapsody). + +## Skip markers + +Tests auto-skip what the environment can't support: + +| Marker | Active when | +|---------------------------------|--------------------------------------------------------| +| `requires_multi_node` | `System().nnodes >= 2` | +| `requires_gpu` | `identify_gpu()` reports at least one device | +| `requires_mpi` | `DRAGON_CI_HAS_PMI=1` is set (opt-in) | +| `requires_telemetry_collector` | `dragon.telemetry.collector` is importable | + +MPI launches are opt-in because a failing PMIx init aborts the entire Dragon +runtime — not just the calling test — and so cannot be probed safely. + +## Adding tests + +- One contract per test. Prefer parametrized signature/attribute checks for + enumerations of kwargs or methods. +- For callables passed to `Batch.function()` / `ProcessTemplate(target=…)`, + define the function in `_dragon_ci_helpers.py`, **not** in the test file. + Top-level callables defined in `test_*.py` files cannot be re-imported by + Dragon workers (pytest module names aren't on `PYTHONPATH`); the worker + fails silently to unpickle and `task.get()` blocks forever. The conftest + injects this directory into `PYTHONPATH` so the helpers module IS visible + to workers. +- Use the `batch` session fixture when possible. Use `fresh_batch` only for + tests that intentionally kill workers — those leave the Batch unrecoverable. +- New "documents a Dragon bug" tests should each live in their own file, + use `@pytest.mark.timeout(N, method="thread")`, and contain only the one + test (see the existing `*_hangs.py` / `*_blocks.py` files as templates). +- Do **not** import `rhapsody` from anywhere in this directory. diff --git a/tests/dragon_ci/_dragon_ci_helpers.py b/tests/dragon_ci/_dragon_ci_helpers.py new file mode 100644 index 0000000..98c97e1 --- /dev/null +++ b/tests/dragon_ci/_dragon_ci_helpers.py @@ -0,0 +1,77 @@ +"""Top-level helpers used by tests. + +Must live in a regular importable module (not in a ``test_*.py`` file) so +Dragon's workers can re-import it via cloudpickle's by-reference path. See +the conftest module docstring for the full story. +""" + +import asyncio +import os +import signal +import sys +import time + + +# --- Batch.function targets ---------------------------------------------- + + +def add(a, b): + return a + b + + +def kwfn(x, *, mult=1): + return x * mult + + +def slow_double(x, delay): + time.sleep(delay) + return x * 2 + + +def print_and_return(value, msg): + print(msg) + print("err msg", file=sys.stderr) + return value + + +def raise_value_error(message): + raise ValueError(message) + + +# --- async dispatch (test_async_in_batch.py) ----------------------------- + + +async def async_double(x): + await asyncio.sleep(0) + return x * 2 + + +def async_run_shim(*args, **kwargs): + """Mirror Rhapsody's wrapper in dragon.py V3.""" + return asyncio.run(async_double(*args, **kwargs)) + + +# --- ProcessGroup workers (test_process_group.py, test_queue_event.py) --- + + +def pg_worker(queue, shutdown_event): + """Push one identity message, then loop until shutdown_event is set.""" + queue.put({"pid": os.getpid(), "host": os.uname().nodename}, timeout=10.0) + while not shutdown_event.wait(timeout=0.05): + pass + + +# --- worker failure-mode probes (test_worker_failure_modes.py) ----------- + + +def fn_assert_false(_x): + assert False, "intentional dragon_ci probe" + + +def fn_sys_exit_one(_x): + sys.exit(1) + + +def fn_kill_self(_x): + """Abnormal worker termination — no chance for Dragon's exit-detector.""" + os.kill(os.getpid(), signal.SIGKILL) diff --git a/tests/dragon_ci/_offpath/marker_helper.py b/tests/dragon_ci/_offpath/marker_helper.py new file mode 100644 index 0000000..c81a0c0 --- /dev/null +++ b/tests/dragon_ci/_offpath/marker_helper.py @@ -0,0 +1,7 @@ +"""Helper module deliberately placed in a subdirectory that the conftest +does NOT add to ``PYTHONPATH``. Tests that want to reproduce the runtime +``sys.path.insert`` regression import from here. +""" + +def offpath_add(a, b): + return a + b diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py new file mode 100644 index 0000000..52b8244 --- /dev/null +++ b/tests/dragon_ci/conftest.py @@ -0,0 +1,159 @@ +"""Pytest configuration for the Dragon CI contract suite. + +Run with:: + + dragon python -m pytest -c tests/dragon_ci/pytest.ini \ + --rootdir=tests/dragon_ci tests/dragon_ci/ + +The ``-c`` / ``--rootdir`` flags keep pytest from picking up the parent +``tests/conftest.py`` (which imports Rhapsody). + +Worker-importability note +------------------------- +cloudpickle pickles top-level functions **by reference** (just the +``module.qualname`` pair). Dragon workers re-import the module by that +name, and their ``sys.path`` is NOT inherited from the parent — only the +``PYTHONPATH`` env var is. So a helper that the parent can import only +because of a runtime ``sys.path.insert`` will fail in the worker, the +result is never written to the DDict, and ``task.get()`` blocks forever +(see ``test_worker_failure_modes.py``). + +To avoid the trap, this conftest puts the suite directory on both +``sys.path`` (for the parent) and ``PYTHONPATH`` (for workers). All +callables passed to Batch tasks must live in ``_dragon_ci_helpers``, +not in a ``test_*.py`` file. +""" + +from __future__ import annotations + +import os +import sys + +import pytest +import pytest_asyncio + + +# --- Make _dragon_ci_helpers importable from workers ---------------------- + +_HERE = os.path.dirname(os.path.abspath(__file__)) +if _HERE not in sys.path: + sys.path.insert(0, _HERE) +os.environ["PYTHONPATH"] = _HERE + os.pathsep + os.environ.get("PYTHONPATH", "") + + +# --- Dragon must be importable ------------------------------------------- + +pytest.importorskip("dragon", reason="run under `dragon python`") + + +# --- Environment detection (single pass at import time) ------------------ + + +def _safe(fn, default): + try: + return fn() + except Exception: + return default + + +def _detect_nnodes() -> int: + from dragon.native.machine import System + + return int(System().nnodes) + + +def _detect_ngpus() -> int: + from dragon.telemetry.collector import identify_gpu + + _vendor, count = identify_gpu() + # AMD's identify_gpu returns a list; normalize to length. + return len(count) if isinstance(count, list) else int(count or 0) + + +def _has_telemetry_collector() -> bool: + import dragon.telemetry.collector # noqa: F401 + + return True + + +_NNODES = _safe(_detect_nnodes, 0) +_NGPUS = _safe(_detect_ngpus, 0) +_HAS_TELEM = _safe(_has_telemetry_collector, False) +# PMI/PMIx launch failures abort the entire Dragon runtime, so we cannot +# probe at runtime. Gate on explicit opt-in instead. +_HAS_PMI = os.environ.get("DRAGON_CI_HAS_PMI", "").lower() not in ("", "0", "false") + + +# --- Auto-skip markers ---------------------------------------------------- + +# (marker_name, predicate-that-must-be-true-to-run, reason) +_SKIP_RULES = ( + ("requires_multi_node", _NNODES >= 2, f"requires >= 2 Dragon nodes (have {_NNODES})"), + ("requires_gpu", _NGPUS >= 1, "requires at least one GPU"), + ("requires_mpi", _HAS_PMI, "set DRAGON_CI_HAS_PMI=1 to enable MPI/PMIx tests"), + ("requires_telemetry_collector", _HAS_TELEM, "dragon.telemetry.collector not importable"), +) + + +def pytest_collection_modifyitems(config, items): + for marker, ok, reason in _SKIP_RULES: + if ok: + continue + skip = pytest.mark.skip(reason=reason) + for item in items: + if marker in item.keywords: + item.add_marker(skip) + + +# --- Batch fixtures ------------------------------------------------------- + + +def _new_batch(): + """Construct a Batch with Dragon's required mp start method set.""" + import multiprocessing as mp + + from dragon.workflows.batch import Batch + + try: + if mp.get_start_method(allow_none=True) != "dragon": + mp.set_start_method("dragon", force=True) + except RuntimeError: + pass + return Batch(disable_telem=True) + + +@pytest_asyncio.fixture(scope="session", loop_scope="session") +async def batch(): + """One Batch shared across the session. + + Async-fixture so construction happens inside an asyncio loop, mirroring + Rhapsody's own V3 tests. Tests must NOT kill workers or call ``close()`` + on this Batch — use ``fresh_batch`` for that. + """ + b = _new_batch() + try: + yield b + finally: + try: + b.join(timeout=30.0) + except Exception: + pass + + +@pytest_asyncio.fixture(loop_scope="session") +async def fresh_batch(): + """A throwaway Batch for tests that intentionally kill workers — the + session ``batch`` cannot recover once a worker is SIGKILL'd. + + Skips the graceful ``join()`` because dead workers make it block for + the full timeout; ``terminate()`` can also raise ``DragonUserCodeError`` + on dead workers — both are swallowed. + """ + b = _new_batch() + try: + yield b + finally: + try: + b.terminate() + except Exception: + pass diff --git a/tests/dragon_ci/pytest.ini b/tests/dragon_ci/pytest.ini new file mode 100644 index 0000000..4656f0a --- /dev/null +++ b/tests/dragon_ci/pytest.ini @@ -0,0 +1,14 @@ +[pytest] +testpaths = . +python_files = test_*.py +python_classes = Test* +python_functions = test_* +addopts = --tb=short -ra +asyncio_mode = auto +asyncio_default_fixture_loop_scope = session +asyncio_default_test_loop_scope = session +markers = + requires_multi_node: needs an allocation with >= 2 Dragon nodes + requires_gpu: needs at least one GPU visible to Dragon's telemetry collector + requires_mpi: needs DRAGON_CI_HAS_PMI=1 (real MPI launch) + requires_telemetry_collector: needs dragon.telemetry.collector importable diff --git a/tests/dragon_ci/test_async_in_batch.py b/tests/dragon_ci/test_async_in_batch.py new file mode 100644 index 0000000..2f5a1f3 --- /dev/null +++ b/tests/dragon_ci/test_async_in_batch.py @@ -0,0 +1,38 @@ +"""Contract: async-coroutine target dispatch via ``Batch.function``. + +Rhapsody V3's ``build_task`` detects ``asyncio.iscoroutinefunction(target)`` +and wraps the coroutine in a ``asyncio.run(...)`` shim before passing it to +``batch.function``. This test pins that pattern, and also verifies the +underlying primitives Rhapsody depends on: + +- ``asyncio.iscoroutinefunction`` recognises the user's function, +- ``Batch.function`` accepts a plain (non-coroutine) callable, +- a coroutine result reaches ``batch.results_ddict[uid]`` unchanged after + the wrapper applies ``asyncio.run``. + +Helpers live in ``_dragon_ci_helpers`` — see the conftest module docstring +for why callables passed to Batch tasks cannot be defined in this file. +""" + +from __future__ import annotations + +import asyncio + +from dragon.workflows.batch import Batch + +from _dragon_ci_helpers import async_double, async_run_shim # noqa: E402 + + +def test_iscoroutinefunction_detects_async_def(): + assert asyncio.iscoroutinefunction(async_double) + assert not asyncio.iscoroutinefunction(async_run_shim) + + +def test_async_result_reaches_results_ddict(batch: Batch): + """The shimmed coroutine result must surface as a plain int in the DDict.""" + task = batch.function(async_run_shim, 21) + assert task.get(timeout=60.0) == 42 + + result, _tb, raised, _stdout, _stderr = batch.results_ddict[task.uid] + assert raised is False + assert result == 42 diff --git a/tests/dragon_ci/test_batch_function.py b/tests/dragon_ci/test_batch_function.py new file mode 100644 index 0000000..d217821 --- /dev/null +++ b/tests/dragon_ci/test_batch_function.py @@ -0,0 +1,85 @@ +"""Contract: ``Batch.function()`` mode and the results-DDict tuple shape. + +Rhapsody's V3 backend reads results directly from the Batch-owned DDict:: + + result, tb, raised, stdout, stderr = batch.results_ddict[task.uid] + +so the tuple layout, the truthiness of ``raised``, and the types of +``stdout``/``stderr`` are all part of the contract. ``Function.get()`` is the +documented public alternative and is also pinned. + +Helpers live in ``_dragon_ci_helpers`` — see the conftest module docstring +for why callables passed to Batch tasks cannot be defined in this file. +""" + +from __future__ import annotations + +import pytest + +from dragon.workflows.batch import Batch +from dragon.workflows.batch.batch import Function, TaskNotReadyError + +from _dragon_ci_helpers import ( # noqa: E402 (must be importable from workers) + add, kwfn, print_and_return, raise_value_error, slow_double, +) + + +def test_function_returns_function_handle(batch: Batch): + """``batch.function()`` returns a ``Function`` with ``.uid`` and ``.get``.""" + task = batch.function(add, 2, 3) + assert isinstance(task, Function) + assert task.uid and callable(task.get) + assert task.get(timeout=60.0) == 5 + + +def test_function_get_reraises_exception(batch: Batch): + task = batch.function(raise_value_error, "intentional failure") + with pytest.raises(ValueError, match="intentional failure"): + task.get(timeout=60.0) + + +def test_function_supports_kwargs(batch: Batch): + """``Batch.function`` must forward ``**kwargs`` to the target.""" + assert batch.function(kwfn, 6, mult=7).get(timeout=60.0) == 42 + + +def test_function_get_non_blocking_raises_when_not_ready(batch: Batch): + """``Function.get(block=False)`` raises ``TaskNotReadyError`` pre-completion.""" + task = batch.function(slow_double, 21, 1.0) + with pytest.raises(TaskNotReadyError): + task.get(block=False) + assert task.get(timeout=60.0) == 42 + + +def test_results_ddict_five_tuple_on_success(batch: Batch): + """Rhapsody unpacks exactly five fields. Order and shape are the contract.""" + task = batch.function(add, 1, 1) + task.get(timeout=60.0) + entry = batch.results_ddict[task.uid] + assert isinstance(entry, tuple) and len(entry) == 5, ( + f"results_ddict tuple shape changed: {entry!r}" + ) + result, tb, raised, stdout, stderr = entry + assert result == 2 + assert tb is None and raised is False + assert isinstance(stdout, str) and isinstance(stderr, str) + + +def test_results_ddict_five_tuple_on_failure(batch: Batch): + """On exception: ``raised=True``, ``result`` holds the exception, ``tb`` is str.""" + task = batch.function(raise_value_error, "boom") + with pytest.raises(ValueError): + task.get(timeout=60.0) + result, tb, raised, stdout, stderr = batch.results_ddict[task.uid] + assert raised is True + assert isinstance(result, BaseException) + assert isinstance(tb, str) and "ValueError" in tb + assert isinstance(stdout, str) and isinstance(stderr, str) + + +def test_function_captures_stdout_and_stderr(batch: Batch): + task = batch.function(print_and_return, 7, "hello-from-function") + assert task.get(timeout=60.0) == 7 + _r, _tb, _raised, stdout, stderr = batch.results_ddict[task.uid] + assert "hello-from-function" in stdout + assert "err msg" in stderr diff --git a/tests/dragon_ci/test_batch_job.py b/tests/dragon_ci/test_batch_job.py new file mode 100644 index 0000000..02c698a --- /dev/null +++ b/tests/dragon_ci/test_batch_job.py @@ -0,0 +1,53 @@ +"""Contract: ``Batch.job()`` — multi-rank / MPI launch mode. + +Rhapsody V3 builds jobs as a list of ``(nranks, ProcessTemplate)`` tuples. +Wire-only checks always run; actual launches are guarded by ``requires_mpi`` +(opt-in via ``DRAGON_CI_HAS_PMI=1``) because a failing PMIx init aborts +the whole Dragon runtime, not just the test. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.infrastructure.facts import PMIBackend +from dragon.native.process import ProcessTemplate +from dragon.workflows.batch import Batch +from dragon.workflows.batch.batch import Job + + +@pytest.mark.parametrize("kwarg", ["process_templates", "name", "timeout", "pmi"]) +def test_batch_job_kwarg(kwarg): + assert kwarg in inspect.signature(Batch.job).parameters, ( + f"Batch.job({kwarg}=) removed — Rhapsody depends on it" + ) + + +def test_pmi_backend_has_pmix(): + """Rhapsody passes ``PMIBackend.PMIX`` as the cross-vendor portable backend.""" + assert "PMIX" in {m.name for m in PMIBackend} + + +@pytest.mark.requires_mpi +@pytest.mark.parametrize("nranks", [1, 2]) +def test_batch_job_launch(batch: Batch, nranks): + """A real PMIx launch must produce a ``Job`` handle and complete cleanly. + + Skips on hosts where PMIx is unavailable (documented failure mode is + ``RuntimeError: Unable to initialize PMIx server`` from + ``DragonPMIxJob.__init__``). + """ + try: + job = batch.job( + [(nranks, ProcessTemplate("/bin/true", args=()))], + name=f"dragon-ci-{nranks}rank", pmi=PMIBackend.PMIX, + ) + job.get(timeout=120.0) + except Exception as exc: # noqa: BLE001 + pytest.skip(f"PMIX launch unavailable on this host: {exc!r}") + + assert isinstance(job, Job) + result, tb, raised, _stdout, _stderr = batch.results_ddict[job.uid] + assert raised is False, f"{nranks}-rank job failed: result={result!r} tb={tb!r}" diff --git a/tests/dragon_ci/test_batch_lifecycle.py b/tests/dragon_ci/test_batch_lifecycle.py new file mode 100644 index 0000000..467fb50 --- /dev/null +++ b/tests/dragon_ci/test_batch_lifecycle.py @@ -0,0 +1,62 @@ +"""Contract: ``dragon.workflows.batch.Batch`` construction, attributes, teardown. + +Rhapsody's V3 backend constructs ``Batch`` with the kwargs below, reads +``num_workers`` / ``num_managers`` for logging, and shuts down via +``close()`` then ``join(timeout=...)`` with ``terminate()`` as a fallback. +``fence()`` is exposed to user code as a barrier. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.workflows.batch import Batch + + +_BATCH_CTOR_KWARGS = ("num_nodes", "pool_nodes", "disable_telem", + "scheduler_workers", "results_ddict_mem") +_BATCH_METHODS = ("function", "process", "job", "fence", "close", "join", "terminate") + + +@pytest.mark.parametrize("kwarg", _BATCH_CTOR_KWARGS) +def test_batch_ctor_kwarg(kwarg): + assert kwarg in inspect.signature(Batch.__init__).parameters, ( + f"Batch.__init__ no longer accepts {kwarg!r}" + ) + + +@pytest.mark.parametrize("method", _BATCH_METHODS) +def test_batch_method_present(method): + assert callable(getattr(Batch, method, None)), f"Batch.{method}() removed" + + +def test_batch_num_workers_and_num_managers(batch): + """Rhapsody logs ``batch.num_workers`` and ``batch.num_managers`` at startup.""" + assert isinstance(batch.num_workers, int) and batch.num_workers >= 1 + assert isinstance(batch.num_managers, int) and batch.num_managers >= 1 + + +def test_batch_results_ddict_supports_lookup(batch): + """Rhapsody reads ``batch.results_ddict[task.uid]`` directly.""" + rd = batch.results_ddict + assert hasattr(rd, "__contains__") and hasattr(rd, "__getitem__") + + +def test_batch_fence_callable_on_idle_batch(batch): + """Rhapsody exposes ``Batch.fence()`` to user code as a barrier. On an + idle Batch it must complete without raising.""" + batch.fence() + + +def test_batch_close_join_terminate_lifecycle(): + """A throwaway Batch must accept close()+join()+terminate() without raising. + + Combined into one test so we only pay one Batch startup. Note ``close()`` + is deprecated as of Dragon 0.14 (no-op); only ``join()`` is real teardown. + """ + b = Batch(disable_telem=True) + b.close() + b.join(timeout=30.0) + b.terminate() diff --git a/tests/dragon_ci/test_batch_process.py b/tests/dragon_ci/test_batch_process.py new file mode 100644 index 0000000..58f237b --- /dev/null +++ b/tests/dragon_ci/test_batch_process.py @@ -0,0 +1,84 @@ +"""Contract: ``Batch.process()`` mode and ``ProcessTemplate`` plumbing. + +Rhapsody V3 launches subprocess tasks via:: + + batch.process(ProcessTemplate(target, args=(...), cwd=..., policy=...)) + +Key contract pinned by these tests: **a bare ``ProcessTemplate`` (no +``stdout=Popen.PIPE``) does NOT capture child stdout/stderr into the +``results_ddict`` 5-tuple — those slots come back as empty strings.** +Rhapsody works around this by wrapping the command in a shell script that +redirects to files (see the V3 ``capture_stdio`` code path). +""" + +from __future__ import annotations + +import pytest + +from dragon.native.process import ProcessTemplate +from dragon.workflows.batch import Batch + + +def test_process_returns_five_tuple(batch: Batch): + """A trivial ``/bin/echo`` task completes and yields the 5-tuple shape.""" + task = batch.process(ProcessTemplate("/bin/echo", args=("hello-process",))) + entry = batch.results_ddict[task.uid] + assert isinstance(entry, tuple) and len(entry) == 5, ( + f"results_ddict shape changed for batch.process: {entry!r}" + ) + _result, _tb, raised, _stdout, _stderr = entry + assert raised is False + + +def test_process_default_stdio_is_empty(batch: Batch): + """Without ``stdout=Popen.PIPE``, the stdout/stderr slots are empty strings. + + If Dragon ever starts capturing by default, Rhapsody's ``capture_stdio`` + workaround can be retired — this test will fail then. + """ + task = batch.process(ProcessTemplate("/bin/echo", args=("hello-process",))) + _r, _tb, _raised, stdout, stderr = batch.results_ddict[task.uid] + assert stdout == "" and stderr == "", ( + f"Dragon may now capture stdio by default: stdout={stdout!r} stderr={stderr!r}" + ) + + +def test_process_template_args_none_raises_typeerror(batch: Batch): + """``ProcessTemplate(...).args`` defaults to None, which makes + ``Batch.process`` raise ``TypeError: 'NoneType' object is not iterable``. + Pass ``args=()`` to avoid it.""" + with pytest.raises(TypeError, match="NoneType"): + batch.process(ProcessTemplate("/bin/true")) + + +def test_process_non_zero_exit_is_visible(batch: Batch): + """``/bin/false`` (exit 1) must surface failure through the 5-tuple.""" + task = batch.process(ProcessTemplate("/bin/false", args=())) + result, tb, raised, _stdout, _stderr = batch.results_ddict[task.uid] + # Dragon may store the failure as ``raised=True`` or as a non-zero result. + failure_visible = raised or bool(tb) or result not in (None, 0, True) + assert failure_visible, ( + "Non-zero exit produced a clean-looking tuple — cannot distinguish failure" + ) + + +def test_process_capture_stdio_via_shell_redirect(batch: Batch, tmp_path): + """Pins Rhapsody's ``capture_stdio`` workaround. + + A wrapper bash script redirects its own stdout/stderr to files, then + runs the real command. Dragon launches the bash; the files end up with + the captured output regardless of Dragon's PIPE behaviour. + """ + stdout_path = tmp_path / "task.stdout" + script_path = tmp_path / "task.sh" + script_path.write_text( + f"#!/usr/bin/bash\n" + f"/bin/echo 'stdout-from-wrapped' 1>{stdout_path}\n" + ) + script_path.chmod(0o755) + + task = batch.process(ProcessTemplate("/bin/bash", args=(str(script_path),))) + _ = batch.results_ddict[task.uid] # wait for completion + + assert stdout_path.exists(), "wrapper script did not produce stdout file" + assert "stdout-from-wrapped" in stdout_path.read_text() diff --git a/tests/dragon_ci/test_ddict.py b/tests/dragon_ci/test_ddict.py new file mode 100644 index 0000000..f2aa2b2 --- /dev/null +++ b/tests/dragon_ci/test_ddict.py @@ -0,0 +1,90 @@ +"""Contract: ``dragon.data.ddict.ddict.DDict`` core surface. + +Rhapsody V3 reads ``batch.results_ddict``; downstream users construct +DDicts directly via the kwargs and methods checked below. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.data.ddict.ddict import DDict + + +_DDICT_KWARGS = ( + "managers_per_node", "n_nodes", "total_mem", + "working_set_size", "wait_for_keys", "policy", "timeout", +) +_DDICT_METHODS = ("pput", "bput", "clear", "destroy", "detach", "keys", "items", "freeze") + + +@pytest.mark.parametrize("kwarg", _DDICT_KWARGS) +def test_ddict_ctor_kwarg(kwarg): + assert kwarg in inspect.signature(DDict.__init__).parameters, ( + f"DDict.__init__ no longer accepts {kwarg!r}" + ) + + +@pytest.mark.parametrize("method", _DDICT_METHODS) +def test_ddict_method_present(method): + assert callable(getattr(DDict, method, None)), f"DDict.{method}() removed" + + +@pytest.fixture(scope="module") +def _shared_ddict(): + """One DDict built once per module — tear down on exit.""" + d = DDict(n_nodes=1, managers_per_node=1, total_mem=4 * 1024 * 1024, + wait_for_keys=False, working_set_size=1) + try: + yield d + finally: + d.destroy() + + +@pytest.fixture +def ddict(_shared_ddict): + """Per-test view on the shared DDict, cleared each call.""" + _shared_ddict.clear() + return _shared_ddict + + +def test_ddict_setitem_getitem_contains(ddict): + ddict["a"] = 1 + assert "a" in ddict and ddict["a"] == 1 + + +def test_ddict_pput(ddict): + ddict.pput("k", {"nested": [1, 2, 3]}) + assert ddict["k"] == {"nested": [1, 2, 3]} + + +def test_ddict_delitem(ddict): + ddict["k"] = "v" + del ddict["k"] + assert "k" not in ddict + + +def test_ddict_clear_removes_all(ddict): + ddict["x"] = 1 + ddict["y"] = 2 + ddict.clear() + assert "x" not in ddict and "y" not in ddict + + +def test_ddict_keyerror_on_missing(ddict): + with pytest.raises(KeyError): + _ = ddict["missing"] + + +def test_ddict_instance_has_is_frozen(ddict): + """``is_frozen`` lives on the instance (not on the class) in Dragon 0.14.""" + assert hasattr(ddict, "is_frozen") + + +def test_ddict_round_trip_complex_payload(ddict): + payload = {"list": [1, 2, 3], "tuple": (4, 5), "set": {6, 7}, + "nested": {"a": [(0, "z")]}} + ddict["complex"] = payload + assert ddict["complex"] == payload diff --git a/tests/dragon_ci/test_ddict_unknown_key_blocks.py b/tests/dragon_ci/test_ddict_unknown_key_blocks.py new file mode 100644 index 0000000..f7e9696 --- /dev/null +++ b/tests/dragon_ci/test_ddict_unknown_key_blocks.py @@ -0,0 +1,39 @@ +"""Documented Dragon bug — kept failing until Dragon fixes it. + +A DDict built with ``wait_for_keys=True`` (the setting Batch's results-DDict +uses) **blocks indefinitely** when a never-written key is read, instead of +raising ``KeyError``. Rhapsody's V3 monitor loop has a +``try/except KeyError: continue`` polling pattern that depends on the +expected KeyError-on-missing contract — when that contract isn't honored, +the monitor loop deadlocks on the first never-arriving result. + +This is the silent-hang failure mode that prompted the entire test suite. +We pin it via ``pytest.mark.timeout`` so pytest emits a clear failure +report with the exact blocking call in the trace. When Dragon fixes the +bug, ``KeyError`` will fire fast, the ``pytest.raises`` block will pass, +and the test goes green. + +Lives in its own file because ``method="thread"`` leaks the test thread +(it's still wedged inside Dragon's C-level channel read) and the leaked +thread holds resources that break any subsequent test in the same pytest +invocation. +""" + +import pytest + +from dragon.data.ddict.ddict import DDict + + +@pytest.mark.timeout(5, method="thread") +def test_ddict_get_unknown_key_blocks_forever(): + # wait_for_keys=True requires working_set_size > 1 (Dragon enforces it). + d = DDict(n_nodes=1, managers_per_node=1, total_mem=4 * 1024 * 1024, + wait_for_keys=True, working_set_size=2) + try: + with pytest.raises(KeyError): + _ = d["never-written"] + finally: + try: + d.destroy() + except Exception: + pass diff --git a/tests/dragon_ci/test_machine_system.py b/tests/dragon_ci/test_machine_system.py new file mode 100644 index 0000000..a6a4d72 --- /dev/null +++ b/tests/dragon_ci/test_machine_system.py @@ -0,0 +1,46 @@ +"""Contract: ``dragon.native.machine.System`` shape and node enumeration. + +Rhapsody's telemetry adapter calls ``System().hostname_policies()`` to obtain +one ``Policy`` per node, then pins one worker per Policy. +""" + +from __future__ import annotations + +import pytest + +from dragon.infrastructure.policy import Policy +from dragon.native.machine import System + + +@pytest.mark.parametrize("attr", ["nnodes", "hostname_policies"]) +def test_system_attribute_present(attr): + """Rhapsody reads ``System().nnodes`` (V1 DDict sizing) and calls + ``hostname_policies()`` (telemetry adapter, one worker per node).""" + assert hasattr(System(), attr), f"System.{attr} removed" + + +def test_system_nnodes_is_int_ge_1(): + assert isinstance(System().nnodes, int) and System().nnodes >= 1 + + +def test_hostname_policies_returns_one_policy_per_node(): + s = System() + policies = s.hostname_policies() + assert isinstance(policies, list) + assert len(policies) == s.nnodes + assert all(isinstance(p, Policy) for p in policies) + + +def test_hostname_policies_identify_each_node(): + """Each per-node Policy must carry an identifier (host_name or host_id).""" + for p in System().hostname_policies(): + assert p.host_name or p.host_id >= 0 or p.placement == Policy.Placement.HOST_NAME, ( + f"hostname_policies returned a Policy with no node identifier: {p}" + ) + + +@pytest.mark.requires_multi_node +def test_hostname_policies_are_distinct(): + policies = System().hostname_policies() + keys = [(p.host_name, p.host_id) for p in policies] + assert len(set(keys)) == len(policies), f"duplicate identifiers: {keys}" diff --git a/tests/dragon_ci/test_policy.py b/tests/dragon_ci/test_policy.py new file mode 100644 index 0000000..a245ffd --- /dev/null +++ b/tests/dragon_ci/test_policy.py @@ -0,0 +1,57 @@ +"""Contract: ``dragon.infrastructure.policy.Policy`` shape. + +Rhapsody V3 builds Policy instances with combinations of ``host_id``, +``distribution``, ``placement``, ``host_name``, and ``gpu_affinity``. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.infrastructure.policy import Policy + + +_POLICY_KWARGS = ( + "placement", "host_name", "host_id", "distribution", + "cpu_affinity", "gpu_affinity", "wait_mode", +) + + +@pytest.mark.parametrize("kwarg", _POLICY_KWARGS) +def test_policy_ctor_kwarg(kwarg): + assert kwarg in inspect.signature(Policy.__init__).parameters, ( + f"Policy.__init__ no longer accepts {kwarg!r}" + ) + + +@pytest.mark.parametrize("attr", _POLICY_KWARGS) +def test_policy_attribute_present(attr): + assert hasattr(Policy(), attr), f"Policy().{attr} attribute removed" + + +@pytest.mark.parametrize( + "enum, required", + [ + (Policy.Distribution, ("BLOCK", "ROUNDROBIN", "DEFAULT")), + (Policy.Placement, ("DEFAULT", "HOST_NAME", "HOST_ID", "ANYWHERE", "LOCAL")), + ], +) +def test_policy_enum_members(enum, required): + members = {m.name for m in enum} + missing = set(required) - members + assert not missing, f"{enum.__name__} missing members: {missing}; got {members}" + + +def test_policy_with_host_id_and_block_distribution(): + p = Policy(host_id=0, distribution=Policy.Distribution.BLOCK) + assert p.host_id == 0 + assert p.distribution == Policy.Distribution.BLOCK + + +def test_policy_with_placement_host_name_and_gpu_affinity(): + p = Policy(placement=Policy.Placement.HOST_NAME, host_name="x", gpu_affinity=[0, 1]) + assert p.placement == Policy.Placement.HOST_NAME + assert p.host_name == "x" + assert p.gpu_affinity == [0, 1] diff --git a/tests/dragon_ci/test_process_group.py b/tests/dragon_ci/test_process_group.py new file mode 100644 index 0000000..ce54657 --- /dev/null +++ b/tests/dragon_ci/test_process_group.py @@ -0,0 +1,69 @@ +"""Contract: native ``ProcessGroup`` lifecycle. + +The telemetry adapter spawns one worker per node via:: + + grp = ProcessGroup(restart=False, pmi=None) + grp.add_process(nproc=1, template=ProcessTemplate(target=worker_fn, ...)) + grp.init(); grp.start(); ...; grp.join(timeout=...); grp.close() + +and treats ``DragonUserCodeError`` as an expected exception class on +graceful shutdown. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.native.event import Event +from dragon.native.process import ProcessTemplate +from dragon.native.process_group import DragonUserCodeError, ProcessGroup +from dragon.native.queue import Queue + +from _dragon_ci_helpers import pg_worker # noqa: E402 + + +@pytest.mark.parametrize("kwarg", ["restart", "pmi", "policy"]) +def test_process_group_ctor_kwarg(kwarg): + assert kwarg in inspect.signature(ProcessGroup.__init__).parameters, ( + f"ProcessGroup.__init__ no longer accepts {kwarg!r}" + ) + + +@pytest.mark.parametrize("method", ["add_process", "init", "start", "join", "stop", "close"]) +def test_process_group_method_present(method): + assert callable(getattr(ProcessGroup, method, None)), ( + f"ProcessGroup.{method}() removed" + ) + + +@pytest.mark.parametrize("kwarg", ["nproc", "template"]) +def test_process_group_add_process_kwarg(kwarg): + assert kwarg in inspect.signature(ProcessGroup.add_process).parameters, ( + f"ProcessGroup.add_process no longer accepts {kwarg!r}" + ) + + +def test_dragon_user_code_error_importable(): + """Rhapsody catches ``DragonUserCodeError`` during graceful shutdown.""" + assert isinstance(DragonUserCodeError, type) + assert issubclass(DragonUserCodeError, BaseException) + + +def test_process_group_lifecycle_smoke(): + """Build, init, start, join, close a 1-worker group end-to-end.""" + q = Queue(maxsize=10) + shutdown = Event() + + grp = ProcessGroup(restart=False, pmi=None) + grp.add_process(nproc=1, template=ProcessTemplate(target=pg_worker, args=(q, shutdown))) + grp.init() + grp.start() + + msg = q.get(timeout=30.0) + assert msg.get("pid") and msg.get("host") + + shutdown.set() + grp.join(timeout=30.0) + grp.close() diff --git a/tests/dragon_ci/test_process_template.py b/tests/dragon_ci/test_process_template.py new file mode 100644 index 0000000..cbb3015 --- /dev/null +++ b/tests/dragon_ci/test_process_template.py @@ -0,0 +1,80 @@ +"""Contract: ``dragon.native.process.ProcessTemplate`` constructor and attributes. + +Rhapsody V3 builds ProcessTemplate objects with these kwargs:: + + ProcessTemplate(target, args=(...), kwargs={...}, cwd=..., env=..., + policy=Policy(...), stdin=Popen.DEVNULL, + stdout=Popen.PIPE, stderr=Popen.PIPE) + +and later introspects ``pt.cwd``, ``pt.policy``, ``pt.env`` and unpacks +``pt.argdata`` via ``cloudpickle.loads`` to verify ``(target, args, kwargs)`` +were preserved. +""" + +from __future__ import annotations + +import inspect + +import cloudpickle +import pytest + +from dragon.infrastructure.policy import Policy +from dragon.native.process import Popen, ProcessTemplate + + +_TEMPLATE_KWARGS = ( + "target", "args", "kwargs", "cwd", "env", "stdin", "stdout", "stderr", "policy", +) + + +@pytest.mark.parametrize("kwarg", _TEMPLATE_KWARGS) +def test_process_template_ctor_kwarg(kwarg): + assert kwarg in inspect.signature(ProcessTemplate.__init__).parameters, ( + f"ProcessTemplate.__init__ no longer accepts {kwarg!r}" + ) + + +@pytest.mark.parametrize("name", ["PIPE", "DEVNULL"]) +def test_popen_stdio_sentinel(name): + """Rhapsody uses ``Popen.PIPE`` and ``Popen.DEVNULL`` as stream sentinels.""" + assert hasattr(Popen, name), f"Popen.{name} constant removed" + + +def test_template_cwd_attribute(): + pt = ProcessTemplate("/bin/echo", args=("x",), cwd="/tmp") + assert pt.cwd == "/tmp" + + +def test_template_policy_identity_preserved(): + """``pt.policy is policy`` — Rhapsody's own tests rely on the same identity.""" + policy = Policy(gpu_affinity=[0, 1, 2, 3]) + pt = ProcessTemplate("/bin/echo", args=("x",), policy=policy) + assert pt.policy is policy + assert pt.policy.gpu_affinity == [0, 1, 2, 3] + + +def test_template_env_dict_preserved(): + pt = ProcessTemplate("/usr/bin/env", args=(), env={"FOO": "BAR"}) + assert pt.env.get("FOO") == "BAR" + + +def test_template_stdio_sentinels_accepted(): + """PIPE/DEVNULL must be acceptable to the stdin/stdout/stderr kwargs.""" + ProcessTemplate( + "/bin/echo", args=("x",), + stdin=Popen.DEVNULL, stdout=Popen.PIPE, stderr=Popen.PIPE, + ) + + +def test_template_argdata_round_trips_via_cloudpickle(): + """Rhapsody round-trips ``(target, args, kwargs)`` out of ``pt.argdata``.""" + def _target(x, y): + return x + y + + pt = ProcessTemplate(_target, args=(1, 2), kwargs={"z": 3}) + target, stored_args, stored_kwargs = cloudpickle.loads(pt.argdata) + # ``target`` is reconstituted across cloudpickle, so identity won't hold — + # compare behaviour instead. + assert target(1, 2) == _target(1, 2) + assert tuple(stored_args) == (1, 2) + assert stored_kwargs == {"z": 3} diff --git a/tests/dragon_ci/test_queue_event.py b/tests/dragon_ci/test_queue_event.py new file mode 100644 index 0000000..e0d95d6 --- /dev/null +++ b/tests/dragon_ci/test_queue_event.py @@ -0,0 +1,74 @@ +"""Contract: ``dragon.native.queue.Queue`` and ``dragon.native.event.Event``. + +The telemetry adapter uses these as the cross-process result channel and +shutdown signal: ``Queue(maxsize=)`` with ``put(timeout=)``/``get(timeout=)``, +``Event`` with ``wait(timeout=)``/``set()``. +""" + +from __future__ import annotations + +import inspect +import time + +import pytest + +from dragon.native.event import Event +from dragon.native.process import ProcessTemplate +from dragon.native.process_group import ProcessGroup +from dragon.native.queue import Queue + +from _dragon_ci_helpers import pg_worker # noqa: E402 + + +def test_queue_ctor_accepts_maxsize(): + assert "maxsize" in inspect.signature(Queue.__init__).parameters + + +@pytest.mark.parametrize("method", ["put", "get", "close"]) +def test_queue_method_present(method): + assert callable(getattr(Queue, method, None)), f"Queue.{method}() removed" + + +@pytest.mark.parametrize("method", ["wait", "set", "clear", "is_set"]) +def test_event_method_present(method): + assert callable(getattr(Event, method, None)), f"Event.{method}() removed" + + +def test_queue_put_get_roundtrip_in_process(): + q = Queue(maxsize=4) + q.put({"k": "v"}, timeout=5.0) + assert q.get(timeout=5.0) == {"k": "v"} + + +def test_event_wait_returns_truthy_when_set(): + e = Event() + assert not e.is_set() + e.set() + assert e.is_set() + assert e.wait(timeout=1.0) + + +def test_event_wait_returns_falsy_on_timeout(): + e = Event() + t0 = time.time() + rv = e.wait(timeout=0.2) + assert not rv, f"Event.wait should be falsy on timeout, got {rv!r}" + assert time.time() - t0 >= 0.15, "Event.wait returned before its timeout" + + +def test_queue_event_cross_process_roundtrip(): + """A ProcessGroup worker pushes to a Queue and observes an Event.""" + q = Queue(maxsize=8) + ev = Event() + + grp = ProcessGroup(restart=False, pmi=None) + grp.add_process(nproc=1, template=ProcessTemplate(target=pg_worker, args=(q, ev))) + grp.init() + grp.start() + try: + msg = q.get(timeout=30.0) + assert msg.get("pid") and msg.get("host") + finally: + ev.set() + grp.join(timeout=30.0) + grp.close() diff --git a/tests/dragon_ci/test_sigkill_in_worker_hangs.py b/tests/dragon_ci/test_sigkill_in_worker_hangs.py new file mode 100644 index 0000000..adb2c7c --- /dev/null +++ b/tests/dragon_ci/test_sigkill_in_worker_hangs.py @@ -0,0 +1,27 @@ +"""Documented Dragon bug — kept failing until Dragon fixes it. + +``SIGKILL`` inside a function task denies Dragon a clean exit signal, so +the result-DDict entry is never written and the dispatcher has no way to +learn the worker died. ``task.get()`` then blocks indefinitely — its +``timeout=`` kwarg only governs manager selection, not the blocking +DDict read. + +When Dragon adds abnormal-worker-death detection (or finally honours +``task.get(timeout=)``), the call below will return, pytest-timeout will +stop firing, and the test goes green. + +Lives in its own file because ``method="thread"`` leaks the test thread +(still wedged inside Dragon's blocking read) and the leaked thread holds +resources that break any subsequent test in the same pytest invocation. +""" + +import pytest + +from dragon.workflows.batch import Batch + +from _dragon_ci_helpers import fn_kill_self # noqa: E402 + + +@pytest.mark.timeout(5, method="thread") +def test_sigkill_in_worker_hangs(fresh_batch: Batch): + fresh_batch.function(fn_kill_self, 1).get(timeout=2.0) diff --git a/tests/dragon_ci/test_telemetry_collector.py b/tests/dragon_ci/test_telemetry_collector.py new file mode 100644 index 0000000..68a5228 --- /dev/null +++ b/tests/dragon_ci/test_telemetry_collector.py @@ -0,0 +1,63 @@ +"""Contract: ``dragon.telemetry.collector`` + ``AccVendor``. + +Rhapsody's ``DragonTelemetryAdapter`` imports these helpers directly to +collect per-node GPU metrics inside a Dragon ProcessGroup worker. +""" + +from __future__ import annotations + +import inspect + +import pytest + +from dragon.infrastructure.gpu_desc import AccVendor, find_accelerators +from dragon.telemetry.collector import ( + get_amd_metrics, + get_intel_metrics, + get_nvidia_metrics, + identify_gpu, +) + +# All tests here need dragon.telemetry.collector — auto-skip if unavailable. +pytestmark = pytest.mark.requires_telemetry_collector + + +@pytest.mark.parametrize( + "fn", + [identify_gpu, get_amd_metrics, get_intel_metrics, get_nvidia_metrics, find_accelerators], +) +def test_collector_symbol_callable(fn): + assert callable(fn), f"{fn} no longer callable" + + +@pytest.mark.parametrize("name", ["NVIDIA", "AMD", "INTEL"]) +def test_acc_vendor_member(name): + assert name in {m.name for m in AccVendor}, f"AccVendor.{name} removed" + + +def test_identify_gpu_signature_and_return_shape(): + assert len(inspect.signature(identify_gpu).parameters) == 0, ( + "identify_gpu() now takes arguments" + ) + vendor, _count = identify_gpu() + if vendor is not None: + assert vendor in list(AccVendor), f"unknown vendor: {vendor!r}" + + +def test_dragon_telemetry_module_importable(): + """The bare ``dragon.telemetry`` import is used as the adapter gate.""" + import dragon.telemetry # noqa: F401 + + +@pytest.mark.requires_gpu +def test_get_nvidia_metrics_returns_metric_value_dicts(): + """Telemetry adapter expects ``[{metric: ..., value: ...}, ...]``.""" + vendor, _count = identify_gpu() + if vendor != AccVendor.NVIDIA: + pytest.skip(f"NVIDIA GPU expected, got {vendor!r}") + + metrics = get_nvidia_metrics(0, telemetry_level=3) + assert metrics, "get_nvidia_metrics returned empty for present GPU" + for m in metrics: + assert isinstance(m, dict) + assert "metric" in m and "value" in m, f"metric dict shape changed: {m!r}" diff --git a/tests/dragon_ci/test_unpickleable_function_hangs.py b/tests/dragon_ci/test_unpickleable_function_hangs.py new file mode 100644 index 0000000..3cfc5c3 --- /dev/null +++ b/tests/dragon_ci/test_unpickleable_function_hangs.py @@ -0,0 +1,42 @@ +"""Documented Dragon bug — kept failing until Dragon fixes it. + +A helper module made importable in the parent only via runtime +``sys.path.insert`` (not via ``PYTHONPATH``) cannot be re-imported by the +worker. cloudpickle pickles the helper by reference, the worker dies +during unpickling, and ``task.get()`` blocks forever — same root cause +as the SIGKILL hang. + +This is the silent-hang failure mode pytest users hit when they define a +helper at the top of a ``test_*.py`` file and pass it to +``batch.function()``. The conftest works around it by adding this +directory to ``PYTHONPATH``; the test below disables that workaround +locally to reproduce the bug. + +Lives in its own file for the same reason as the other ``*_hangs.py`` +files: ``method="thread"`` leaks the test thread which breaks +subsequent tests in the same pytest invocation. +""" + +import os +import sys + +import pytest + +from dragon.workflows.batch import Batch + + +@pytest.mark.timeout(5, method="thread") +def test_unpickleable_function_hangs(fresh_batch: Batch): + here = os.path.dirname(os.path.abspath(__file__)) + offpath = os.path.join(here, "_offpath") + assert offpath not in os.environ.get("PYTHONPATH", "").split(os.pathsep), ( + "test setup error: _offpath ended up on PYTHONPATH" + ) + + sys.path.insert(0, offpath) + try: + from marker_helper import offpath_add # noqa: PLC0415 + finally: + sys.path.remove(offpath) + + fresh_batch.function(offpath_add, 3, 4).get(timeout=2.0) diff --git a/tests/dragon_ci/test_worker_failure_modes.py b/tests/dragon_ci/test_worker_failure_modes.py new file mode 100644 index 0000000..8e5171a --- /dev/null +++ b/tests/dragon_ci/test_worker_failure_modes.py @@ -0,0 +1,44 @@ +"""Contract: Dragon's worker-failure detection — the *clean* paths. + +Two failure modes Dragon DOES detect, surfaced through ``task.get()``: + +1. **Caught exception** in the user function → the exception instance and + its traceback land in the results-DDict 5-tuple (``raised=True``), + and ``task.get()`` re-raises it. +2. **Clean process exit** (``sys.exit(N)``) → no result is written, but + Dragon's dispatcher notices and synthesises + ``RuntimeError("function worker exited without producing output: …")``. + +The silent-hang failure modes (abnormal worker death, unpickleable +function) live in their own files because they leak threads that break +subsequent tests in the same pytest invocation: + +- ``test_sigkill_in_worker_hangs.py`` +- ``test_unpickleable_function_hangs.py`` +- ``test_ddict_unknown_key_blocks.py`` +""" + +from __future__ import annotations + +import pytest + +from dragon.workflows.batch import Batch + +from _dragon_ci_helpers import fn_assert_false, fn_sys_exit_one # noqa: E402 + + +def test_caught_exception_is_reraised_and_tuple_recorded(batch: Batch): + task = batch.function(fn_assert_false, 1) + with pytest.raises(AssertionError, match="intentional dragon_ci probe"): + task.get(timeout=10.0) + result, tb, raised, _stdout, _stderr = batch.results_ddict[task.uid] + assert raised is True + assert isinstance(result, AssertionError) + assert isinstance(tb, str) and "AssertionError" in tb + + +def test_sys_exit_in_worker_raises_runtime_error(batch: Batch): + """Clean ``sys.exit`` surfaces as ``RuntimeError`` with that exact phrase.""" + task = batch.function(fn_sys_exit_one, 1) + with pytest.raises(RuntimeError, match="exited without producing output"): + task.get(timeout=10.0) From 3edae6d1340afa5241bd195e02e3cf67c2c142ee Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Mon, 25 May 2026 08:56:29 +0200 Subject: [PATCH 2/5] linting --- tests/dragon_ci/_dragon_ci_helpers.py | 1 - tests/dragon_ci/conftest.py | 1 - tests/dragon_ci/test_async_in_batch.py | 4 ++-- tests/dragon_ci/test_batch_function.py | 13 +++++++------ tests/dragon_ci/test_batch_job.py | 1 - tests/dragon_ci/test_batch_lifecycle.py | 2 -- tests/dragon_ci/test_batch_process.py | 1 - tests/dragon_ci/test_ddict.py | 2 -- tests/dragon_ci/test_ddict_unknown_key_blocks.py | 1 - tests/dragon_ci/test_machine_system.py | 1 - tests/dragon_ci/test_policy.py | 2 -- tests/dragon_ci/test_process_group.py | 7 +++---- tests/dragon_ci/test_process_template.py | 5 ++--- tests/dragon_ci/test_queue_event.py | 4 +--- tests/dragon_ci/test_sigkill_in_worker_hangs.py | 4 +--- tests/dragon_ci/test_telemetry_collector.py | 14 ++++++-------- .../dragon_ci/test_unpickleable_function_hangs.py | 1 - tests/dragon_ci/test_worker_failure_modes.py | 5 ++--- 18 files changed, 24 insertions(+), 45 deletions(-) diff --git a/tests/dragon_ci/_dragon_ci_helpers.py b/tests/dragon_ci/_dragon_ci_helpers.py index 98c97e1..ccd51dc 100644 --- a/tests/dragon_ci/_dragon_ci_helpers.py +++ b/tests/dragon_ci/_dragon_ci_helpers.py @@ -11,7 +11,6 @@ import sys import time - # --- Batch.function targets ---------------------------------------------- diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py index 52b8244..6ccfef8 100644 --- a/tests/dragon_ci/conftest.py +++ b/tests/dragon_ci/conftest.py @@ -32,7 +32,6 @@ import pytest import pytest_asyncio - # --- Make _dragon_ci_helpers importable from workers ---------------------- _HERE = os.path.dirname(os.path.abspath(__file__)) diff --git a/tests/dragon_ci/test_async_in_batch.py b/tests/dragon_ci/test_async_in_batch.py index 2f5a1f3..71a560c 100644 --- a/tests/dragon_ci/test_async_in_batch.py +++ b/tests/dragon_ci/test_async_in_batch.py @@ -18,10 +18,10 @@ import asyncio +from _dragon_ci_helpers import async_double # noqa: E402 +from _dragon_ci_helpers import async_run_shim # noqa: E402 from dragon.workflows.batch import Batch -from _dragon_ci_helpers import async_double, async_run_shim # noqa: E402 - def test_iscoroutinefunction_detects_async_def(): assert asyncio.iscoroutinefunction(async_double) diff --git a/tests/dragon_ci/test_batch_function.py b/tests/dragon_ci/test_batch_function.py index d217821..dc71c26 100644 --- a/tests/dragon_ci/test_batch_function.py +++ b/tests/dragon_ci/test_batch_function.py @@ -15,13 +15,14 @@ from __future__ import annotations import pytest - +from _dragon_ci_helpers import add # noqa: E402 (must be importable from workers) +from _dragon_ci_helpers import kwfn # noqa: E402 (must be importable from workers) +from _dragon_ci_helpers import print_and_return # noqa: E402 (must be importable from workers) +from _dragon_ci_helpers import raise_value_error # noqa: E402 (must be importable from workers) +from _dragon_ci_helpers import slow_double # noqa: E402 (must be importable from workers) from dragon.workflows.batch import Batch -from dragon.workflows.batch.batch import Function, TaskNotReadyError - -from _dragon_ci_helpers import ( # noqa: E402 (must be importable from workers) - add, kwfn, print_and_return, raise_value_error, slow_double, -) +from dragon.workflows.batch.batch import Function +from dragon.workflows.batch.batch import TaskNotReadyError def test_function_returns_function_handle(batch: Batch): diff --git a/tests/dragon_ci/test_batch_job.py b/tests/dragon_ci/test_batch_job.py index 02c698a..8d429cd 100644 --- a/tests/dragon_ci/test_batch_job.py +++ b/tests/dragon_ci/test_batch_job.py @@ -11,7 +11,6 @@ import inspect import pytest - from dragon.infrastructure.facts import PMIBackend from dragon.native.process import ProcessTemplate from dragon.workflows.batch import Batch diff --git a/tests/dragon_ci/test_batch_lifecycle.py b/tests/dragon_ci/test_batch_lifecycle.py index 467fb50..4300987 100644 --- a/tests/dragon_ci/test_batch_lifecycle.py +++ b/tests/dragon_ci/test_batch_lifecycle.py @@ -11,10 +11,8 @@ import inspect import pytest - from dragon.workflows.batch import Batch - _BATCH_CTOR_KWARGS = ("num_nodes", "pool_nodes", "disable_telem", "scheduler_workers", "results_ddict_mem") _BATCH_METHODS = ("function", "process", "job", "fence", "close", "join", "terminate") diff --git a/tests/dragon_ci/test_batch_process.py b/tests/dragon_ci/test_batch_process.py index 58f237b..c1434a2 100644 --- a/tests/dragon_ci/test_batch_process.py +++ b/tests/dragon_ci/test_batch_process.py @@ -14,7 +14,6 @@ from __future__ import annotations import pytest - from dragon.native.process import ProcessTemplate from dragon.workflows.batch import Batch diff --git a/tests/dragon_ci/test_ddict.py b/tests/dragon_ci/test_ddict.py index f2aa2b2..b7508ea 100644 --- a/tests/dragon_ci/test_ddict.py +++ b/tests/dragon_ci/test_ddict.py @@ -9,10 +9,8 @@ import inspect import pytest - from dragon.data.ddict.ddict import DDict - _DDICT_KWARGS = ( "managers_per_node", "n_nodes", "total_mem", "working_set_size", "wait_for_keys", "policy", "timeout", diff --git a/tests/dragon_ci/test_ddict_unknown_key_blocks.py b/tests/dragon_ci/test_ddict_unknown_key_blocks.py index f7e9696..ebf250c 100644 --- a/tests/dragon_ci/test_ddict_unknown_key_blocks.py +++ b/tests/dragon_ci/test_ddict_unknown_key_blocks.py @@ -20,7 +20,6 @@ """ import pytest - from dragon.data.ddict.ddict import DDict diff --git a/tests/dragon_ci/test_machine_system.py b/tests/dragon_ci/test_machine_system.py index a6a4d72..45ec310 100644 --- a/tests/dragon_ci/test_machine_system.py +++ b/tests/dragon_ci/test_machine_system.py @@ -7,7 +7,6 @@ from __future__ import annotations import pytest - from dragon.infrastructure.policy import Policy from dragon.native.machine import System diff --git a/tests/dragon_ci/test_policy.py b/tests/dragon_ci/test_policy.py index a245ffd..7405806 100644 --- a/tests/dragon_ci/test_policy.py +++ b/tests/dragon_ci/test_policy.py @@ -9,10 +9,8 @@ import inspect import pytest - from dragon.infrastructure.policy import Policy - _POLICY_KWARGS = ( "placement", "host_name", "host_id", "distribution", "cpu_affinity", "gpu_affinity", "wait_mode", diff --git a/tests/dragon_ci/test_process_group.py b/tests/dragon_ci/test_process_group.py index ce54657..69f9132 100644 --- a/tests/dragon_ci/test_process_group.py +++ b/tests/dragon_ci/test_process_group.py @@ -15,14 +15,13 @@ import inspect import pytest - +from _dragon_ci_helpers import pg_worker # noqa: E402 from dragon.native.event import Event from dragon.native.process import ProcessTemplate -from dragon.native.process_group import DragonUserCodeError, ProcessGroup +from dragon.native.process_group import DragonUserCodeError +from dragon.native.process_group import ProcessGroup from dragon.native.queue import Queue -from _dragon_ci_helpers import pg_worker # noqa: E402 - @pytest.mark.parametrize("kwarg", ["restart", "pmi", "policy"]) def test_process_group_ctor_kwarg(kwarg): diff --git a/tests/dragon_ci/test_process_template.py b/tests/dragon_ci/test_process_template.py index cbb3015..b48e183 100644 --- a/tests/dragon_ci/test_process_template.py +++ b/tests/dragon_ci/test_process_template.py @@ -17,10 +17,9 @@ import cloudpickle import pytest - from dragon.infrastructure.policy import Policy -from dragon.native.process import Popen, ProcessTemplate - +from dragon.native.process import Popen +from dragon.native.process import ProcessTemplate _TEMPLATE_KWARGS = ( "target", "args", "kwargs", "cwd", "env", "stdin", "stdout", "stderr", "policy", diff --git a/tests/dragon_ci/test_queue_event.py b/tests/dragon_ci/test_queue_event.py index e0d95d6..e7dbd2e 100644 --- a/tests/dragon_ci/test_queue_event.py +++ b/tests/dragon_ci/test_queue_event.py @@ -11,14 +11,12 @@ import time import pytest - +from _dragon_ci_helpers import pg_worker # noqa: E402 from dragon.native.event import Event from dragon.native.process import ProcessTemplate from dragon.native.process_group import ProcessGroup from dragon.native.queue import Queue -from _dragon_ci_helpers import pg_worker # noqa: E402 - def test_queue_ctor_accepts_maxsize(): assert "maxsize" in inspect.signature(Queue.__init__).parameters diff --git a/tests/dragon_ci/test_sigkill_in_worker_hangs.py b/tests/dragon_ci/test_sigkill_in_worker_hangs.py index adb2c7c..11624a8 100644 --- a/tests/dragon_ci/test_sigkill_in_worker_hangs.py +++ b/tests/dragon_ci/test_sigkill_in_worker_hangs.py @@ -16,10 +16,8 @@ """ import pytest - -from dragon.workflows.batch import Batch - from _dragon_ci_helpers import fn_kill_self # noqa: E402 +from dragon.workflows.batch import Batch @pytest.mark.timeout(5, method="thread") diff --git a/tests/dragon_ci/test_telemetry_collector.py b/tests/dragon_ci/test_telemetry_collector.py index 68a5228..30dbdfc 100644 --- a/tests/dragon_ci/test_telemetry_collector.py +++ b/tests/dragon_ci/test_telemetry_collector.py @@ -9,14 +9,12 @@ import inspect import pytest - -from dragon.infrastructure.gpu_desc import AccVendor, find_accelerators -from dragon.telemetry.collector import ( - get_amd_metrics, - get_intel_metrics, - get_nvidia_metrics, - identify_gpu, -) +from dragon.infrastructure.gpu_desc import AccVendor +from dragon.infrastructure.gpu_desc import find_accelerators +from dragon.telemetry.collector import get_amd_metrics +from dragon.telemetry.collector import get_intel_metrics +from dragon.telemetry.collector import get_nvidia_metrics +from dragon.telemetry.collector import identify_gpu # All tests here need dragon.telemetry.collector — auto-skip if unavailable. pytestmark = pytest.mark.requires_telemetry_collector diff --git a/tests/dragon_ci/test_unpickleable_function_hangs.py b/tests/dragon_ci/test_unpickleable_function_hangs.py index 3cfc5c3..4cb034d 100644 --- a/tests/dragon_ci/test_unpickleable_function_hangs.py +++ b/tests/dragon_ci/test_unpickleable_function_hangs.py @@ -21,7 +21,6 @@ import sys import pytest - from dragon.workflows.batch import Batch diff --git a/tests/dragon_ci/test_worker_failure_modes.py b/tests/dragon_ci/test_worker_failure_modes.py index 8e5171a..eb354d9 100644 --- a/tests/dragon_ci/test_worker_failure_modes.py +++ b/tests/dragon_ci/test_worker_failure_modes.py @@ -21,11 +21,10 @@ from __future__ import annotations import pytest - +from _dragon_ci_helpers import fn_assert_false # noqa: E402 +from _dragon_ci_helpers import fn_sys_exit_one # noqa: E402 from dragon.workflows.batch import Batch -from _dragon_ci_helpers import fn_assert_false, fn_sys_exit_one # noqa: E402 - def test_caught_exception_is_reraised_and_tuple_recorded(batch: Batch): task = batch.function(fn_assert_false, 1) From 935ee5329589248dca67d2287d75c26abe53ee56 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Wed, 3 Jun 2026 15:12:52 +0200 Subject: [PATCH 3/5] finalize dragon API tests --- tests/dragon_ci/conftest.py | 35 +++++++++++++++++++ tests/dragon_ci/test_batch_process.py | 4 +-- .../test_ddict_unknown_key_blocks.py | 35 +++++++++++++------ .../dragon_ci/test_sigkill_in_worker_hangs.py | 27 +++++++++++--- .../test_unpickleable_function_hangs.py | 23 +++++++++++- 5 files changed, 106 insertions(+), 18 deletions(-) diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py index 6ccfef8..7809405 100644 --- a/tests/dragon_ci/conftest.py +++ b/tests/dragon_ci/conftest.py @@ -83,6 +83,41 @@ def _has_telemetry_collector() -> bool: _HAS_PMI = os.environ.get("DRAGON_CI_HAS_PMI", "").lower() not in ("", "0", "false") +# --- Early-abort gate: must run under `dragon python` -------------------- + + +def pytest_configure(config): + """Abort early if not running under the Dragon launcher. + + Without the gate, ``pytest tests/dragon_ci`` collects fine (dragon is + importable from the venv) but each test fails individually with cryptic + ``DRAGON_MEMORY_ERRNO`` channel-attach errors. The gate gives a single + actionable message instead. + + Fast path: scan for any ``DRAGON_*`` env var — the launcher exports + several of them. Slow path: if env vars are present but the runtime + is unreachable (e.g. a stale dragon session), the behavioural probe + catches that. + """ + dragon_envs = [k for k in os.environ if k.startswith("DRAGON_")] + if not dragon_envs: + pytest.exit( + "Dragon CI suite must run under the Dragon launcher:\n" + " dragon python -m pytest -c tests/dragon_ci/pytest.ini " + "--rootdir=tests/dragon_ci tests/dragon_ci/", + returncode=2, + ) + try: + _detect_nnodes() + except Exception as exc: + pytest.exit( + f"DRAGON_* env vars present ({len(dragon_envs)} found) but the " + f"Dragon runtime is unreachable — likely a stale dragon session. " + f"Re-launch with `dragon python -m pytest ...`.\nProbe error: {exc!r}", + returncode=2, + ) + + # --- Auto-skip markers ---------------------------------------------------- # (marker_name, predicate-that-must-be-true-to-run, reason) diff --git a/tests/dragon_ci/test_batch_process.py b/tests/dragon_ci/test_batch_process.py index c1434a2..76d100f 100644 --- a/tests/dragon_ci/test_batch_process.py +++ b/tests/dragon_ci/test_batch_process.py @@ -19,8 +19,8 @@ def test_process_returns_five_tuple(batch: Batch): - """A trivial ``/bin/echo`` task completes and yields the 5-tuple shape.""" - task = batch.process(ProcessTemplate("/bin/echo", args=("hello-process",))) + """A trivial subprocess task completes and yields the 5-tuple shape.""" + task = batch.process(ProcessTemplate("/bin/true", args=())) entry = batch.results_ddict[task.uid] assert isinstance(entry, tuple) and len(entry) == 5, ( f"results_ddict shape changed for batch.process: {entry!r}" diff --git a/tests/dragon_ci/test_ddict_unknown_key_blocks.py b/tests/dragon_ci/test_ddict_unknown_key_blocks.py index ebf250c..7572674 100644 --- a/tests/dragon_ci/test_ddict_unknown_key_blocks.py +++ b/tests/dragon_ci/test_ddict_unknown_key_blocks.py @@ -1,4 +1,4 @@ -"""Documented Dragon bug — kept failing until Dragon fixes it. +"""Documented Dragon bug — **test disabled**. A DDict built with ``wait_for_keys=True`` (the setting Batch's results-DDict uses) **blocks indefinitely** when a never-written key is read, instead of @@ -8,22 +8,35 @@ the monitor loop deadlocks on the first never-arriving result. This is the silent-hang failure mode that prompted the entire test suite. -We pin it via ``pytest.mark.timeout`` so pytest emits a clear failure -report with the exact blocking call in the trace. When Dragon fixes the -bug, ``KeyError`` will fire fast, the ``pytest.raises`` block will pass, -and the test goes green. - -Lives in its own file because ``method="thread"`` leaks the test thread -(it's still wedged inside Dragon's C-level channel read) and the leaked -thread holds resources that break any subsequent test in the same pytest -invocation. + +The test below is **skipped** because letting it hit ``pytest.mark.timeout`` +produces a hard failure that doesn't compose with ``pytest.mark.xfail`` +(pytest-timeout raises a ``BaseException`` subclass that xfail does not +catch). To check whether Dragon has fixed the bug, remove the +``pytest.mark.skip`` decorator and run this file under +``dragon python -m pytest`` — the test will either return cleanly (Dragon +fixed it, re-enable permanently) or pytest-timeout will fire (still broken). """ +import warnings + import pytest from dragon.data.ddict.ddict import DDict +warnings.warn( + "test_ddict_get_unknown_key_blocks_forever is DISABLED — Dragon bug " + "(DDict[unknown_key] blocks instead of raising KeyError when " + "wait_for_keys=True) is not actively checked. Re-enable to retest.", + UserWarning, + stacklevel=2, +) + -@pytest.mark.timeout(5, method="thread") +@pytest.mark.skip( + reason="DISABLED — Dragon bug: DDict[unknown_key] blocks instead of raising " + "KeyError when wait_for_keys=True. Re-enable to check if fixed." +) +@pytest.mark.timeout(10, method="thread") def test_ddict_get_unknown_key_blocks_forever(): # wait_for_keys=True requires working_set_size > 1 (Dragon enforces it). d = DDict(n_nodes=1, managers_per_node=1, total_mem=4 * 1024 * 1024, diff --git a/tests/dragon_ci/test_sigkill_in_worker_hangs.py b/tests/dragon_ci/test_sigkill_in_worker_hangs.py index 11624a8..c0d48d8 100644 --- a/tests/dragon_ci/test_sigkill_in_worker_hangs.py +++ b/tests/dragon_ci/test_sigkill_in_worker_hangs.py @@ -1,4 +1,4 @@ -"""Documented Dragon bug — kept failing until Dragon fixes it. +"""Documented Dragon bug — **test disabled**. ``SIGKILL`` inside a function task denies Dragon a clean exit signal, so the result-DDict entry is never written and the dispatcher has no way to @@ -6,20 +6,39 @@ ``timeout=`` kwarg only governs manager selection, not the blocking DDict read. -When Dragon adds abnormal-worker-death detection (or finally honours -``task.get(timeout=)``), the call below will return, pytest-timeout will -stop firing, and the test goes green. +The test below is **skipped** because letting it hit ``pytest.mark.timeout`` +produces a hard failure that doesn't compose with ``pytest.mark.xfail`` +(pytest-timeout raises a ``BaseException`` subclass that xfail does not +catch). To check whether Dragon has fixed the bug, remove the +``pytest.mark.skip`` decorator and run this file under +``dragon python -m pytest`` — the test will either return cleanly (Dragon +fixed it, re-enable permanently) or pytest-timeout will fire (still broken). Lives in its own file because ``method="thread"`` leaks the test thread (still wedged inside Dragon's blocking read) and the leaked thread holds resources that break any subsequent test in the same pytest invocation. """ +import warnings + import pytest from _dragon_ci_helpers import fn_kill_self # noqa: E402 from dragon.workflows.batch import Batch +warnings.warn( + "test_sigkill_in_worker_hangs is DISABLED — Dragon bug (abnormal worker " + "termination is not detected; task.get() blocks indefinitely and its " + "timeout= kwarg is not honored on the blocking DDict read) is not " + "actively checked. Re-enable to retest.", + UserWarning, + stacklevel=2, +) + +@pytest.mark.skip( + reason="DISABLED — Dragon bug: SIGKILL'd worker is undetected, task.get() " + "blocks forever, timeout= is ignored. Re-enable to check if fixed." +) @pytest.mark.timeout(5, method="thread") def test_sigkill_in_worker_hangs(fresh_batch: Batch): fresh_batch.function(fn_kill_self, 1).get(timeout=2.0) diff --git a/tests/dragon_ci/test_unpickleable_function_hangs.py b/tests/dragon_ci/test_unpickleable_function_hangs.py index 4cb034d..0697f27 100644 --- a/tests/dragon_ci/test_unpickleable_function_hangs.py +++ b/tests/dragon_ci/test_unpickleable_function_hangs.py @@ -1,4 +1,4 @@ -"""Documented Dragon bug — kept failing until Dragon fixes it. +"""Documented Dragon bug — **test disabled**. A helper module made importable in the parent only via runtime ``sys.path.insert`` (not via ``PYTHONPATH``) cannot be re-imported by the @@ -12,6 +12,14 @@ directory to ``PYTHONPATH``; the test below disables that workaround locally to reproduce the bug. +The test is **skipped** because letting it hit ``pytest.mark.timeout`` +produces a hard failure that doesn't compose with ``pytest.mark.xfail`` +(pytest-timeout raises a ``BaseException`` subclass that xfail does not +catch). To check whether Dragon has fixed the bug, remove the +``pytest.mark.skip`` decorator and run this file under +``dragon python -m pytest`` — the test will either return cleanly (Dragon +fixed it, re-enable permanently) or pytest-timeout will fire (still broken). + Lives in its own file for the same reason as the other ``*_hangs.py`` files: ``method="thread"`` leaks the test thread which breaks subsequent tests in the same pytest invocation. @@ -19,11 +27,24 @@ import os import sys +import warnings import pytest from dragon.workflows.batch import Batch +warnings.warn( + "test_unpickleable_function_hangs is DISABLED — Dragon bug (worker dies " + "during cloudpickle unpickling without diagnostic; task.get() blocks " + "forever) is not actively checked. Re-enable to retest.", + UserWarning, + stacklevel=2, +) + +@pytest.mark.skip( + reason="DISABLED — Dragon bug: worker silently dies during cloudpickle " + "unpickling; task.get() blocks forever. Re-enable to check if fixed." +) @pytest.mark.timeout(5, method="thread") def test_unpickleable_function_hangs(fresh_batch: Batch): here = os.path.dirname(os.path.abspath(__file__)) From 60ef7539ae01210ff957fa5ed4c77c4a545ca9c4 Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 30 Jun 2026 12:37:06 +0200 Subject: [PATCH 4/5] tests/dragon_ci: satisfy ruff-format/ruff/docformatter pre-commit gate Pure formatting (ruff-format + docformatter) across the contract suite, plus one ruff B011 fix: fn_assert_false now raises AssertionError explicitly instead of `assert False`, so the failure probe still fires under `python -O` (the test already expects that exact AssertionError). No behavioural change. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/dragon_ci/_dragon_ci_helpers.py | 4 +++- tests/dragon_ci/_offpath/marker_helper.py | 7 +++++-- tests/dragon_ci/conftest.py | 4 ++-- tests/dragon_ci/test_batch_function.py | 5 ++++- tests/dragon_ci/test_batch_job.py | 3 ++- tests/dragon_ci/test_batch_lifecycle.py | 15 +++++++++---- tests/dragon_ci/test_batch_process.py | 19 ++++++++--------- tests/dragon_ci/test_ddict.py | 21 +++++++++++++------ .../test_ddict_unknown_key_blocks.py | 11 +++++++--- tests/dragon_ci/test_machine_system.py | 4 ++-- tests/dragon_ci/test_policy.py | 9 ++++++-- tests/dragon_ci/test_process_group.py | 4 +--- tests/dragon_ci/test_process_template.py | 18 +++++++++++++--- .../dragon_ci/test_sigkill_in_worker_hangs.py | 2 +- .../test_unpickleable_function_hangs.py | 2 +- 15 files changed, 86 insertions(+), 42 deletions(-) diff --git a/tests/dragon_ci/_dragon_ci_helpers.py b/tests/dragon_ci/_dragon_ci_helpers.py index ccd51dc..65762d2 100644 --- a/tests/dragon_ci/_dragon_ci_helpers.py +++ b/tests/dragon_ci/_dragon_ci_helpers.py @@ -64,7 +64,9 @@ def pg_worker(queue, shutdown_event): def fn_assert_false(_x): - assert False, "intentional dragon_ci probe" + # Raise explicitly rather than `assert False` so the probe still fires under + # `python -O` (which strips asserts); the test expects this AssertionError. + raise AssertionError("intentional dragon_ci probe") def fn_sys_exit_one(_x): diff --git a/tests/dragon_ci/_offpath/marker_helper.py b/tests/dragon_ci/_offpath/marker_helper.py index c81a0c0..19fc43d 100644 --- a/tests/dragon_ci/_offpath/marker_helper.py +++ b/tests/dragon_ci/_offpath/marker_helper.py @@ -1,7 +1,10 @@ -"""Helper module deliberately placed in a subdirectory that the conftest -does NOT add to ``PYTHONPATH``. Tests that want to reproduce the runtime +"""Helper module deliberately placed in a subdirectory that the conftest does NOT add to +``PYTHONPATH``. + +Tests that want to reproduce the runtime ``sys.path.insert`` regression import from here. """ + def offpath_add(a, b): return a + b diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py index 7809405..7700967 100644 --- a/tests/dragon_ci/conftest.py +++ b/tests/dragon_ci/conftest.py @@ -176,8 +176,8 @@ async def batch(): @pytest_asyncio.fixture(loop_scope="session") async def fresh_batch(): - """A throwaway Batch for tests that intentionally kill workers — the - session ``batch`` cannot recover once a worker is SIGKILL'd. + """A throwaway Batch for tests that intentionally kill workers — the session ``batch`` cannot + recover once a worker is SIGKILL'd. Skips the graceful ``join()`` because dead workers make it block for the full timeout; ``terminate()`` can also raise ``DragonUserCodeError`` diff --git a/tests/dragon_ci/test_batch_function.py b/tests/dragon_ci/test_batch_function.py index dc71c26..35ca79a 100644 --- a/tests/dragon_ci/test_batch_function.py +++ b/tests/dragon_ci/test_batch_function.py @@ -53,7 +53,10 @@ def test_function_get_non_blocking_raises_when_not_ready(batch: Batch): def test_results_ddict_five_tuple_on_success(batch: Batch): - """Rhapsody unpacks exactly five fields. Order and shape are the contract.""" + """Rhapsody unpacks exactly five fields. + + Order and shape are the contract. + """ task = batch.function(add, 1, 1) task.get(timeout=60.0) entry = batch.results_ddict[task.uid] diff --git a/tests/dragon_ci/test_batch_job.py b/tests/dragon_ci/test_batch_job.py index 8d429cd..478fadf 100644 --- a/tests/dragon_ci/test_batch_job.py +++ b/tests/dragon_ci/test_batch_job.py @@ -41,7 +41,8 @@ def test_batch_job_launch(batch: Batch, nranks): try: job = batch.job( [(nranks, ProcessTemplate("/bin/true", args=()))], - name=f"dragon-ci-{nranks}rank", pmi=PMIBackend.PMIX, + name=f"dragon-ci-{nranks}rank", + pmi=PMIBackend.PMIX, ) job.get(timeout=120.0) except Exception as exc: # noqa: BLE001 diff --git a/tests/dragon_ci/test_batch_lifecycle.py b/tests/dragon_ci/test_batch_lifecycle.py index 4300987..45c3c98 100644 --- a/tests/dragon_ci/test_batch_lifecycle.py +++ b/tests/dragon_ci/test_batch_lifecycle.py @@ -13,8 +13,13 @@ import pytest from dragon.workflows.batch import Batch -_BATCH_CTOR_KWARGS = ("num_nodes", "pool_nodes", "disable_telem", - "scheduler_workers", "results_ddict_mem") +_BATCH_CTOR_KWARGS = ( + "num_nodes", + "pool_nodes", + "disable_telem", + "scheduler_workers", + "results_ddict_mem", +) _BATCH_METHODS = ("function", "process", "job", "fence", "close", "join", "terminate") @@ -43,8 +48,10 @@ def test_batch_results_ddict_supports_lookup(batch): def test_batch_fence_callable_on_idle_batch(batch): - """Rhapsody exposes ``Batch.fence()`` to user code as a barrier. On an - idle Batch it must complete without raising.""" + """Rhapsody exposes ``Batch.fence()`` to user code as a barrier. + + On an idle Batch it must complete without raising. + """ batch.fence() diff --git a/tests/dragon_ci/test_batch_process.py b/tests/dragon_ci/test_batch_process.py index 76d100f..1038129 100644 --- a/tests/dragon_ci/test_batch_process.py +++ b/tests/dragon_ci/test_batch_process.py @@ -43,9 +43,11 @@ def test_process_default_stdio_is_empty(batch: Batch): def test_process_template_args_none_raises_typeerror(batch: Batch): - """``ProcessTemplate(...).args`` defaults to None, which makes - ``Batch.process`` raise ``TypeError: 'NoneType' object is not iterable``. - Pass ``args=()`` to avoid it.""" + """``ProcessTemplate(...).args`` defaults to None, which makes ``Batch.process`` raise + ``TypeError: 'NoneType' object is not iterable``. + + Pass ``args=()`` to avoid it. + """ with pytest.raises(TypeError, match="NoneType"): batch.process(ProcessTemplate("/bin/true")) @@ -64,16 +66,13 @@ def test_process_non_zero_exit_is_visible(batch: Batch): def test_process_capture_stdio_via_shell_redirect(batch: Batch, tmp_path): """Pins Rhapsody's ``capture_stdio`` workaround. - A wrapper bash script redirects its own stdout/stderr to files, then - runs the real command. Dragon launches the bash; the files end up with - the captured output regardless of Dragon's PIPE behaviour. + A wrapper bash script redirects its own stdout/stderr to files, then runs the real command. + Dragon launches the bash; the files end up with the captured output regardless of Dragon's PIPE + behaviour. """ stdout_path = tmp_path / "task.stdout" script_path = tmp_path / "task.sh" - script_path.write_text( - f"#!/usr/bin/bash\n" - f"/bin/echo 'stdout-from-wrapped' 1>{stdout_path}\n" - ) + script_path.write_text(f"#!/usr/bin/bash\n/bin/echo 'stdout-from-wrapped' 1>{stdout_path}\n") script_path.chmod(0o755) task = batch.process(ProcessTemplate("/bin/bash", args=(str(script_path),))) diff --git a/tests/dragon_ci/test_ddict.py b/tests/dragon_ci/test_ddict.py index b7508ea..b0bd057 100644 --- a/tests/dragon_ci/test_ddict.py +++ b/tests/dragon_ci/test_ddict.py @@ -12,8 +12,13 @@ from dragon.data.ddict.ddict import DDict _DDICT_KWARGS = ( - "managers_per_node", "n_nodes", "total_mem", - "working_set_size", "wait_for_keys", "policy", "timeout", + "managers_per_node", + "n_nodes", + "total_mem", + "working_set_size", + "wait_for_keys", + "policy", + "timeout", ) _DDICT_METHODS = ("pput", "bput", "clear", "destroy", "detach", "keys", "items", "freeze") @@ -33,8 +38,13 @@ def test_ddict_method_present(method): @pytest.fixture(scope="module") def _shared_ddict(): """One DDict built once per module — tear down on exit.""" - d = DDict(n_nodes=1, managers_per_node=1, total_mem=4 * 1024 * 1024, - wait_for_keys=False, working_set_size=1) + d = DDict( + n_nodes=1, + managers_per_node=1, + total_mem=4 * 1024 * 1024, + wait_for_keys=False, + working_set_size=1, + ) try: yield d finally: @@ -82,7 +92,6 @@ def test_ddict_instance_has_is_frozen(ddict): def test_ddict_round_trip_complex_payload(ddict): - payload = {"list": [1, 2, 3], "tuple": (4, 5), "set": {6, 7}, - "nested": {"a": [(0, "z")]}} + payload = {"list": [1, 2, 3], "tuple": (4, 5), "set": {6, 7}, "nested": {"a": [(0, "z")]}} ddict["complex"] = payload assert ddict["complex"] == payload diff --git a/tests/dragon_ci/test_ddict_unknown_key_blocks.py b/tests/dragon_ci/test_ddict_unknown_key_blocks.py index 7572674..b077348 100644 --- a/tests/dragon_ci/test_ddict_unknown_key_blocks.py +++ b/tests/dragon_ci/test_ddict_unknown_key_blocks.py @@ -34,13 +34,18 @@ @pytest.mark.skip( reason="DISABLED — Dragon bug: DDict[unknown_key] blocks instead of raising " - "KeyError when wait_for_keys=True. Re-enable to check if fixed." + "KeyError when wait_for_keys=True. Re-enable to check if fixed." ) @pytest.mark.timeout(10, method="thread") def test_ddict_get_unknown_key_blocks_forever(): # wait_for_keys=True requires working_set_size > 1 (Dragon enforces it). - d = DDict(n_nodes=1, managers_per_node=1, total_mem=4 * 1024 * 1024, - wait_for_keys=True, working_set_size=2) + d = DDict( + n_nodes=1, + managers_per_node=1, + total_mem=4 * 1024 * 1024, + wait_for_keys=True, + working_set_size=2, + ) try: with pytest.raises(KeyError): _ = d["never-written"] diff --git a/tests/dragon_ci/test_machine_system.py b/tests/dragon_ci/test_machine_system.py index 45ec310..b7e4250 100644 --- a/tests/dragon_ci/test_machine_system.py +++ b/tests/dragon_ci/test_machine_system.py @@ -13,8 +13,8 @@ @pytest.mark.parametrize("attr", ["nnodes", "hostname_policies"]) def test_system_attribute_present(attr): - """Rhapsody reads ``System().nnodes`` (V1 DDict sizing) and calls - ``hostname_policies()`` (telemetry adapter, one worker per node).""" + """Rhapsody reads ``System().nnodes`` (V1 DDict sizing) and calls ``hostname_policies()`` + (telemetry adapter, one worker per node).""" assert hasattr(System(), attr), f"System.{attr} removed" diff --git a/tests/dragon_ci/test_policy.py b/tests/dragon_ci/test_policy.py index 7405806..20aacf1 100644 --- a/tests/dragon_ci/test_policy.py +++ b/tests/dragon_ci/test_policy.py @@ -12,8 +12,13 @@ from dragon.infrastructure.policy import Policy _POLICY_KWARGS = ( - "placement", "host_name", "host_id", "distribution", - "cpu_affinity", "gpu_affinity", "wait_mode", + "placement", + "host_name", + "host_id", + "distribution", + "cpu_affinity", + "gpu_affinity", + "wait_mode", ) diff --git a/tests/dragon_ci/test_process_group.py b/tests/dragon_ci/test_process_group.py index 69f9132..29ff830 100644 --- a/tests/dragon_ci/test_process_group.py +++ b/tests/dragon_ci/test_process_group.py @@ -32,9 +32,7 @@ def test_process_group_ctor_kwarg(kwarg): @pytest.mark.parametrize("method", ["add_process", "init", "start", "join", "stop", "close"]) def test_process_group_method_present(method): - assert callable(getattr(ProcessGroup, method, None)), ( - f"ProcessGroup.{method}() removed" - ) + assert callable(getattr(ProcessGroup, method, None)), f"ProcessGroup.{method}() removed" @pytest.mark.parametrize("kwarg", ["nproc", "template"]) diff --git a/tests/dragon_ci/test_process_template.py b/tests/dragon_ci/test_process_template.py index b48e183..4354f9a 100644 --- a/tests/dragon_ci/test_process_template.py +++ b/tests/dragon_ci/test_process_template.py @@ -22,7 +22,15 @@ from dragon.native.process import ProcessTemplate _TEMPLATE_KWARGS = ( - "target", "args", "kwargs", "cwd", "env", "stdin", "stdout", "stderr", "policy", + "target", + "args", + "kwargs", + "cwd", + "env", + "stdin", + "stdout", + "stderr", + "policy", ) @@ -60,13 +68,17 @@ def test_template_env_dict_preserved(): def test_template_stdio_sentinels_accepted(): """PIPE/DEVNULL must be acceptable to the stdin/stdout/stderr kwargs.""" ProcessTemplate( - "/bin/echo", args=("x",), - stdin=Popen.DEVNULL, stdout=Popen.PIPE, stderr=Popen.PIPE, + "/bin/echo", + args=("x",), + stdin=Popen.DEVNULL, + stdout=Popen.PIPE, + stderr=Popen.PIPE, ) def test_template_argdata_round_trips_via_cloudpickle(): """Rhapsody round-trips ``(target, args, kwargs)`` out of ``pt.argdata``.""" + def _target(x, y): return x + y diff --git a/tests/dragon_ci/test_sigkill_in_worker_hangs.py b/tests/dragon_ci/test_sigkill_in_worker_hangs.py index c0d48d8..cf23693 100644 --- a/tests/dragon_ci/test_sigkill_in_worker_hangs.py +++ b/tests/dragon_ci/test_sigkill_in_worker_hangs.py @@ -37,7 +37,7 @@ @pytest.mark.skip( reason="DISABLED — Dragon bug: SIGKILL'd worker is undetected, task.get() " - "blocks forever, timeout= is ignored. Re-enable to check if fixed." + "blocks forever, timeout= is ignored. Re-enable to check if fixed." ) @pytest.mark.timeout(5, method="thread") def test_sigkill_in_worker_hangs(fresh_batch: Batch): diff --git a/tests/dragon_ci/test_unpickleable_function_hangs.py b/tests/dragon_ci/test_unpickleable_function_hangs.py index 0697f27..eaf4800 100644 --- a/tests/dragon_ci/test_unpickleable_function_hangs.py +++ b/tests/dragon_ci/test_unpickleable_function_hangs.py @@ -43,7 +43,7 @@ @pytest.mark.skip( reason="DISABLED — Dragon bug: worker silently dies during cloudpickle " - "unpickling; task.get() blocks forever. Re-enable to check if fixed." + "unpickling; task.get() blocks forever. Re-enable to check if fixed." ) @pytest.mark.timeout(5, method="thread") def test_unpickleable_function_hangs(fresh_batch: Batch): From 87f14b917f4aa43ca081719d76dc4628cca0080c Mon Sep 17 00:00:00 2001 From: Andre Merzky Date: Tue, 30 Jun 2026 12:56:55 +0200 Subject: [PATCH 5/5] tests/dragon_ci: address gemini review - test_process_group / test_batch_lifecycle: wrap the body in try/finally so a timed-out get() or failed assertion can't leak the worker ProcessGroup / Batch. - conftest: don't append a trailing PYTHONPATH separator when it's empty (an empty entry is implicitly the cwd). - test_batch_job: narrow the skip to RuntimeError + an actual PMIx message, and re-raise everything else so real regressions don't get silently skipped. - test_queue_event: measure the wait with time.monotonic() instead of time.time(). Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/dragon_ci/conftest.py | 5 ++++- tests/dragon_ci/test_batch_job.py | 8 ++++++-- tests/dragon_ci/test_batch_lifecycle.py | 9 ++++++--- tests/dragon_ci/test_process_group.py | 15 +++++++++------ tests/dragon_ci/test_queue_event.py | 4 ++-- 5 files changed, 27 insertions(+), 14 deletions(-) diff --git a/tests/dragon_ci/conftest.py b/tests/dragon_ci/conftest.py index 7700967..e05cb89 100644 --- a/tests/dragon_ci/conftest.py +++ b/tests/dragon_ci/conftest.py @@ -37,7 +37,10 @@ _HERE = os.path.dirname(os.path.abspath(__file__)) if _HERE not in sys.path: sys.path.insert(0, _HERE) -os.environ["PYTHONPATH"] = _HERE + os.pathsep + os.environ.get("PYTHONPATH", "") +# Only append a separator when PYTHONPATH is already populated; a trailing +# separator (empty entry) is implicitly treated as the cwd, which we don't want. +_existing_pp = os.environ.get("PYTHONPATH") +os.environ["PYTHONPATH"] = f"{_HERE}{os.pathsep}{_existing_pp}" if _existing_pp else _HERE # --- Dragon must be importable ------------------------------------------- diff --git a/tests/dragon_ci/test_batch_job.py b/tests/dragon_ci/test_batch_job.py index 478fadf..2dc7fcb 100644 --- a/tests/dragon_ci/test_batch_job.py +++ b/tests/dragon_ci/test_batch_job.py @@ -45,8 +45,12 @@ def test_batch_job_launch(batch: Batch, nranks): pmi=PMIBackend.PMIX, ) job.get(timeout=120.0) - except Exception as exc: # noqa: BLE001 - pytest.skip(f"PMIX launch unavailable on this host: {exc!r}") + except RuntimeError as exc: + # Only skip for a genuine PMIx-unavailable host; let other RuntimeErrors + # (and any TypeError/AttributeError/etc.) surface as real failures. + if "pmi" in str(exc).lower(): + pytest.skip(f"PMIX launch unavailable on this host: {exc!r}") + raise assert isinstance(job, Job) result, tb, raised, _stdout, _stderr = batch.results_ddict[job.uid] diff --git a/tests/dragon_ci/test_batch_lifecycle.py b/tests/dragon_ci/test_batch_lifecycle.py index 45c3c98..6dcf857 100644 --- a/tests/dragon_ci/test_batch_lifecycle.py +++ b/tests/dragon_ci/test_batch_lifecycle.py @@ -62,6 +62,9 @@ def test_batch_close_join_terminate_lifecycle(): is deprecated as of Dragon 0.14 (no-op); only ``join()`` is real teardown. """ b = Batch(disable_telem=True) - b.close() - b.join(timeout=30.0) - b.terminate() + try: + b.close() + b.join(timeout=30.0) + finally: + # Always terminate so a raising/timed-out join() can't leak the Batch. + b.terminate() diff --git a/tests/dragon_ci/test_process_group.py b/tests/dragon_ci/test_process_group.py index 29ff830..bbb5a77 100644 --- a/tests/dragon_ci/test_process_group.py +++ b/tests/dragon_ci/test_process_group.py @@ -58,9 +58,12 @@ def test_process_group_lifecycle_smoke(): grp.init() grp.start() - msg = q.get(timeout=30.0) - assert msg.get("pid") and msg.get("host") - - shutdown.set() - grp.join(timeout=30.0) - grp.close() + try: + msg = q.get(timeout=30.0) + assert msg.get("pid") and msg.get("host") + finally: + # Guarantee the worker is told to stop and the group is reaped even if + # q.get() times out or the assertion fails, else pg_worker loops forever. + shutdown.set() + grp.join(timeout=30.0) + grp.close() diff --git a/tests/dragon_ci/test_queue_event.py b/tests/dragon_ci/test_queue_event.py index e7dbd2e..621a1b2 100644 --- a/tests/dragon_ci/test_queue_event.py +++ b/tests/dragon_ci/test_queue_event.py @@ -48,10 +48,10 @@ def test_event_wait_returns_truthy_when_set(): def test_event_wait_returns_falsy_on_timeout(): e = Event() - t0 = time.time() + t0 = time.monotonic() rv = e.wait(timeout=0.2) assert not rv, f"Event.wait should be falsy on timeout, got {rv!r}" - assert time.time() - t0 >= 0.15, "Event.wait returned before its timeout" + assert time.monotonic() - t0 >= 0.15, "Event.wait returned before its timeout" def test_queue_event_cross_process_roundtrip():