Integration Tests

End-to-end tests that exercise small but realistic generation paths.

Warning

Mixed maturity. Most examples on this page are proposed contracts and may change before their code lands. The HIL execution contract below is sourced from the implemented Layer 44 runner and its focused test suite.

Integration tests

End-to-end small datasets verifying multiple layers compose correctly.

HIL execution contract

tests/unit/test_layer44_hil_execution.py is the focused contract suite for the HIL runner in src/rfgen/hil_execution.py. It runs against controlled backend results and artifact stores so that development machines need no radio hardware. It verifies the security and data-correctness boundary rather than claiming that a local fake is a physical test.

Contract exercised locally

Required result

Plan admission

Strict, self-hashed plan fields reject invalid physical limits, expired or too-short operator authorization, and deadlines beyond the phase caps.

Lifecycle and replay

Only the documented NEW through CLEANING transitions are legal; phase nonces replay byte-identical results only for identical requests.

Hardware identity

A report cannot pass without a configured pinned trust root and a correctly signed, plan-bound hardware attestation in the calibration evidence.

Safety and metrics

The suite requires the seven safety checks and exactly five named metric thresholds; a negative non-EVM error magnitude is rejected, while negative EVM remains valid.

Artifact binding

The six artifact kinds occur once in normative order, hashes verify, semantic records bind the plan and run, and SigMF capture fields agree with the plan.

Failure and cleanup

UNAVAILABLE is allowed only before acquisition; acquired failures run all five cleanup actions and a missing, failed, RF-enabled, or timed-out cleanup cannot pass.

The suite also checks that each public phase checks the operator authorization against its supplied deadline before dispatching a replayed result or a backend operation. This prevents an expired authorization from being hidden by a cached response.

Hardware qualification limitation

No local integration test can establish that a radio transmitted safely, captured a real waveform, or was calibrated at a site. The shipped runner does not include a pretend UHD or GNU Radio adapter that yields PASS. A qualification claim instead needs an approved lab execution with a pinned site trust root, root-signed hardware attestation, the six retained artifacts, passing safety and metric records, and at least 20 passes from each of two distinct serial numbers. Radiated testing additionally requires the separately approved lab policy.

test_quickstart.py

def test_quickstart_completes(tmp_path):
    """100-sample narrowband_classifier_baseline run completes without error."""
    import subprocess
    output = tmp_path / "quickstart"
    result = subprocess.run([
        "rfgen", "generate",
        "+preset=narrowband_classifier_baseline",
        "run.num_samples=100",
        f"storage.path={output}",
    ], capture_output=True, text=True, check=True)
    assert (output / "samples.zarr").exists()
    assert (output / "manifest.json").exists()

test_torchsig_roundtrip.py

def test_hdf5_roundtrip_byte_exact(tmp_path):
    """Generate → write HDF5 → read with TorchSig loader → round-trip back."""
    from torchsig.utils.dataset import StaticTorchSigDataset
    from rfgen.storage.hdf5_store import HDF5Store

    # Generate via rfgen
    rfgen.generate("+preset=wideband_baseline_xs", output=tmp_path)

    # Read with TorchSig
    ts_ds = StaticTorchSigDataset(root=str(tmp_path / "hdf5"))
    rfgen_ds = HDF5Store().open(str(tmp_path / "hdf5"), mode="r")

    for ts_sample, rfgen_sample in zip(ts_ds, rfgen_ds):
        # IQ must be byte-exact
        assert np.array_equal(ts_sample.iq, rfgen_sample.iq.numpy())
        # Bbox fields TorchSig understands must round-trip
        for ts_bb, rg_bb in zip(ts_sample.metadata, rfgen_sample.bboxes):
            assert ts_bb.lower_freq == rg_bb.abs.low_freq_hz
            assert ts_bb.start_in_samples == rg_bb.abs.start_sample

test_reproducibility.py

def test_byte_exact_reproduction(tmp_path):
    """Same config + same seed → byte-identical output."""
    out1 = tmp_path / "run1"
    out2 = tmp_path / "run2"
    cmd_template = [
        "rfgen", "generate", "+preset=narrowband_classifier_baseline_xs",
        "run.seed=42", "run.num_samples=100",
    ]
    subprocess.run([*cmd_template, f"storage.path={out1}"], check=True)
    subprocess.run([*cmd_template, f"storage.path={out2}"], check=True)

    diff = subprocess.run(
        ["python", "scripts/diff_zarr.py", out1, out2, "--strict-iq", "--strict-labels"],
        capture_output=True, text=True,
    )
    assert diff.returncode == 0, f"Reproduction not byte-exact:\n{diff.stdout}"

test_zarr_phase2_append.py

def test_phase2_idempotent_append(tmp_path):
    """Re-running annotate_store against an already-annotated sample is a no-op."""
    from rfgen.inference.clients import OpenAIClient
    from rfgen.inference.testing import client_for_testing
    from rfgen.annotators import AnnotationType, MetadataAnnotator
    from rfgen.storage import StoreMode, ZarrStore
    import zarr

    rfgen.generate("+preset=narrowband_xs", output=tmp_path)
    dataset = tmp_path / "samples.zarr"
    store = ZarrStore(uri=f"file://{dataset}")
    dataset_uri = store.dataset_uri()
    assert dataset_uri is not None
    client = client_for_testing(
        OpenAIClient,
        transport=lambda _payload: {
            "id": "req-1",
            "model": "test-model",
            "choices": [{
                "message": {"content": "{\"caption\": \"two narrowband emitters\"}"},
                "finish_reason": "stop",
            }],
            "usage": {
                "prompt_tokens": 3,
                "completion_tokens": 5,
                "prompt_tokens_details": {"cached_tokens": 0},
            },
        },
    )
    annotator = MetadataAnnotator.from_components(client=client)

    with store.open(dataset_uri, StoreMode.APPEND) as handle:
        sample_ids = handle.sample_ids()
        assert len(sample_ids) == 1
        sample_id = sample_ids[0]
        annotator.annotate_store(
            handle,
            sample_id=sample_id,
            annotation_type=AnnotationType.CAPTION,
            template_id="caption.dense.v3",
            run_id="run-1",
        )

    ds = zarr.open_group(str(dataset), mode="r")
    sample_group = ds["samples"][sample_id]
    text_before = bytes(sample_group["text_cbor"][:].tolist())
    canonical_before = bytes(sample_group["canonical_cbor"][:].tolist())
    sample_keys_before = tuple(sorted(ds["samples"].group_keys()))

    with store.open(dataset_uri, StoreMode.APPEND) as handle:
        annotator.annotate_store(
            handle,
            sample_id=sample_id,
            annotation_type=AnnotationType.CAPTION,
            template_id="caption.dense.v3",
            run_id="run-1",
        )

    ds = zarr.open_group(str(dataset), mode="r")
    sample_group = ds["samples"][sample_id]
    assert bytes(sample_group["text_cbor"][:].tolist()) == text_before
    assert bytes(sample_group["canonical_cbor"][:].tolist()) == canonical_before
    assert tuple(sorted(ds["samples"].group_keys())) == sample_keys_before

See Also