test_reader

Tests for FrameReader context manager.

TestFrameReaderLifecycle

Tests for open/close and context manager behaviour.

test_close_is_idempotent

test_close_is_idempotent(single_frame_file)

Calling close() twice does not raise.

Source code in gwframe/tests/test_reader.py

def test_close_is_idempotent(self, single_frame_file):
    """Calling close() twice does not raise."""
    f = gwframe.FrameReader(single_frame_file)
    f.close()
    f.close()

test_context_manager

test_context_manager(single_frame_file)

Stream is usable inside with-block and cleaned up after.

Source code in gwframe/tests/test_reader.py

def test_context_manager(self, single_frame_file):
    """Stream is usable inside with-block and cleaned up after."""
    with gwframe.FrameReader(single_frame_file) as f:
        assert f.num_frames == 1
    # After exit, reads should fail
    with pytest.raises(ValueError, match="closed"):
        f.read("L1:CH_A")

test_explicit_close

test_explicit_close(single_frame_file)

close() makes subsequent reads raise.

Source code in gwframe/tests/test_reader.py

def test_explicit_close(self, single_frame_file):
    """close() makes subsequent reads raise."""
    f = gwframe.FrameReader(single_frame_file)
    f.read("L1:CH_A")  # should work
    f.close()
    with pytest.raises(ValueError, match="closed"):
        f.read("L1:CH_A")

test_metadata_after_close_raises

test_metadata_after_close_raises(single_frame_file)

Accessing metadata properties after close raises ValueError.

Source code in gwframe/tests/test_reader.py

def test_metadata_after_close_raises(self, single_frame_file):
    """Accessing metadata properties after close raises ValueError."""
    f = gwframe.FrameReader(single_frame_file)
    f.close()
    with pytest.raises(ValueError, match="closed"):
        _ = f.num_frames
    with pytest.raises(ValueError, match="closed"):
        _ = f.channels
    with pytest.raises(ValueError, match="closed"):
        _ = f.info

test_nonexistent_file_raises

test_nonexistent_file_raises()

Opening a nonexistent file raises RuntimeError.

Source code in gwframe/tests/test_reader.py

def test_nonexistent_file_raises(self):
    """Opening a nonexistent file raises RuntimeError."""
    with pytest.raises(RuntimeError):
        gwframe.FrameReader("/nonexistent/file.gwf")

test_repr_closed

test_repr_closed(single_frame_file)

repr shows after close.

Source code in gwframe/tests/test_reader.py

def test_repr_closed(self, single_frame_file):
    """repr shows <closed> after close."""
    f = gwframe.FrameReader(single_frame_file)
    f.close()
    assert repr(f) == "FrameReader(<closed>)"

test_repr_open

test_repr_open(single_frame_file)

repr shows filename, frame count, and channel count.

Source code in gwframe/tests/test_reader.py

def test_repr_open(self, single_frame_file):
    """repr shows filename, frame count, and channel count."""
    with gwframe.FrameReader(single_frame_file) as f:
        r = repr(f)
        assert "FrameReader(" in r
        assert "1 frames" in r
        assert "2 channels" in r

TestFrameReaderMetadata

Tests for metadata properties matching get_info / get_channel_details.

test_channel_details

test_channel_details(single_frame_file)

channel_details matches get_channel_details.

Source code in gwframe/tests/test_reader.py

def test_channel_details(self, single_frame_file):
    """channel_details matches get_channel_details."""
    expected = get_channel_details(single_frame_file)
    with gwframe.FrameReader(single_frame_file) as f:
        details = f.channel_details()
        assert len(details) == len(expected)
        for d, e in zip(details, expected):
            assert isinstance(d, ChannelInfo)
            assert d.name == e.name
            assert d.type == e.type
            assert d.dtype == e.dtype
            assert d.sample_rate == e.sample_rate
            assert d.n_samples == e.n_samples

test_channels

test_channels(single_frame_file)

channels matches get_info.

Source code in gwframe/tests/test_reader.py

def test_channels(self, single_frame_file):
    """channels matches get_info."""
    info = get_info(single_frame_file)
    with gwframe.FrameReader(single_frame_file) as f:
        assert sorted(f.channels) == sorted(info.channels)

test_filename

test_filename(single_frame_file)

filename property returns the path.

Source code in gwframe/tests/test_reader.py

def test_filename(self, single_frame_file):
    """filename property returns the path."""
    with gwframe.FrameReader(single_frame_file) as f:
        assert f.filename == str(single_frame_file)

test_frames

test_frames(multi_frame_file)

frames list matches get_info.

Source code in gwframe/tests/test_reader.py

def test_frames(self, multi_frame_file):
    """frames list matches get_info."""
    info = get_info(multi_frame_file)
    with gwframe.FrameReader(multi_frame_file) as f:
        assert len(f.frames) == len(info.frames)
        for reader_fi, info_fi in zip(f.frames, info.frames):
            assert reader_fi.index == info_fi.index
            assert reader_fi.start == info_fi.start
            assert reader_fi.duration == info_fi.duration

test_info

test_info(single_frame_file)

info property matches get_info.

Source code in gwframe/tests/test_reader.py

def test_info(self, single_frame_file):
    """info property matches get_info."""
    info = get_info(single_frame_file)
    with gwframe.FrameReader(single_frame_file) as f:
        assert f.info.num_frames == info.num_frames
        assert f.info.compression == info.compression

test_info_is_cached

test_info_is_cached(single_frame_file)

info property returns same object on repeated access.

Source code in gwframe/tests/test_reader.py

def test_info_is_cached(self, single_frame_file):
    """info property returns same object on repeated access."""
    with gwframe.FrameReader(single_frame_file) as f:
        info1 = f.info
        info2 = f.info
        assert info1 is info2

test_num_frames

test_num_frames(multi_frame_file)

num_frames matches get_info.

Source code in gwframe/tests/test_reader.py

def test_num_frames(self, multi_frame_file):
    """num_frames matches get_info."""
    info = get_info(multi_frame_file)
    with gwframe.FrameReader(multi_frame_file) as f:
        assert f.num_frames == info.num_frames

TestFrameReaderMultiChannel

Tests for multi-channel reads matching gwframe.read().

test_read_all_channels

test_read_all_channels(single_frame_file)

channels=None reads all channels, matching gwframe.read().

Source code in gwframe/tests/test_reader.py

def test_read_all_channels(self, single_frame_file):
    """channels=None reads all channels, matching gwframe.read()."""
    expected = gwframe.read(single_frame_file, channels=None)
    with gwframe.FrameReader(single_frame_file) as f:
        result = f.read()

    assert isinstance(result, dict)
    assert set(result.keys()) == set(expected.keys())
    for ch in expected:
        assert np.array_equal(result[ch].array, expected[ch].array)

test_read_channel_list

test_read_channel_list(single_frame_file)

Reading a specific list of channels matches gwframe.read().

Source code in gwframe/tests/test_reader.py

def test_read_channel_list(self, single_frame_file):
    """Reading a specific list of channels matches gwframe.read()."""
    channels = ["L1:CH_A", "L1:CH_B"]
    expected = gwframe.read(single_frame_file, channels)
    with gwframe.FrameReader(single_frame_file) as f:
        result = f.read(channels)

    assert isinstance(result, dict)
    assert set(result.keys()) == set(expected.keys())
    for ch in expected:
        assert np.array_equal(result[ch].array, expected[ch].array)

TestFrameReaderMultipleReads

Tests for performing multiple reads from the same reader.

test_interleaved_reads

test_interleaved_reads(multi_frame_file)

Interleaving single and multi-channel reads.

Source code in gwframe/tests/test_reader.py

def test_interleaved_reads(self, multi_frame_file):
    """Interleaving single and multi-channel reads."""
    with gwframe.FrameReader(multi_frame_file) as f:
        single = f.read("L1:TEST", frame_index=0)
        all_ch = f.read(channels=None, frame_index=1)
        single2 = f.read("L1:TEST", frame_index=2)

    assert np.all(single.array == 0.0)
    assert isinstance(all_ch, dict)
    assert np.all(single2.array == 2.0)

test_multiple_channels_same_reader

test_multiple_channels_same_reader(single_frame_file)

Reading different channels sequentially from same reader.

Source code in gwframe/tests/test_reader.py

def test_multiple_channels_same_reader(self, single_frame_file):
    """Reading different channels sequentially from same reader."""
    with gwframe.FrameReader(single_frame_file) as f:
        a = f.read("L1:CH_A")
        b = f.read("L1:CH_B")

    assert a.name == "L1:CH_A"
    assert b.name == "L1:CH_B"
    assert len(a.array) == 1000
    assert len(b.array) == 500

test_multiple_frames_same_reader

test_multiple_frames_same_reader(multi_frame_file)

Reading different frame indices sequentially from same reader.

Source code in gwframe/tests/test_reader.py

def test_multiple_frames_same_reader(self, multi_frame_file):
    """Reading different frame indices sequentially from same reader."""
    with gwframe.FrameReader(multi_frame_file) as f:
        results = [f.read("L1:TEST", frame_index=i) for i in range(3)]

    for i, ts in enumerate(results):
        assert np.all(ts.array == float(i))

TestFrameReaderSingleChannel

Tests for single-channel reads matching gwframe.read().

test_invalid_frame_index_raises

test_invalid_frame_index_raises(single_frame_file)

Reading an out-of-bounds frame index raises FrameIndexError.

Source code in gwframe/tests/test_reader.py

def test_invalid_frame_index_raises(self, single_frame_file):
    """Reading an out-of-bounds frame index raises FrameIndexError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(FrameIndexError),
    ):
        f.read("L1:CH_A", frame_index=999)

test_nonexistent_channel_raises

test_nonexistent_channel_raises(single_frame_file)

Reading a nonexistent channel raises ChannelNotFoundError.

Source code in gwframe/tests/test_reader.py

def test_nonexistent_channel_raises(self, single_frame_file):
    """Reading a nonexistent channel raises ChannelNotFoundError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(ChannelNotFoundError),
    ):
        f.read("NONEXISTENT:CHANNEL")

test_single_channel_different_frame_index

test_single_channel_different_frame_index(multi_frame_file)

Reading different frame indices produces correct data.

Source code in gwframe/tests/test_reader.py

def test_single_channel_different_frame_index(self, multi_frame_file):
    """Reading different frame indices produces correct data."""
    with gwframe.FrameReader(multi_frame_file) as f:
        for i in range(3):
            expected = gwframe.read(multi_frame_file, "L1:TEST", frame_index=i)
            result = f.read("L1:TEST", frame_index=i)
            assert np.array_equal(result.array, expected.array)
            assert result.start == expected.start

test_single_channel_matches_read

test_single_channel_matches_read(single_frame_file)

FrameReader.read() produces identical arrays to gwframe.read().

Source code in gwframe/tests/test_reader.py

def test_single_channel_matches_read(self, single_frame_file):
    """FrameReader.read() produces identical arrays to gwframe.read()."""
    expected = gwframe.read(single_frame_file, "L1:CH_A")
    with gwframe.FrameReader(single_frame_file) as f:
        result = f.read("L1:CH_A")
    assert isinstance(result, gwframe.TimeSeries)
    assert np.array_equal(result.array, expected.array)
    assert result.start == expected.start
    assert result.dt == expected.dt
    assert result.sample_rate == expected.sample_rate
    assert result.name == expected.name

TestFrameReaderTimeSlice

Tests for time-sliced reads matching gwframe.read().

test_invalid_time_range_raises

test_invalid_time_range_raises(time_slice_file)

Out-of-range time slice raises InvalidTimeRangeError.

Source code in gwframe/tests/test_reader.py

def test_invalid_time_range_raises(self, time_slice_file):
    """Out-of-range time slice raises InvalidTimeRangeError."""
    with (
        gwframe.FrameReader(time_slice_file) as f,
        pytest.raises(InvalidTimeRangeError),
    ):
        f.read("L1:TEST", start=9999999999.0, end=9999999999.5)

test_time_slice_multi_channel

test_time_slice_multi_channel(time_slice_file)

Time-sliced multi-channel read matches gwframe.read().

Source code in gwframe/tests/test_reader.py

def test_time_slice_multi_channel(self, time_slice_file):
    """Time-sliced multi-channel read matches gwframe.read()."""
    start = 1234567891.0
    end = 1234567893.0
    expected = gwframe.read(time_slice_file, channels=None, start=start, end=end)
    with gwframe.FrameReader(time_slice_file) as f:
        result = f.read(channels=None, start=start, end=end)

    assert isinstance(result, dict)
    assert set(result.keys()) == set(expected.keys())
    for ch in expected:
        assert np.array_equal(result[ch].array, expected[ch].array)

test_time_slice_single_channel

test_time_slice_single_channel(time_slice_file)

Time-sliced single-channel read matches gwframe.read().

Source code in gwframe/tests/test_reader.py

def test_time_slice_single_channel(self, time_slice_file):
    """Time-sliced single-channel read matches gwframe.read()."""
    start = 1234567891.0
    end = 1234567893.0
    expected = gwframe.read(time_slice_file, "L1:TEST", start=start, end=end)
    with gwframe.FrameReader(time_slice_file) as f:
        result = f.read("L1:TEST", start=start, end=end)

    assert isinstance(result, gwframe.TimeSeries)
    assert np.array_equal(result.array, expected.array)
    assert result.start == expected.start
    assert result.duration == expected.duration

time_slice_file

time_slice_file(tmp_path)

Create a multi-frame file suitable for time slicing.

Source code in gwframe/tests/test_reader.py

@pytest.fixture
def time_slice_file(self, tmp_path):
    """Create a multi-frame file suitable for time slicing."""
    path = tmp_path / "timeslice.gwf"
    with gwframe.FrameWriter(str(path)) as writer:
        for i in range(5):
            data = np.full(100, float(i), dtype=np.float64)
            writer.write(
                data,
                start=1234567890.0 + i,
                sample_rate=100,
                name="L1:TEST",
                unit="counts",
            )
    return path

TestFrameReaderValidation

Tests for parameter validation.

test_end_without_start_raises

test_end_without_start_raises(single_frame_file)

end without start raises ValueError.

Source code in gwframe/tests/test_reader.py

def test_end_without_start_raises(self, single_frame_file):
    """end without start raises ValueError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(ValueError, match="start and end must be specified"),
    ):
        f.read("L1:CH_A", end=1234567891.0)

test_invalid_channels_type_raises

test_invalid_channels_type_raises(single_frame_file)

Passing wrong type for channels raises TypeError.

Source code in gwframe/tests/test_reader.py

def test_invalid_channels_type_raises(self, single_frame_file):
    """Passing wrong type for channels raises TypeError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(TypeError),
    ):
        f.read(123)

test_start_end_with_frame_index_raises

test_start_end_with_frame_index_raises(single_frame_file)

start/end with non-zero frame_index raises ValueError.

Source code in gwframe/tests/test_reader.py

def test_start_end_with_frame_index_raises(self, single_frame_file):
    """start/end with non-zero frame_index raises ValueError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(ValueError, match="mutually exclusive"),
    ):
        f.read(
            "L1:CH_A",
            frame_index=1,
            start=1234567890.0,
            end=1234567891.0,
        )

test_start_without_end_raises

test_start_without_end_raises(single_frame_file)

start without end raises ValueError.

Source code in gwframe/tests/test_reader.py

def test_start_without_end_raises(self, single_frame_file):
    """start without end raises ValueError."""
    with (
        gwframe.FrameReader(single_frame_file) as f,
        pytest.raises(ValueError, match="start and end must be specified"),
    ):
        f.read("L1:CH_A", start=1234567890.0)

test_validate_checksum

test_validate_checksum(single_frame_file)

validate_checksum=True works without error on valid files.

Source code in gwframe/tests/test_reader.py

def test_validate_checksum(self, single_frame_file):
    """validate_checksum=True works without error on valid files."""
    with gwframe.FrameReader(single_frame_file, validate_checksum=True) as f:
        result = f.read("L1:CH_A")
    assert len(result.array) > 0

multi_frame_file

multi_frame_file(tmp_path)

Create a multi-frame file with known per-frame data.

Source code in gwframe/tests/test_reader.py

@pytest.fixture
def multi_frame_file(tmp_path):
    """Create a multi-frame file with known per-frame data."""
    path = tmp_path / "multi.gwf"
    with gwframe.FrameWriter(str(path)) as writer:
        for i in range(3):
            data = np.full(1000, float(i), dtype=np.float64)
            writer.write(
                data,
                start=1234567890.0 + i,
                sample_rate=1000,
                name="L1:TEST",
                unit="counts",
            )
    return path

single_frame_file

single_frame_file(tmp_path)

Create a single-frame file with two proc channels.

Source code in gwframe/tests/test_reader.py

@pytest.fixture
def single_frame_file(tmp_path):
    """Create a single-frame file with two proc channels."""
    path = tmp_path / "single.gwf"
    gwframe.write(
        str(path),
        {
            "L1:CH_A": np.arange(1000, dtype=np.float64),
            "L1:CH_B": np.ones(500, dtype=np.float32),
        },
        start=1234567890.0,
        sample_rate=1000,
    )
    return path