Add hardware nPlay/Hub2 clock-sync stress tests + fixture hardening

pycbsdk/pyproject.toml

@@ -49,6 +49,8 @@ lint = [
 test = [
     "pytest",
     "pytest-timeout",
+    "numpy",
+    "pylsl",
 ]
 
 [tool.setuptools.packages.find]
@@ -60,6 +62,7 @@ pycbsdk = ["*.dll", "*.so", "*.so.*", "*.dylib"]
 [tool.pytest.ini_options]
 markers = [
     "integration: tests requiring a running nPlayServer instance",
+    "lsl: nPlay+LSL tests; need an EXCLUSIVE nPlayServer (one shmem per device type). Auto-skipped when file-based nPlay tests are also collected — run with `pytest -m lsl`.",
 ]
 
 [tool.setuptools_scm]

pycbsdk/tests/_lsl_generator.py

@@ -0,0 +1,72 @@
+"""Standalone 30 kHz int16 LSL stream generator for nPlay+LSL tests.
+
+Launched as a subprocess by the ``nplay_lsl`` fixture.  nPlayServer
+(``--device=LSL <name>``) resolves this stream by name and replays it onto the
+Cerebus protocol, deriving each packet's device timestamp from the stream's
+ORIGINATING time scaled to a 30 kHz sample counter.  We therefore pace pushes to
+real time so the originating timestamps advance at the true sample rate — the
+condition CereLink's clock sync must map to a monotonic host timeline.
+
+Usage:
+    python _lsl_generator.py <stream_name> [n_channels]
+
+Runs until killed (SIGTERM/SIGINT) by the fixture.
+"""
+
+import signal
+import sys
+import time
+
+import numpy as np
+import pylsl
+
+SRATE = 30000
+CHUNK_SAMPLES = 300  # 10 ms per push at 30 kHz
+
+
+def main() -> int:
+    name = sys.argv[1]
+    n_ch = int(sys.argv[2]) if len(sys.argv) > 2 else 8
+
+    info = pylsl.StreamInfo(
+        name=name,
+        type="EEG",
+        channel_count=n_ch,
+        nominal_srate=SRATE,
+        channel_format=pylsl.cf_int16,
+        source_id=f"{name}_src",
+    )
+    outlet = pylsl.StreamOutlet(info, chunk_size=CHUNK_SAMPLES, max_buffered=360)
+
+    # Exit cleanly on the fixture's terminate() so liblsl tears the outlet down.
+    running = {"go": True}
+
+    def _stop(*_a):
+        running["go"] = False
+
+    signal.signal(signal.SIGTERM, _stop)
+    signal.signal(signal.SIGINT, _stop)
+
+    # A slow sine across channels — content is irrelevant to the timing test,
+    # but non-constant data makes the stream easy to eyeball in a viewer.
+    phase = np.arange(CHUNK_SAMPLES, dtype=np.float64)
+    chan_offsets = np.linspace(0.0, np.pi, n_ch, dtype=np.float64)
+
+    t0 = time.monotonic()
+    idx = 0
+    while running["go"]:
+        angle = 2.0 * np.pi * (phase[:, None] + idx) / 3000.0 + chan_offsets[None, :]
+        chunk = (np.sin(angle) * 1000.0).astype(np.int16)
+        outlet.push_chunk(chunk)
+        idx += CHUNK_SAMPLES
+        # Pace to real time: sample idx should be emitted at t0 + idx/SRATE.
+        target = t0 + idx / SRATE
+        dt = target - time.monotonic()
+        if dt > 0:
+            time.sleep(dt)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

pycbsdk/tests/conftest.py

@@ -17,7 +17,7 @@
 
 import pytest
 
-CERELINK_RELEASE_URL = "https://github.com/CerebusOSS/CereLink/releases/download/v9.3.0"
+CERELINK_RELEASE_URL = "https://github.com/CerebusOSS/CereLink/releases/download/v9.11.0"
 CACHE_DIR = Path(__file__).parent / ".test_cache"
 
 
@@ -41,6 +41,89 @@ def _sem_unlink(name: str) -> None:
     libc.sem_unlink(sem_name)  # ignore errors (ENOENT is fine)
 
 
+def _remove_lockfile(name: str) -> None:
+    """Remove nPlayServer's lock FILE ($TMPDIR/<name>.lock).
+
+    nPlayServer's lock is a file guarded by O_EXCL (not a POSIX semaphore), so a
+    hard-killed instance leaves it behind — which makes the next launch with the
+    same name fail with "Cannot run multiple instances".  Safe when absent.
+    """
+    if platform.system() == "Windows":
+        return
+    tmp = os.environ.get("TMPDIR") or "/tmp"
+    (Path(tmp.rstrip("/")) / f"{name}.lock").unlink(missing_ok=True)
+
+
+def pytest_configure(config):
+    """Fail fast if the DEV-built libcbsdk is OLDER than the C++ sources.
+
+    In a source checkout pycbsdk loads ``cmake-build-*/src/cbsdk/libcbsdk.dylib``
+    (there is no wheel-bundled lib).  If that dylib was not rebuilt after a C++
+    change, the tests silently run STALE code — which once masked a clock-sync
+    fix for an entire debugging session.  Only a build-tree dylib is checked (an
+    installed/wheel-bundled lib has no sources to compare against); bypass with
+    ``PYCBSDK_ALLOW_STALE_LIB=1``.
+    """
+    if os.environ.get("PYCBSDK_ALLOW_STALE_LIB"):
+        return
+    try:
+        from pycbsdk._lib import _find_library
+        lib_path = Path(_find_library())
+    except Exception:
+        return  # can't resolve the lib — let the normal load path report it
+    lib_str = str(lib_path)
+    if "cmake-build" not in lib_str and f"{os.sep}build{os.sep}" not in lib_str:
+        return  # installed/bundled lib, not a dev build tree
+    if not lib_path.exists():
+        return
+    repo_root = Path(__file__).resolve().parents[2]
+    src_root = repo_root / "src"
+    if not src_root.exists():
+        return
+    lib_mtime = lib_path.stat().st_mtime
+    newest, newest_file = lib_mtime, None
+    for f in src_root.rglob("*"):
+        if f.suffix in (".cpp", ".h", ".hpp", ".c", ".cc") and f.is_file():
+            m = f.stat().st_mtime
+            if m > newest:
+                newest, newest_file = m, f
+    if newest_file is not None:
+        raise pytest.UsageError(
+            f"Stale libcbsdk: {lib_path}\n"
+            f"  built {time.ctime(lib_mtime)} is OLDER than C++ source "
+            f"{newest_file.relative_to(repo_root)} ({time.ctime(newest)}).\n"
+            f"  Tests would run pre-change code. Rebuild the dev dylib:\n"
+            f"    cmake --build {lib_path.parents[2]} --target cbsdk_shared\n"
+            f"  (or set PYCBSDK_ALLOW_STALE_LIB=1 to bypass this check)."
+        )
+
+
+@pytest.hookimpl(trylast=True)
+def pytest_collection_modifyitems(config, items):
+    """Auto-skip ``lsl`` tests when file-based nPlay tests are also collected.
+
+    Both bring up a STANDALONE ``Session(DeviceType.NPLAY)`` whose shared memory
+    is keyed by device type, so two nPlayServer instances cannot coexist.  When a
+    single run collects both, skip the LSL ones with guidance to run them alone.
+
+    Runs ``trylast`` so pytest's own ``-m``/``-k`` deselection happens FIRST:
+    under ``pytest -m lsl`` the file-based tests are already removed from
+    ``items``, so the LSL tests are NOT skipped and actually run.
+    """
+    lsl_items = [it for it in items if it.get_closest_marker("lsl")]
+    nplay_file_items = [
+        it for it in items
+        if it.get_closest_marker("integration") and not it.get_closest_marker("lsl")
+    ]
+    if lsl_items and nplay_file_items:
+        skip = pytest.mark.skip(
+            reason="LSL nPlay tests need an exclusive nPlayServer "
+                   "(one shmem per device type); run separately: pytest -m lsl"
+        )
+        for it in lsl_items:
+            it.add_marker(skip)
+
+
 def _nplay_asset_name() -> str | None:
     """Return the nPlayServer zip asset name for this platform, or None."""
     system = platform.system()
@@ -162,11 +245,29 @@ def nplayserver_binary() -> Path | None:
     for candidate in extract_dir.rglob(f"nPlayServer{exe_suffix}"):
         if candidate.is_file():
             if platform.system() != "Windows":
-                candidate.chmod(0o755)
+                _ensure_executable(candidate)
             return candidate
     return None
 
 
+def _ensure_executable(path: Path) -> None:
+    """Make *path* executable, tolerating a transient EPERM on macOS.
+
+    A freshly-downloaded SIGNED binary is briefly locked by Gatekeeper /
+    syspolicyd on first access, so chmod can raise PermissionError.  The file is
+    usually already 0755 (preserved by the archive), so we skip when it's already
+    executable and otherwise retry a few times before giving up.
+    """
+    for _ in range(5):
+        if os.access(path, os.X_OK):
+            return
+        try:
+            path.chmod(0o755)
+            return
+        except PermissionError:
+            time.sleep(0.5)
+
+
 @pytest.fixture(scope="session")
 def nplayserver(nplayserver_binary: Path | None, ns6_path: Path):
     """Start nPlayServer playing back test data; kill on teardown.
@@ -244,3 +345,105 @@ def client_session(nplay_session):
             "is the nplay_session fixture alive?"
         )
         yield session
+
+
+# ---------------------------------------------------------------------------
+# nPlay + LSL fixtures (mutually exclusive with the file-playback fixtures above;
+# see pytest_collection_modifyitems).  Run with: pytest -m lsl
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(scope="session")
+def lsl_stream_name() -> str:
+    """Unique LSL stream name for this test process."""
+    return f"pycbsdk_lsl_{os.getpid()}"
+
+
+@pytest.fixture(scope="session")
+def lsl_generator(lsl_stream_name: str):
+    """Launch the 30 kHz int16 LSL generator and wait until its stream resolves.
+
+    Skips dependent tests if pylsl is unavailable.  Yields the stream name.
+    """
+    pylsl = pytest.importorskip("pylsl", reason="pylsl required for LSL tests")
+
+    gen_script = Path(__file__).parent / "_lsl_generator.py"
+    proc = subprocess.Popen(
+        [sys.executable, str(gen_script), lsl_stream_name, "8"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+    )
+
+    streams = pylsl.resolve_byprop("name", lsl_stream_name, timeout=10)
+    if not streams:
+        proc.terminate()
+        proc.wait()
+        out = proc.stdout.read().decode(errors="replace") if proc.stdout else ""
+        pytest.fail(f"LSL generator stream '{lsl_stream_name}' never appeared\n{out}")
+
+    yield lsl_stream_name
+
+    proc.terminate()
+    try:
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+
+
+@pytest.fixture(scope="session")
+def nplay_lsl(nplayserver_binary: Path | None, lsl_generator: str):
+    """Start nPlayServer as an LSL inlet reading the generator stream.
+
+    nPlayServer derives device timestamps from the stream's originating time
+    (× 30 kHz), so this exercises the LSL clock path rather than file playback.
+    """
+    if nplayserver_binary is None:
+        pytest.skip(
+            f"nPlayServer not available for {platform.system()}/{platform.machine()}"
+        )
+
+    lock_name = f"nplay_lsl_{os.getpid()}"
+    _remove_lockfile(lock_name)  # clear any stale lock from a hard-killed prior run
+
+    proc = subprocess.Popen(
+        [
+            str(nplayserver_binary),
+            "--audio", "none",
+            "--lockfile", lock_name,
+            "--device=LSL", lsl_generator,
+            "-A",  # autostart
+        ],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+    # Allow time to resolve the LSL inlet and bind its UDP ports.
+    time.sleep(3)
+
+    if proc.poll() is not None:
+        stdout = proc.stdout.read().decode(errors="replace") if proc.stdout else ""
+        stderr = proc.stderr.read().decode(errors="replace") if proc.stderr else ""
+        _remove_lockfile(lock_name)
+        pytest.fail(
+            f"nPlayServer --device=LSL exited immediately (rc={proc.returncode})\n"
+            f"stdout: {stdout}\nstderr: {stderr}"
+        )
+
+    yield proc
+
+    proc.terminate()
+    try:
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait()
+    _remove_lockfile(lock_name)
+
+
+@pytest.fixture(scope="session")
+def nplay_lsl_session(nplay_lsl):
+    """A STANDALONE Session connected to the LSL-fed nPlayServer."""
+    from pycbsdk import DeviceType, Session
+
+    with Session(DeviceType.NPLAY) as session:
+        yield session