tests/dragon_ci: standalone pre-release contract suite for Dragon

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.

tests/dragon_ci/_dragon_ci_helpers.py

@@ -0,0 +1,78 @@
+"""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):
+    # 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):
+    sys.exit(1)
+
+
+def fn_kill_self(_x):
+    """Abnormal worker termination — no chance for Dragon's exit-detector."""
+    os.kill(os.getpid(), signal.SIGKILL)

tests/dragon_ci/_offpath/marker_helper.py

@@ -0,0 +1,10 @@
+"""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

tests/dragon_ci/conftest.py

@@ -0,0 +1,196 @@
+"""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)
+# 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 -------------------------------------------
+
+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")
+
+
+# --- 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)
+_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

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