Contract Tests

Required per-ABC tests that every registered plugin implementation must satisfy.

Warning

Pre-implementation. This page describes proposed contracts. Class signatures, parameter types, config field names, and behavior are subject to change before code lands. Once implementation exists, content here will be regenerated from docstrings or sourced from running tests.

Contract tests

Each ABC has one parametrized contract test that runs against every registered implementation. New implementations join the suite by registering; no test modifications needed.

test_base_emitter_contract.py

import pytest
import torch
from rfgen.emitters.base import BaseEmitter
from rfgen.core.registry import all_emitters
from rfgen.core.rng import _make_generator

# Discover every registered emitter and parametrize. New emitters auto-join.
@pytest.fixture(params=all_emitters(), ids=lambda em: em.family + "/" + em.supported_classes[0])
def emitter_and_class(request):
    cls = request.param
    return cls(), cls.supported_classes[0]


class TestBaseEmitterContract:
    """Contract every BaseEmitter implementation must satisfy."""

    def test_output_shape(self, emitter_and_class, tiny_scene):
        emitter, class_label = emitter_and_class
        rng = _make_generator(42)
        signal = emitter.generate(
            class_label=class_label,
            sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s,
            f_offset_hz=0.0,
            rng=rng.torch_rng,
        )
        expected_n = int(tiny_scene.sample_rate * tiny_scene.duration_s)
        assert signal.iq.shape == (2, expected_n), \
            f"IQ shape {signal.iq.shape} != expected (2, {expected_n})"

    def test_output_dtype_float32(self, emitter_and_class, tiny_scene):
        emitter, class_label = emitter_and_class
        signal = emitter.generate(
            class_label=class_label,
            sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s,
            f_offset_hz=0.0,
            rng=_make_generator(42).torch_rng,
        )
        assert signal.iq.dtype == torch.float32

    def test_snr_db_is_inf(self, emitter_and_class, tiny_scene):
        emitter, class_label = emitter_and_class
        signal = emitter.generate(
            class_label=class_label,
            sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s,
            f_offset_hz=0.0,
            rng=_make_generator(42).torch_rng,
        )
        assert signal.metadata.snr_db == float("inf"), \
            "Emitter must not add noise; channel sets SNR"

    def test_metadata_self_consistent(self, emitter_and_class, tiny_scene):
        emitter, class_label = emitter_and_class
        signal = emitter.generate(
            class_label=class_label,
            sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s,
            f_offset_hz=0.0,
            rng=_make_generator(42).torch_rng,
        )
        meta = signal.metadata
        assert 0.0 <= meta.t_start < meta.t_stop <= tiny_scene.duration_s
        nyq = tiny_scene.sample_rate / 2
        assert -nyq <= meta.f_low < meta.f_high <= nyq

    def test_determinism(self, emitter_and_class, tiny_scene):
        emitter, class_label = emitter_and_class
        rng1 = _make_generator(42).torch_rng
        rng2 = _make_generator(42).torch_rng
        s1 = emitter.generate(class_label=class_label, sample_rate=tiny_scene.sample_rate,
                              duration_s=tiny_scene.duration_s, f_offset_hz=0.0, rng=rng1)
        s2 = emitter.generate(class_label=class_label, sample_rate=tiny_scene.sample_rate,
                              duration_s=tiny_scene.duration_s, f_offset_hz=0.0, rng=rng2)
        assert torch.equal(s1.iq, s2.iq), "Same rng + params must produce byte-identical IQ"

    def test_baseband_centering(self, emitter_and_class, tiny_scene):
        """f_offset_hz=0 → spectral peak within ±0.5*bandwidth_hz of DC."""
        emitter, class_label = emitter_and_class
        signal = emitter.generate(
            class_label=class_label, sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s, f_offset_hz=0.0,
            rng=_make_generator(42).torch_rng,
        )
        iq = signal.iq
        complex_iq = iq[0] + 1j * iq[1]
        spec = torch.abs(torch.fft.fftshift(torch.fft.fft(complex_iq))) ** 2
        freqs = torch.fft.fftshift(torch.fft.fftfreq(len(complex_iq), d=1/tiny_scene.sample_rate))
        peak_freq = freqs[spec.argmax()].item()
        assert abs(peak_freq) < signal.metadata.bandwidth_hz / 2, \
            f"Spectral peak at {peak_freq} Hz, expected within ±{signal.metadata.bandwidth_hz/2}"

    def test_frequency_offset_applied(self, emitter_and_class, tiny_scene):
        """f_offset_hz=1 MHz → spectral peak shifts by ~1 MHz."""
        emitter, class_label = emitter_and_class
        target_offset = 1e6
        signal = emitter.generate(
            class_label=class_label, sample_rate=tiny_scene.sample_rate,
            duration_s=tiny_scene.duration_s, f_offset_hz=target_offset,
            rng=_make_generator(42).torch_rng,
        )
        iq = signal.iq
        complex_iq = iq[0] + 1j * iq[1]
        spec = torch.abs(torch.fft.fftshift(torch.fft.fft(complex_iq))) ** 2
        freqs = torch.fft.fftshift(torch.fft.fftfreq(len(complex_iq), d=1/tiny_scene.sample_rate))
        peak_freq = freqs[spec.argmax()].item()
        # Peak should be near target_offset, within bandwidth/2
        assert abs(peak_freq - target_offset) < signal.metadata.bandwidth_hz, \
            f"Expected peak near {target_offset}; got {peak_freq}"

    def test_no_global_state(self, emitter_and_class, tiny_scene):
        """Two emitter instances with same rng produce same output."""
        cls = type(emitter_and_class[0])
        e1, e2 = cls(), cls()
        kwargs = dict(class_label=emitter_and_class[1],
                      sample_rate=tiny_scene.sample_rate,
                      duration_s=tiny_scene.duration_s, f_offset_hz=0.0)
        s1 = e1.generate(rng=_make_generator(42).torch_rng, **kwargs)
        s2 = e2.generate(rng=_make_generator(42).torch_rng, **kwargs)
        assert torch.equal(s1.iq, s2.iq)

    def test_pydantic_schema_returns_basemodel(self, emitter_and_class):
        emitter, _ = emitter_and_class
        from pydantic import BaseModel
        schema_cls = emitter.schema()
        assert issubclass(schema_cls, BaseModel)

    def test_invalid_class_label_raises(self, emitter_and_class, tiny_scene):
        emitter, _ = emitter_and_class
        with pytest.raises(ValueError):
            emitter.generate(
                class_label="__not_a_valid_class__",
                sample_rate=tiny_scene.sample_rate,
                duration_s=tiny_scene.duration_s, f_offset_hz=0.0,
                rng=_make_generator(42).torch_rng,
            )

test_base_channel_contract.py

class TestBaseChannelContract:
    @pytest.fixture(params=all_channels())
    def channel(self, request):
        return request.param()

    def test_iq_shape_preserved(self, channel, tiny_scene):
        """For stochastic channels, output shape == input shape."""
        rng = _make_generator(42).torch_rng
        signal_in = _dummy_signal(n_samples=10000, sample_rate=10e6)
        signal_out = channel.apply(signal_in, rng=rng)
        if not channel.may_extend_tail:
            assert signal_out.iq.shape == signal_in.iq.shape

    def test_deterministic(self, channel, tiny_scene):
        signal_in = _dummy_signal(n_samples=10000, sample_rate=10e6)
        rng1, rng2 = _make_generator(42).torch_rng, _make_generator(42).torch_rng
        out1 = channel.apply(signal_in, rng=rng1)
        out2 = channel.apply(signal_in, rng=rng2)
        assert torch.allclose(out1.iq, out2.iq, atol=1e-6)  # GPU-non-determinism caveat

    def test_complex64_dtype(self, channel, tiny_scene):
        signal_in = _dummy_signal(n_samples=10000, sample_rate=10e6)
        out = channel.apply(signal_in, rng=_make_generator(42).torch_rng)
        assert out.iq.dtype == torch.float32

    def test_channel_metadata_recorded(self, channel, tiny_scene):
        signal_in = _dummy_signal(n_samples=10000, sample_rate=10e6)
        out = channel.apply(signal_in, rng=_make_generator(42).torch_rng)
        # channel_profile must be set in signal.metadata.extras
        assert "channel_profile" in out.metadata.extras or hasattr(channel, "_no_profile_required")

    def test_unsupported_multi_rx_raises(self, channel):
        if channel.supports_multi_rx:
            pytest.skip("channel supports multi-RX")
        signal_in = _dummy_signal(n_samples=1000, sample_rate=10e6)
        with pytest.raises((NotImplementedError, ValueError)):
            channel.apply(
                signal_in,
                rng=_make_generator(42).torch_rng,
                params={"num_rx": 4},
            )

test_base_scene_composer_contract.py

class TestBaseSceneComposerContract:
    @pytest.fixture
    def composer(self):
        from rfgen.scene import DefaultSceneComposer
        return DefaultSceneComposer()

    def test_build_returns_signal(self, composer, simple_scene_config, fixed_rng):
        signal = composer.build(
            scene_cfg=simple_scene_config,
            emitter_pool={"comms": _torchsig_bpsk()},
            channel=AwgnChannel(),
            rng=fixed_rng.torch_rng,
        )
        assert signal.iq.shape == (2, simple_scene_config.duration_samples)
        assert len(signal.component_signals) == simple_scene_config.density.max_emitters

    def test_per_emitter_metadata_in_scene_frame(self, composer, simple_scene_config):
        scene_signal = composer.build(...)
        for component_signal in scene_signal.component_signals:
            em_meta = component_signal.metadata
            assert 0 <= em_meta.t_start < em_meta.t_stop <= simple_scene_config.duration_s
            scene_low = simple_scene_config.center_hz - simple_scene_config.bandwidth_hz/2
            scene_high = simple_scene_config.center_hz + simple_scene_config.bandwidth_hz/2
            assert scene_low <= em_meta.f_low < em_meta.f_high <= scene_high

    def test_density_realized_within_tolerance(self, composer, simple_scene_config):
        n_runs = 100
        counts = [
            len(composer.build(...).component_signals)
            for _ in range(n_runs)
        ]
        expected = simple_scene_config.density.expected_count()
        observed = np.mean(counts)
        assert abs(observed - expected) / expected < 0.10

    def test_overlap_policy_reject(self, composer, dense_scene_config):
        """With overlap_policy=reject, no two bboxes should overlap."""
        dense_scene_config.overlap_policy = SceneOverlapPolicy.REJECT
        scene_signal = composer.build(...)
        component_metas = [c.metadata for c in scene_signal.component_signals]
        for i, a in enumerate(component_metas):
            for b in component_metas[i+1:]:
                assert not _rectangles_overlap(a, b)

    def test_idempotency(self, composer, simple_scene_config, fixed_rng):
        rng1 = _make_generator(42).torch_rng
        rng2 = _make_generator(42).torch_rng
        s1 = composer.build(scene_cfg=..., rng=rng1, ...)
        s2 = composer.build(scene_cfg=..., rng=rng2, ...)
        assert torch.equal(s1.iq, s2.iq)

test_base_labeler_contract.py

class TestBaseLabelerContract:
    @pytest.fixture(params=all_labelers())
    def labeler(self, request):
        return request.param()

    def test_pure_function(self, labeler, sample_signal):
        labels1 = labeler.label(iq=sample_signal.iq,
                                emitters=sample_signal.component_metas,
                                scene=sample_signal.scene)
        labels2 = labeler.label(iq=sample_signal.iq,
                                emitters=sample_signal.component_metas,
                                scene=sample_signal.scene)
        assert _labels_equal(labels1, labels2)  # bit-exact

    def test_advertises_what_it_produces(self, labeler):
        produces = labeler.produces
        assert produces
        assert all(_is_registry_safe_label_key(key) for key in produces)
        labels = labeler.label(...)
        assert set(labels) == set(produces)

    def test_bbox_within_scene_bounds(self, labeler, sample_signal):
        labels = labeler.label(...)
        if "bbox" not in labeler.produces:
            pytest.skip()
        for bbox in labels["bbox"]:
            assert 0 <= bbox.abs.start_sample < bbox.abs.stop_sample
            assert bbox.abs.stop_sample <= sample_signal.scene.duration_samples

    def test_yolo_derivation_within_1_ulp(self, labeler, sample_signal):
        labels = labeler.label(...)
        for bbox in labels["bbox"]:
            cx_recomputed = (bbox.abs.start_sample + bbox.abs.stop_sample) / 2 / sample_signal.scene.duration_samples
            assert abs(bbox.yolo.cx - cx_recomputed) < 1e-6

    def test_segmentation_shape(self, labeler, sample_signal):
        if "segmentation" not in labeler.produces:
            pytest.skip()
        labels = labeler.label(...)
        seg = labels["segmentation"]
        F = labeler.n_fft
        T = (sample_signal.scene.duration_samples - F) // labeler.hop + 1
        if seg.ndim == 2:
            assert seg.shape == (F, T)
        else:
            assert seg.shape[1:] == (F, T)

    def test_cross_modality_consistency(self, labeler, sample_signal):
        """The four invariants from Reference / Metrics § Schema validation."""
        labels = labeler.label(...)
        # 1. IQ ↔ bbox
        for b in labels["bbox"]:
            assert 0 <= b.abs.start_sample < b.abs.stop_sample <= sample_signal.scene.duration_samples
        # 2. bbox ↔ segmentation (±1 hop tolerance)
        if "segmentation" in labeler.produces and "bbox" in labeler.produces:
            assert _bbox_segmentation_consistent(labels["bbox"], labels["segmentation"],
                                                 tolerance_hops=1)
        # 3. bbox ↔ per_emitter
        # 4. per_emitter ↔ scene
        # ...

test_base_annotator_contract.py

class TestBaseAnnotatorContract:
    @pytest.fixture(params=all_annotators())
    def annotator(self, request):
        return request.param()

    def test_annotation_type_in_set(self, annotator):
        from rfgen.enums import AnnotationType
        assert annotator.annotation_type in set(AnnotationType)

    def test_metadata_only_input(self, annotator, sample_record, mock_llm_client):
        """Annotator must not access iq.zarr or seg.zarr, only whitelisted fields."""
        # Decorate `mock_llm_client` to capture what was sent in prompts
        with sample_record.assert_no_iq_access():
            annotator.annotate(emitters=sample_record.per_emitter,
                               scene=sample_record.scene,
                               llm=mock_llm_client,
                               rng=_make_generator(42))

    def test_output_validates_against_schema(self, annotator, sample_record, mock_llm_client):
        result = annotator.annotate(...)
        # Schema for this annotation type
        schema_cls = ANNOTATION_OUTPUT_SCHEMAS[annotator.annotation_type]
        schema_cls.model_validate(result)

    def test_closed_vocabulary_compliance(self, annotator, sample_record, mock_llm_client):
        result = annotator.annotate(...)
        text = _extract_text(result, annotator.annotation_type)
        unknown_tokens = _find_oov_physical_attribute_tokens(text)
        assert len(unknown_tokens) == 0, \
            f"OOV tokens claiming physical attributes: {unknown_tokens}"

test_base_store_contract.py

class TestBaseStoreContract:
    @pytest.fixture(params=all_stores())
    def store_class(self, request):
        return request.param

    def test_write_then_read_roundtrip(self, store_class, tmp_path, sample):
        store = store_class()
        path = str(tmp_path / "test.store")
        with store.open(path, mode="w") as h:
            sample_id = h.write(sample)
        with store.open(path, mode="r") as h:
            recovered = h.read(sample_id)
        assert torch.equal(recovered.iq, sample.iq)
        assert recovered.scene == sample.scene

    def test_idempotent_write(self, store_class, tmp_path, sample):
        store = store_class()
        path = str(tmp_path / "test.store")
        with store.open(path, mode="w") as h:
            id1 = h.write(sample, sample_id="known-id")
            id2 = h.write(sample, sample_id="known-id")
        assert id1 == id2

    def test_iter_returns_all(self, store_class, tmp_path, samples_list):
        store = store_class()
        path = str(tmp_path / "test.store")
        with store.open(path, mode="w") as h:
            for s in samples_list:
                h.write(s)
        with store.open(path, mode="r") as h:
            assert len(list(iter(h))) == len(samples_list)
            assert len(h) == len(samples_list)

Plugin conformance validation evidence

Layer 34 validates a plugin wheel through a fixed check matrix and fixture-only runtime probes. The focused Layer 34 suite completed with 57 passed.

Linux validation uses the util-linux launcher profile below. It creates a user namespace mapped to the caller and a separate network namespace; it does not require a private /proc mount.

unshare --user --map-root-user --net -- <checker command>

  • Launcher preflight failure is fail-closed: validation reports SANDBOX_UNAVAILABLE rather than attributing the failure to the plugin.

  • A timeout terminates and reaps the checker process group. The contract test starts a descendant that would write a marker after the timeout, then asserts that the marker is absent.

  • Debian 12 on GCP verified that the user-and-network namespace profile launches without --mount-proc and that the checker cannot reach the external network.

macOS remains PARTIAL: it provides the sanitized environment, temporary working directory, subprocess timeout, and typed result, but does not claim network or address-space isolation.

See Also