Implement testing suite: pytest scaffolding, unit tests for existing pure logic, CI workflow

.claude/skills/aetherscan-repo-context/SKILL.md

@@ -114,7 +114,10 @@ utils/                   # fetch_run_outputs.sh, find_optimal_configs.py,
                          # run_container.sh, start_tmux_session.sh,
                          # verify_train_test_files.py
 docs/                    # GPU_RUNTIME_GUIDE.md, CONFIG_AND_CLI.md, README.md, assets/
-tests/                   # Placeholder — no test suite yet
+tests/                   # Pytest suite: conftest.py (singleton-reset fixtures, synthetic
+                         # data factories), unit/, integration/ (gpu+cluster-marked smokes).
+                         # Default selection: pytest -m "not gpu and not cluster" -q
+                         # (runs in CI via .github/workflows/tests.yml); see CONTRIBUTING.md
 ```
 
 ---

.github/workflows/tests.yml

@@ -0,0 +1,57 @@
+# Unit test suite (pytest)
+name: Tests
+
+# Run on every PR and on pushes to master
+on:
+  push:
+    branches: ["master"]
+  pull_request:
+    branches: ["**"]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  tests:
+    runs-on: ubuntu-latest
+    strategy:
+      # Let both Python versions report rather than cancelling the sibling on first failure
+      fail-fast: false
+      matrix:
+        # The two supported runtimes: conda env (3.10) and NGC container (3.12)
+        python-version: ["3.10", "3.12"]
+
+    steps:
+      # Pull repo contents into GitHub Actions runner
+      - name: Checkout code
+        uses: actions/checkout@v5
+
+      # Install the matrix Python version, with built-in pip caching keyed on the
+      # dependency pins (invalidates when requirements-container.txt changes)
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+          cache: "pip"
+          # Include the workflow itself so the inline pins below also invalidate the cache
+          cache-dependency-path: |
+            requirements-container.txt
+            .github/workflows/tests.yml
+
+      # Mirror the NGC container's dependency surface on a CPU-only runner:
+      # tensorflow-cpu stands in for the container's GPU TF 2.17 build, the pinned
+      # extras come from requirements-container.txt, and h5py/hdf5plugin/pandas/
+      # psutil/pytest are installed explicitly because the NGC base image ships
+      # them (so the requirements file intentionally omits them; hdf5plugin is
+      # pinned explicitly in environment.yml too — don't rely on setigen pulling
+      # it transitively)
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install "tensorflow-cpu==2.17.*" -r requirements-container.txt h5py hdf5plugin pandas psutil pytest
+
+      # GPU- and cluster-marked tests need physical GPUs / cluster-resident data;
+      # everything else (including slow CPU TF graph tests) runs here
+      - name: Run test suite
+        run: pytest -m "not gpu and not cluster" -q

CLAUDE.md

@@ -4,7 +4,7 @@ Always-on, bare-essentials rules for any coding agent in this repo. The full dee
 
 ## Project
 
-Aetherscan: Breakthrough Listen's deep-learning SETI pipeline. Two-stage ML (Beta-VAE → Random Forest), single-node multi-GPU distributed train/inference. Sole entry point: `src/aetherscan/main.py`. `tests/` is a placeholder — **no test suite yet**, so don't claim `pytest` passes; verify changes another way and say how.
+Aetherscan: Breakthrough Listen's deep-learning SETI pipeline. Two-stage ML (Beta-VAE → Random Forest), single-node multi-GPU distributed train/inference. Sole entry point: `src/aetherscan/main.py`. Pytest suite lives in `tests/` (unit + gpu/cluster-marked integration; see [CONTRIBUTING.md](CONTRIBUTING.md#testing)) — run it before claiming a change works, and ship unit tests with new logic. Most test modules import TensorFlow at collection; if the full dependency stack isn't available in your environment, run the subset you can and say exactly what you ran.
 
 ## Run / lint
 
@@ -15,6 +15,8 @@ Aetherscan: Breakthrough Listen's deep-learning SETI pipeline. Two-stage ML (Bet
 PYTHONPATH=src python -m aetherscan.main {train|inference} --save-tag final_v1
 # Lint + format (also enforced via pre-commit)
 ruff check src/ && ruff format src/
+# Tests (default selection = what CI runs; gpu/cluster-marked tests need a cluster)
+pytest -m "not gpu and not cluster" -q
 ```
 
 ## Hard rules (don't break these)

CONTRIBUTING.md

@@ -190,7 +190,10 @@ Aetherscan/
 │       ├── __init__.py         # Manager exports
 │       └── manager.py          # Resource lifecycle management
 ├── docs/                       # Documentation (placeholder; no docs yet)
-├── tests/                      # Test suite (placeholder; no tests yet)
+├── tests/                      # Pytest suite (see Testing section below)
+│   ├── conftest.py                  # Singleton-reset fixtures, tmp paths, synthetic data factories
+│   ├── unit/                        # Fast, hardware-independent unit tests (run in CI)
+│   └── integration/                 # gpu+cluster-marked end-to-end smoke tests
 ├── utils/                      # Utility scripts
 │   ├── fetch_run_outputs.sh         # rsync a run's outputs from a remote node
 │   ├── find_optimal_configs.py      # Per-host config helper
@@ -510,9 +513,43 @@ Prefer descriptive, full-word names: `num_training_rounds`, `per_replica_batch_s
 
 ## Testing
 
-> [!WARNING]
->
-> # TODO: update this section once test suite is operational
+The test suite lives in `tests/` and runs on [pytest](https://docs.pytest.org/) (configured in `pyproject.toml` under `[tool.pytest.ini_options]` — `pythonpath = ["src"]` makes `import aetherscan` work without a `PYTHONPATH` prefix).
+
+### Running the suite
+
+```bash
+# Default selection — everything that doesn't need GPUs or cluster-resident data.
+# This is exactly what CI runs (.github/workflows/tests.yml).
+pytest -m "not gpu and not cluster" -q
+
+# Cluster integration smokes (end-to-end train + CSV inference), inside the NGC container:
+./utils/run_container.sh python -m pytest tests/ -m "gpu or cluster" -q
+
+# A single file / test while iterating:
+pytest tests/unit/test_config.py -q
+pytest tests/unit/test_db.py::TestFlushSentinel -q
+```
+
+### Markers
+
+| Marker        | Meaning                                                                                     |
+| ------------- | ------------------------------------------------------------------------------------------- |
+| `slow`        | Slower tests (e.g. builds TF graphs on CPU); still included in the default CI selection      |
+| `gpu`         | Requires one or more physical GPUs; excluded from CI                                         |
+| `cluster`     | Requires cluster-resident data/models (blpc3/bla0); excluded from CI                         |
+| `integration` | End-to-end pipeline runs; skips the per-test singleton/env isolation fixture in `conftest.py` |
+
+### Test environment & fixtures
+
+- Every non-integration test runs isolated: `tests/conftest.py` points `AETHERSCAN_{DATA,MODEL,OUTPUT}_PATH` at a pytest `tmp_path`, resets all singletons (`Config`, `ResourceManager`, `Database`, `Logger`, `ResourceMonitor`) via their `_reset()` hooks, and re-runs `init_config()` — so tests never touch real data or leak state into each other.
+- Synthetic data factories are provided as fixtures: `make_background_npy` (tiny background plates), `make_h5_observation` (tiny filterbank-style `.h5` files), and `make_inference_csv` (tiny cadence-grouping CSVs).
+- CI (`.github/workflows/tests.yml`) runs the default selection on Ubuntu with `tensorflow-cpu==2.17.*` + the pins from `requirements-container.txt`, on Python 3.10 and 3.12 (the conda and container runtimes respectively). A few tests assert Linux-only behavior (e.g. PSS-based memory accounting) and self-skip elsewhere.
+
+### Writing tests
+
+- Every PR that adds or changes logic should ship unit tests for it (`tests/unit/`), reusing the conftest fixtures rather than instantiating singletons directly.
+- Mark anything that needs hardware or cluster data with the appropriate marker so the default selection stays runnable on a laptop and in CI.
+- Test style follows the same ruff lint+format rules as `src/` (pre-commit runs on `tests/` too).
 
 ---
 

pyproject.toml

@@ -55,6 +55,19 @@ Issues = "https://github.com/zachtheyek/Aetherscan/issues"
 [project.optional-dependencies]
 dev = ["ruff"]  # Python linter
 
+[tool.pytest.ini_options]
+# Make `import aetherscan` work without a PYTHONPATH=src prefix
+pythonpath = ["src"]
+testpaths = ["tests"]
+# Reject undeclared markers so a typo'd @pytest.mark.gpu can't silently slip into CI
+addopts = "--strict-markers"
+markers = [
+    "slow: slower tests (e.g. builds TF graphs on CPU); included in the default CI selection",
+    "gpu: requires one or more physical GPUs; excluded from CI (-m 'not gpu and not cluster')",
+    "cluster: requires cluster-resident data/models (blpc3/bla0); excluded from CI",
+    "integration: end-to-end pipeline runs (skips the per-test singleton/env isolation fixture)",
+]
+
 [tool.ruff]
 # Configure linter
 target-version = "py310"   # Target Python version

tests/conftest.py

@@ -0,0 +1,223 @@
+"""
+Shared pytest fixtures for the Aetherscan test suite.
+
+Provides three things:
+1. An autouse fixture that isolates every (non-integration) test: AETHERSCAN_{DATA,MODEL,OUTPUT}_PATH
+   point at tmp_path, all singletons are reset via their _reset() hooks, and a fresh Config is
+   initialized. Teardown stops any started background threads/pools and resets everything again.
+2. Synthetic data factories: tiny .npy background plates, tiny .h5 observation files, and tiny
+   inference CSVs — sized for speed, shaped like the real thing.
+3. A headless matplotlib backend for CI (set before any aetherscan module imports pyplot).
+"""
+
+# TODO: integration tests that verify clean shutdown under adverse conditions
+#       (SIGTERM/SIGINT mid-run, resource cleanup) — carried over from the old tests/placeholder
+# TODO: build tests that tagged releases deploy properly to various environments
+#       (PyPI, HuggingFace, etc.) once release engineering lands — carried over from placeholder
+
+from __future__ import annotations
+
+import contextlib
+import csv
+import os
+import signal
+import sys
+
+import numpy as np
+import pytest
+
+# Force a non-interactive backend before any test imports matplotlib.pyplot
+# (train.py imports it at module level; CI runners have no display).
+os.environ.setdefault("MPLBACKEND", "Agg")
+# Quiet TF's C++ INFO/WARNING chatter in test output.
+os.environ.setdefault("TF_CPP_MIN_LOG_LEVEL", "2")
+
+
+def _reset_all_singletons():
+    """Reset every singleton class via its _reset() teardown hook.
+
+    Imports are deferred so integration runs (which skip the isolation fixture entirely)
+    never pull TensorFlow into the pytest parent process.
+    """
+    from aetherscan.config import Config  # noqa: PLC0415
+    from aetherscan.db.db import Database  # noqa: PLC0415
+    from aetherscan.logger.logger import Logger  # noqa: PLC0415
+    from aetherscan.manager.manager import ResourceManager  # noqa: PLC0415
+    from aetherscan.monitor.monitor import ResourceMonitor  # noqa: PLC0415
+
+    for cls in (Config, Database, Logger, ResourceManager, ResourceMonitor):
+        cls._reset()
+
+
+def _teardown_singletons():
+    """Stop background resources still held by singletons, then reset them all.
+
+    Mirrors ResourceManager.cleanup_all()'s ordering (pools -> shared memory -> monitor ->
+    db -> logger) without its logging side effects, so a test that leaked a resource can't
+    poison the next test.
+    """
+    from aetherscan.db.db import Database  # noqa: PLC0415
+    from aetherscan.logger.logger import Logger  # noqa: PLC0415
+    from aetherscan.manager.manager import ResourceManager  # noqa: PLC0415
+    from aetherscan.monitor.monitor import ResourceMonitor  # noqa: PLC0415
+
+    manager = ResourceManager._instance
+    if manager is not None:
+        for managed_pool in list(manager._pools):
+            with contextlib.suppress(Exception):
+                managed_pool.close(timeout=5.0)
+        for managed_shm in list(manager._shared_memories):
+            with contextlib.suppress(Exception):
+                managed_shm.close()
+        # Each ResourceManager construction registers an atexit callback; unregister it so
+        # per-test instances don't accumulate stale cleanup hooks over the session.
+        import atexit  # noqa: PLC0415
+
+        atexit.unregister(manager.cleanup_all)
+
+    monitor = ResourceMonitor._instance
+    if monitor is not None:
+        with contextlib.suppress(Exception):
+            monitor.stop()
+
+    db = Database._instance
+    if db is not None:
+        with contextlib.suppress(Exception):
+            db.stop()
+
+    logger_instance = Logger._instance
+    if logger_instance is not None:
+        with contextlib.suppress(Exception):
+            logger_instance.stop()
+
+    _reset_all_singletons()
+
+
+@pytest.fixture(autouse=True)
+def aetherscan_isolated_env(request, tmp_path, monkeypatch):
+    """Autouse: fresh singletons + tmp-path-scoped env for every test.
+
+    Integration tests are exempt (marker `integration`): they exercise the real pipeline as a
+    subprocess and must inherit the real AETHERSCAN_* environment, and their pytest parent
+    process should never import TensorFlow.
+    """
+    if request.node.get_closest_marker("integration"):
+        yield
+        return
+
+    data_path = tmp_path / "data"
+    model_path = tmp_path / "models"
+    output_path = tmp_path / "outputs"
+    for sub in ("training", "testing", "inference"):
+        (data_path / sub).mkdir(parents=True, exist_ok=True)
+    model_path.mkdir(exist_ok=True)
+    output_path.mkdir(exist_ok=True)
+
+    monkeypatch.setenv("AETHERSCAN_DATA_PATH", str(data_path))
+    monkeypatch.setenv("AETHERSCAN_MODEL_PATH", str(model_path))
+    monkeypatch.setenv("AETHERSCAN_OUTPUT_PATH", str(output_path))
+    # Keep unit tests from ever talking to Slack, whatever the host env holds.
+    monkeypatch.delenv("SLACK_BOT_TOKEN", raising=False)
+    monkeypatch.delenv("SLACK_CHANNEL", raising=False)
+
+    # ResourceManager's constructor installs SIGINT/SIGTERM handlers; snapshot and restore
+    # them so per-test instances don't leave stale bound-method handlers behind.
+    original_sigint = signal.getsignal(signal.SIGINT)
+    original_sigterm = signal.getsignal(signal.SIGTERM)
+    original_stdout, original_stderr = sys.stdout, sys.stderr
+
+    from aetherscan.config import init_config  # noqa: PLC0415
+
+    _reset_all_singletons()
+    init_config()
+
+    yield
+
+    _teardown_singletons()
+    signal.signal(signal.SIGINT, original_sigint)
+    signal.signal(signal.SIGTERM, original_sigterm)
+    sys.stdout, sys.stderr = original_stdout, original_stderr
+
+
+@pytest.fixture
+def make_background_npy(tmp_path):
+    """Factory for tiny .npy background plates shaped like real training backgrounds.
+
+    Returns a callable(filename, n_cadences=8, width_bin=512) -> Path that writes a positive
+    float32 array of shape (n_cadences, 6, 16, width_bin) under the tmp data/training dir.
+    """
+
+    def _make(filename="backgrounds.npy", n_cadences=8, width_bin=512):
+        rng = np.random.default_rng(11)
+        # Chi-squared-ish positive noise, loosely resembling detected power spectra.
+        plate = rng.chisquare(df=4, size=(n_cadences, 6, 16, width_bin)).astype(np.float32)
+        path = tmp_path / "data" / "training" / filename
+        path.parent.mkdir(parents=True, exist_ok=True)
+        np.save(path, plate)
+        return path
+
+    return _make
+
+
+@pytest.fixture
+def make_h5_observation(tmp_path):
+    """Factory for tiny .h5 observation files matching the filterbank-style layout.
+
+    Returns a callable(filename, n_chans=2048, time_bins=16) -> Path that writes an .h5 with a
+    'data' dataset of shape (time_bins, 1, n_chans) plus fch1/foff/nchans attrs (plain h5py,
+    no bitshuffle).
+    """
+    import h5py  # noqa: PLC0415
+
+    def _make(filename="obs.h5", n_chans=2048, time_bins=16):
+        rng = np.random.default_rng(7)
+        data = rng.chisquare(df=4, size=(time_bins, 1, n_chans)).astype(np.float32)
+        path = tmp_path / filename
+        with h5py.File(path, "w") as hf:
+            dset = hf.create_dataset("data", data=data)
+            dset.attrs["fch1"] = 8421.386717353016
+            dset.attrs["foff"] = -2.7939677238464355e-06
+            dset.attrs["nchans"] = n_chans
+        return path
+
+    return _make
+
+
+@pytest.fixture
+def make_inference_csv(tmp_path):
+    """Factory for tiny inference CSVs in the cadence-grouping layout.
+
+    Returns a callable(filename, groups) -> Path. `groups` is a list of (key_dict, h5_paths)
+    pairs; each h5 path becomes one row carrying the group's key columns. The h5-path column
+    name is read from InferenceConfig.cadence_h5_path_col on the initialized singleton, and
+    the default group keys mirror InferenceConfig.cadence_group_by_cols.
+    """
+    from aetherscan.config import get_config  # noqa: PLC0415
+
+    def _make(filename="subset.csv", groups=None):
+        if groups is None:
+            groups = [
+                (
+                    {
+                        "Target": "HIP110750",
+                        "Session": "AGBT21B_999_31",
+                        "Band": "L",
+                        "Cadence ID": "0",
+                        "Frequency": "1400",
+                    },
+                    [f"/data/obs_{i}.h5" for i in range(6)],
+                )
+            ]
+        h5_col = get_config().inference.cadence_h5_path_col
+        fieldnames = list(groups[0][0].keys()) + [h5_col]
+        path = tmp_path / "data" / "inference" / filename
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with open(path, "w", newline="") as f:
+            writer = csv.DictWriter(f, fieldnames=fieldnames)
+            writer.writeheader()
+            for key_dict, h5_paths in groups:
+                for h5 in h5_paths:
+                    writer.writerow({**key_dict, h5_col: h5})
+        return path
+
+    return _make