test_cli

Tests for gwframe CLI commands.

TestCLIVsAPI

Tests comparing CLI results with direct API calls.

test_drop_cli_matches_api

test_drop_cli_matches_api(single_frame_file, tmp_path)

Test that CLI drop produces same results as API.

Source code in gwframe/tests/test_cli.py

def test_drop_cli_matches_api(self, single_frame_file, tmp_path):
    """Test that CLI drop produces same results as API."""
    cli_output = tmp_path / "cli_output"
    api_output = tmp_path / "api_output"

    # Run via CLI
    result = runner.invoke(
        app,
        ["drop", str(single_frame_file), "-o", str(cli_output), "-c", "L1:CHAN1"],
    )
    assert result.exit_code == 0

    # Run via API
    drop_channels(
        str(single_frame_file), str(api_output), channels_to_drop=["L1:CHAN1"]
    )

    # Compare results
    cli_file = cli_output / single_frame_file.name
    api_file = api_output / single_frame_file.name

    cli_channels = sorted(get_channels(str(cli_file)))
    api_channels = sorted(get_channels(str(api_file)))
    assert cli_channels == api_channels

test_rename_cli_matches_api

test_rename_cli_matches_api(single_frame_file, tmp_path)

Test that CLI rename produces same results as API.

Source code in gwframe/tests/test_cli.py

def test_rename_cli_matches_api(self, single_frame_file, tmp_path):
    """Test that CLI rename produces same results as API."""
    cli_output = tmp_path / "cli_output"
    api_output = tmp_path / "api_output"

    channel_map = {"L1:CHAN1": "TEST:RENAMED1"}

    # Run via CLI
    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "-o",
            str(cli_output),
            "-m",
            "L1:CHAN1=>TEST:RENAMED1",
        ],
    )
    assert result.exit_code == 0

    # Run via API
    rename_channels(
        str(single_frame_file), str(api_output), channel_map=channel_map
    )

    # Compare results
    cli_file = cli_output / single_frame_file.name
    api_file = api_output / single_frame_file.name

    cli_channels = sorted(get_channels(str(cli_file)))
    api_channels = sorted(get_channels(str(api_file)))
    assert cli_channels == api_channels

    # Compare data
    cli_data = gwframe.read(str(cli_file), "TEST:RENAMED1")
    api_data = gwframe.read(str(api_file), "TEST:RENAMED1")
    assert np.allclose(cli_data.array, api_data.array)

test_resize_cli_matches_api

test_resize_cli_matches_api(multi_frame_file, tmp_path, sample_data)

Test that CLI resize produces same results as API.

Source code in gwframe/tests/test_cli.py

def test_resize_cli_matches_api(self, multi_frame_file, tmp_path, sample_data):
    """Test that CLI resize produces same results as API."""
    cli_output = tmp_path / "cli_output"
    api_output = tmp_path / "api_output"
    target_duration = sample_data["duration"] / 2

    # Run via CLI
    result = runner.invoke(
        app,
        [
            "resize",
            str(multi_frame_file),
            "-o",
            str(cli_output),
            "-d",
            str(target_duration),
        ],
    )
    assert result.exit_code == 0

    # Run via API
    resize_frames(
        str(multi_frame_file), str(api_output), target_duration=target_duration
    )

    # Compare results
    cli_file = cli_output / multi_frame_file.name
    api_file = api_output / multi_frame_file.name

    cli_frames = list(gwframe.read_frames(str(cli_file)))
    api_frames = list(gwframe.read_frames(str(api_file)))
    assert len(cli_frames) == len(api_frames)

    # Compare frame data
    for cli_frame, api_frame in zip(cli_frames, api_frames):
        assert abs(cli_frame.start - api_frame.start) < 1e-9
        assert abs(cli_frame.duration - api_frame.duration) < 1e-9

test_select_cli_matches_api

test_select_cli_matches_api(single_frame_file, tmp_path)

Test that CLI select produces same results as API.

Source code in gwframe/tests/test_cli.py

def test_select_cli_matches_api(self, single_frame_file, tmp_path):
    """Test that CLI select produces same results as API."""
    cli_output = tmp_path / "cli_output"
    api_output = tmp_path / "api_output"

    # Run via CLI
    result = runner.invoke(
        app,
        [
            "select",
            str(single_frame_file),
            "-o",
            str(cli_output),
            "-c",
            "L1:CHAN1",
            "-c",
            "L1:CHAN3",
        ],
    )
    assert result.exit_code == 0

    # Run via API
    select_channels(
        str(single_frame_file),
        str(api_output),
        channels_to_select=["L1:CHAN1", "L1:CHAN3"],
    )

    # Compare results
    cli_file = cli_output / single_frame_file.name
    api_file = api_output / single_frame_file.name

    cli_channels = sorted(get_channels(str(cli_file)))
    api_channels = sorted(get_channels(str(api_file)))
    assert cli_channels == api_channels

    # Compare data
    for chan in ["L1:CHAN1", "L1:CHAN3"]:
        cli_data = gwframe.read(str(cli_file), chan)
        api_data = gwframe.read(str(api_file), chan)
        assert np.allclose(cli_data.array, api_data.array)

TestCombineCommand

Tests for gwframe combine command.

test_combine_errors

test_combine_errors(single_frame_file, tmp_path)

Test combine error handling.

Source code in gwframe/tests/test_cli.py

def test_combine_errors(self, single_frame_file, tmp_path):
    """Test combine error handling."""
    output_dir = tmp_path / "output"

    # Test less than 2 sources
    result = runner.invoke(
        app, ["combine", str(single_frame_file), "-o", str(output_dir)]
    )
    assert result.exit_code != 0
    assert "at least 2" in result.stdout.lower()

test_combine_files_with_filtering

test_combine_files_with_filtering(tmp_path, sample_data, create_test_frame)

Test combine command with files and channel filtering.

Source code in gwframe/tests/test_cli.py

def test_combine_files_with_filtering(
    self, tmp_path, sample_data, create_test_frame
):
    """Test combine command with files and channel filtering."""
    # Create two files with different channels
    file1 = tmp_path / "file1.gwf"
    file2 = tmp_path / "file2.gwf"
    n_samples = sample_data["n_samples"]

    data = np.arange(n_samples, dtype=np.float32)

    with gwframe.FrameWriter(str(file1)) as writer:
        channels = {"L1:CHAN_A": data, "L1:CHAN_B": data * 2}
        frame = create_test_frame(
            sample_data["t0"], sample_data["duration"], channels
        )
        writer.write_frame(frame)

    with gwframe.FrameWriter(str(file2)) as writer:
        channels = {"L1:CHAN_C": data * 3}
        frame = create_test_frame(
            sample_data["t0"], sample_data["duration"], channels
        )
        writer.write_frame(frame)

    output_dir = tmp_path / "output"

    # Test basic combine
    result = runner.invoke(
        app, ["combine", str(file1), str(file2), "-o", str(output_dir)]
    )
    assert result.exit_code == 0, result.stdout

    output_files = list(output_dir.glob("*.gwf"))
    frames = list(gwframe.read_frames(str(output_files[0])))
    assert "L1:CHAN_A" in frames[0]
    assert "L1:CHAN_B" in frames[0]
    assert "L1:CHAN_C" in frames[0]

    # Test with --keep filtering
    output_dir_keep = tmp_path / "output_keep"
    result = runner.invoke(
        app,
        [
            "combine",
            str(file1),
            str(file2),
            "-o",
            str(output_dir_keep),
            "-k",
            "L1:CHAN_A",
            "-k",
            "L1:CHAN_C",
        ],
    )
    assert result.exit_code == 0

    output_files = list(output_dir_keep.glob("*.gwf"))
    frames = list(gwframe.read_frames(str(output_files[0])))
    assert "L1:CHAN_A" in frames[0]
    assert "L1:CHAN_C" in frames[0]
    assert "L1:CHAN_B" not in frames[0]

TestDropCommand

Tests for gwframe drop command.

test_drop_basic

test_drop_basic(single_frame_file, tmp_path)

Test basic drop command.

Source code in gwframe/tests/test_cli.py

def test_drop_basic(self, single_frame_file, tmp_path):
    """Test basic drop command."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "drop",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify channel was dropped
    output_file = output_dir / single_frame_file.name
    channels = get_channels(str(output_file))
    assert "L1:CHAN1" not in channels
    assert "L1:CHAN2" in channels

test_drop_in_place

test_drop_in_place(single_frame_file, tmp_path)

Test drop with --in-place.

Source code in gwframe/tests/test_cli.py

def test_drop_in_place(self, single_frame_file, tmp_path):
    """Test drop with --in-place."""
    test_file = tmp_path / "test.gwf"
    shutil.copy(single_frame_file, test_file)

    result = runner.invoke(app, ["drop", str(test_file), "-i", "-c", "L1:CHAN1"])

    assert result.exit_code == 0, result.stdout

    # Verify channel was dropped
    channels = get_channels(str(test_file))
    assert "L1:CHAN1" not in channels
    assert "L1:CHAN2" in channels

test_drop_multiple_channels

test_drop_multiple_channels(single_frame_file, tmp_path)

Test dropping multiple channels.

Source code in gwframe/tests/test_cli.py

def test_drop_multiple_channels(self, single_frame_file, tmp_path):
    """Test dropping multiple channels."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "drop",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-c",
            "L1:CHAN1",
            "-c",
            "L1:CHAN2",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify both channels were dropped
    output_file = output_dir / single_frame_file.name
    channels = get_channels(str(output_file))
    assert "L1:CHAN1" not in channels
    assert "L1:CHAN2" not in channels

TestErrorHandling

Tests for CLI error handling.

test_invalid_mapping_format

test_invalid_mapping_format(single_frame_file, tmp_path)

Test that invalid mapping format is caught.

Source code in gwframe/tests/test_cli.py

def test_invalid_mapping_format(self, single_frame_file, tmp_path):
    """Test that invalid mapping format is caught."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-m",
            "INVALID_MAPPING",
        ],
    )

    assert result.exit_code != 0
    assert "invalid" in result.stdout.lower()

test_missing_output_dir

test_missing_output_dir(single_frame_file)

Test that missing output directory is caught.

Source code in gwframe/tests/test_cli.py

def test_missing_output_dir(self, single_frame_file):
    """Test that missing output directory is caught."""
    result = runner.invoke(
        app, ["rename", str(single_frame_file), "-m", "L1:CHAN1=>TEST:RENAMED"]
    )

    assert result.exit_code != 0
    assert "output" in result.stdout.lower() or "required" in result.stdout.lower()

test_no_files_found

test_no_files_found(tmp_path)

Test handling when no files are found.

Source code in gwframe/tests/test_cli.py

def test_no_files_found(self, tmp_path):
    """Test handling when no files are found."""
    empty_dir = tmp_path / "empty"
    empty_dir.mkdir()
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        ["rename", str(empty_dir), "-o", str(output_dir), "-m", "A=>B"],
    )

    assert result.exit_code != 0
    assert "no files" in result.stdout.lower()

TestImputeCommand

Tests for gwframe impute command.

test_impute_basic

test_impute_basic(tmp_path)

Test basic impute command.

Source code in gwframe/tests/test_cli.py

def test_impute_basic(self, tmp_path):
    """Test basic impute command."""
    # Create a file with NaN values
    input_file = tmp_path / "input.gwf"
    data_with_nan = np.array([1.0, 2.0, np.nan, 4.0, np.nan])

    gwframe.write(
        str(input_file),
        data_with_nan,
        start=1234567890.0,
        sample_rate=1,
        name="L1:CHAN",
    )

    output_dir = tmp_path / "output"

    result = runner.invoke(
        app, ["impute", str(input_file), "-o", str(output_dir), "-f", "0.0"]
    )

    assert result.exit_code == 0, result.stdout

    # Verify NaNs were replaced
    output_file = output_dir / input_file.name
    data = gwframe.read(str(output_file), "L1:CHAN")
    assert not np.any(np.isnan(data.array))
    assert data.array[2] == 0.0
    assert data.array[4] == 0.0

test_impute_in_place

test_impute_in_place(tmp_path)

Test impute with --in-place.

Source code in gwframe/tests/test_cli.py

def test_impute_in_place(self, tmp_path):
    """Test impute with --in-place."""
    test_file = tmp_path / "test.gwf"
    data_with_nan = np.array([1.0, np.nan, 3.0])

    gwframe.write(
        str(test_file),
        data_with_nan,
        start=1234567890.0,
        sample_rate=1,
        name="L1:CHAN",
    )

    result = runner.invoke(app, ["impute", str(test_file), "-i", "-f", "999.0"])

    assert result.exit_code == 0, result.stdout

    # Verify NaNs were replaced in place
    data = gwframe.read(str(test_file), "L1:CHAN")
    assert data.array[1] == 999.0

test_impute_specific_value

test_impute_specific_value(tmp_path)

Test impute replacing specific value.

Source code in gwframe/tests/test_cli.py

def test_impute_specific_value(self, tmp_path):
    """Test impute replacing specific value."""
    input_file = tmp_path / "input.gwf"
    data = np.array([1.0, -999.0, 3.0, -999.0])

    gwframe.write(
        str(input_file),
        data,
        start=1234567890.0,
        sample_rate=1,
        name="L1:CHAN",
    )

    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "impute",
            str(input_file),
            "-o",
            str(output_dir),
            "-r",
            "-999.0",
            "-f",
            "0.0",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify -999.0 values were replaced
    output_file = output_dir / input_file.name
    result_data = gwframe.read(str(output_file), "L1:CHAN")
    assert result_data.array[1] == 0.0
    assert result_data.array[3] == 0.0

TestInspectCommand

Tests for gwframe inspect command.

invalid_adc_file

invalid_adc_file(tmp_path)

Single frame: one invalid ADC, one valid ADC, one proc channel.

Source code in gwframe/tests/test_cli.py

@pytest.fixture
def invalid_adc_file(self, tmp_path):
    """Single frame: one invalid ADC, one valid ADC, one proc channel."""
    path = tmp_path / "invalid_adc.gwf"
    data = np.arange(64, dtype=np.float64)

    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.add_channel(
        "L1:BAD_ADC",
        np.ma.MaskedArray(data, mask=True),
        sample_rate=64,
        channel_type="adc",
        on_mask_loss="ignore",
    )
    frame.add_channel("L1:GOOD_ADC", data, sample_rate=64, channel_type="adc")
    frame.add_channel("L1:PROC", data, sample_rate=64, channel_type="proc")
    frame.write(str(path))
    return path

multi_frame_invalid_file

multi_frame_invalid_file(tmp_path)

3 frames; L1:BAD_ADC invalid in frames 0 and 2, valid in frame 1.

Source code in gwframe/tests/test_cli.py

@pytest.fixture
def multi_frame_invalid_file(self, tmp_path):
    """3 frames; L1:BAD_ADC invalid in frames 0 and 2, valid in frame 1."""
    path = tmp_path / "multi_invalid.gwf"
    data = np.arange(64, dtype=np.float64)

    with gwframe.FrameWriter(str(path)) as writer:
        for i in range(3):
            frame = gwframe.Frame(
                start=1234567890.0 + i, duration=1.0, name="L1", run=1
            )
            channel_data = data if i == 1 else np.ma.MaskedArray(data, mask=True)
            frame.add_channel(
                "L1:BAD_ADC",
                channel_data,
                sample_rate=64,
                channel_type="adc",
                on_mask_loss="ignore",
            )
            writer.write_frame(frame)
    return path

test_inspect_channel_type_counts

test_inspect_channel_type_counts(tmp_path)

Test inspect summary shows type counts.

Source code in gwframe/tests/test_cli.py

def test_inspect_channel_type_counts(self, tmp_path):
    """Test inspect summary shows type counts."""
    test_file = tmp_path / "type_counts.gwf"

    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.add_channel(
        "L1:ADC1",
        np.arange(100, dtype=np.float32),
        sample_rate=100,
        channel_type="adc",
    )
    frame.add_channel(
        "L1:PROC1",
        np.arange(100, dtype=np.float64),
        sample_rate=100,
        channel_type="proc",
    )
    frame.add_channel(
        "L1:PROC2",
        np.arange(100, dtype=np.float64),
        sample_rate=100,
        channel_type="proc",
    )
    frame.write(str(test_file))

    result = runner.invoke(app, ["inspect", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "1 adc" in output
    assert "2 proc" in output

test_inspect_channel_types

test_inspect_channel_types(tmp_path)

Test inspect correctly shows different channel types.

Source code in gwframe/tests/test_cli.py

def test_inspect_channel_types(self, tmp_path):
    """Test inspect correctly shows different channel types."""
    test_file = tmp_path / "mixed_types.gwf"

    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.add_channel(
        "L1:ADC_CHAN",
        np.arange(100, dtype=np.float32),
        sample_rate=100,
        channel_type="adc",
    )
    frame.add_channel(
        "L1:PROC_CHAN",
        np.arange(100, dtype=np.float64),
        sample_rate=100,
        channel_type="proc",
    )
    frame.write(str(test_file))

    result = runner.invoke(app, ["inspect", "-v", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "L1:ADC_CHAN" in output
    assert "L1:PROC_CHAN" in output
    assert "adc" in output
    assert "proc" in output

test_inspect_default

test_inspect_default(single_frame_file)

Test inspect with no verbosity (level 0) shows file summary.

Source code in gwframe/tests/test_cli.py

def test_inspect_default(self, single_frame_file):
    """Test inspect with no verbosity (level 0) shows file summary."""
    result = runner.invoke(app, ["inspect", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should show filename
    assert single_frame_file.name in output
    # Should show frame and channel counts
    assert "Frames:" in output
    assert "Channels:" in output
    # Should show compression
    assert "Compression:" in output
    # Should show GPS range
    assert "GPS:" in output
    # Should NOT show channel listing or frame table
    assert "CHAN1" not in output

test_inspect_detailed_channel_info

test_inspect_detailed_channel_info(tmp_path)

Test inspect -vvv shows correct per-channel details.

Source code in gwframe/tests/test_cli.py

def test_inspect_detailed_channel_info(self, tmp_path):
    """Test inspect -vvv shows correct per-channel details."""
    test_file = tmp_path / "detail.gwf"

    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    data = np.arange(16384, dtype=np.float64)
    frame.add_channel(
        "L1:STRAIN", data, sample_rate=16384, unit="strain", channel_type="proc"
    )
    frame.write(str(test_file))

    result = runner.invoke(app, ["inspect", "-vvv", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "L1:STRAIN" in output
    assert "float64" in output
    assert "16384" in output  # sample rate or sample count
    assert "strain" in output

test_inspect_empty_frame

test_inspect_empty_frame(tmp_path)

Test inspect on a frame with no channels.

Source code in gwframe/tests/test_cli.py

def test_inspect_empty_frame(self, tmp_path):
    """Test inspect on a frame with no channels."""
    test_file = tmp_path / "empty.gwf"

    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.write(str(test_file))

    result = runner.invoke(app, ["inspect", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "Channels:" in output
    # Should show 0 channels
    assert "0" in output

test_inspect_file_size

test_inspect_file_size(single_frame_file)

Test inspect shows file size.

Source code in gwframe/tests/test_cli.py

def test_inspect_file_size(self, single_frame_file):
    """Test inspect shows file size."""
    result = runner.invoke(app, ["inspect", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    # Should show size in some unit (B, KB, MB, GB)
    output = result.stdout
    assert any(unit in output for unit in (" B", " KB", " MB", " GB"))

test_inspect_gps_range

test_inspect_gps_range(single_frame_file, sample_data)

Test inspect shows correct GPS range.

Source code in gwframe/tests/test_cli.py

def test_inspect_gps_range(self, single_frame_file, sample_data):
    """Test inspect shows correct GPS range."""
    result = runner.invoke(app, ["inspect", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should contain the GPS start time
    assert str(int(sample_data["t0"])) in output

test_inspect_invalid_file

test_inspect_invalid_file(tmp_path)

Test inspect with invalid (non-GWF) file.

Source code in gwframe/tests/test_cli.py

def test_inspect_invalid_file(self, tmp_path):
    """Test inspect with invalid (non-GWF) file."""
    bad_file = tmp_path / "bad.gwf"
    bad_file.write_bytes(b"not a gwf file")

    result = runner.invoke(app, ["inspect", str(bad_file)])
    assert result.exit_code != 0

test_inspect_many_frames_truncation

test_inspect_many_frames_truncation(tmp_path)

Test inspect -vv truncates frame table for large files.

Source code in gwframe/tests/test_cli.py

def test_inspect_many_frames_truncation(self, tmp_path):
    """Test inspect -vv truncates frame table for large files."""
    test_file = tmp_path / "many_frames.gwf"

    with gwframe.FrameWriter(str(test_file)) as writer:
        for i in range(50):
            data = np.zeros(16, dtype=np.float32)
            writer.write(
                data,
                start=1234567890.0 + i,
                sample_rate=16,
                name="L1:TEST",
                unit="counts",
            )

    result = runner.invoke(app, ["inspect", "-vv", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should show truncation indicator
    assert "more" in output.lower()

test_inspect_multiframe

test_inspect_multiframe(multi_frame_file)

Test inspect on multi-frame file.

Source code in gwframe/tests/test_cli.py

def test_inspect_multiframe(self, multi_frame_file):
    """Test inspect on multi-frame file."""
    result = runner.invoke(app, ["inspect", str(multi_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "Frames:" in output
    # Multi-frame file has 3 frames
    assert "3" in output

test_inspect_multiframe_verbose

test_inspect_multiframe_verbose(multi_frame_file)

Test inspect -vv on multi-frame file shows all frames.

Source code in gwframe/tests/test_cli.py

def test_inspect_multiframe_verbose(self, multi_frame_file):
    """Test inspect -vv on multi-frame file shows all frames."""
    result = runner.invoke(app, ["inspect", "-vv", str(multi_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should list all 3 frames in frame table
    assert "GPS Start" in output

test_inspect_nonexistent_file

test_inspect_nonexistent_file(tmp_path)

Test inspect with nonexistent file.

Source code in gwframe/tests/test_cli.py

def test_inspect_nonexistent_file(self, tmp_path):
    """Test inspect with nonexistent file."""
    result = runner.invoke(app, ["inspect", str(tmp_path / "nonexistent.gwf")])
    assert result.exit_code != 0

test_inspect_sections_order_verbose_3

test_inspect_sections_order_verbose_3(single_frame_file)

Test that at -vvv, channels section appears before frames section.

Source code in gwframe/tests/test_cli.py

def test_inspect_sections_order_verbose_3(self, single_frame_file):
    """Test that at -vvv, channels section appears before frames section."""
    result = runner.invoke(app, ["inspect", "-vvv", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Find the section rule headers (not the summary line "Frames: N")
    # The rule headers contain the word surrounded by rule characters
    lines = output.split("\n")
    channels_line = next(
        (
            i
            for i, line in enumerate(lines)
            if "Channels" in line and "Dtype" not in line and ":" not in line
        ),
        None,
    )
    frames_line = next(
        (i for i, line in enumerate(lines) if "Frames" in line and ":" not in line),
        None,
    )
    # Both section headers should exist
    assert channels_line is not None
    assert frames_line is not None
    # Channels section should come before Frames section
    assert channels_line < frames_line

test_inspect_sections_order_verbose_5

test_inspect_sections_order_verbose_5(invalid_adc_file)

Sections at -vvvvv: Channels < Frames < Invalid channels < Preview.

Source code in gwframe/tests/test_cli.py

def test_inspect_sections_order_verbose_5(self, invalid_adc_file):
    """Sections at -vvvvv: Channels < Frames < Invalid channels < Preview."""
    result = runner.invoke(app, ["inspect", "-vvvvv", str(invalid_adc_file)])
    assert result.exit_code == 0, result.stdout

    lines = result.stdout.splitlines()
    channels_line = next(
        i
        for i, line in enumerate(lines)
        if "Channels" in line and "Dtype" not in line and ":" not in line
    )
    frames_line = next(
        i for i, line in enumerate(lines) if "Frames" in line and ":" not in line
    )
    invalid_line = next(
        i for i, line in enumerate(lines) if "Invalid channels" in line
    )
    preview_line = next(i for i, line in enumerate(lines) if "Data preview" in line)
    assert channels_line < frames_line < invalid_line < preview_line

test_inspect_verbose_1

test_inspect_verbose_1(single_frame_file)

Test inspect -v shows channel listing.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_1(self, single_frame_file):
    """Test inspect -v shows channel listing."""
    result = runner.invoke(app, ["inspect", "-v", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should show file summary
    assert "Frames:" in output
    # Should show channel listing with names
    assert "L1:CHAN1" in output
    assert "L1:CHAN2" in output
    assert "L1:CHAN3" in output
    # Should show channel type
    assert "proc" in output
    # Should NOT show per-frame table
    assert "Frame #" not in output

test_inspect_verbose_2

test_inspect_verbose_2(single_frame_file)

Test inspect -vv shows channels and per-frame table.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_2(self, single_frame_file):
    """Test inspect -vv shows channels and per-frame table."""
    result = runner.invoke(app, ["inspect", "-vv", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should show channel listing
    assert "L1:CHAN1" in output
    # Should show frame table
    assert "GPS Start" in output
    assert "Duration" in output
    # Should show frame name
    assert "TEST" in output

test_inspect_verbose_3

test_inspect_verbose_3(single_frame_file)

Test inspect -vvv shows detailed channel info and frame table.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_3(self, single_frame_file):
    """Test inspect -vvv shows detailed channel info and frame table."""
    result = runner.invoke(app, ["inspect", "-vvv", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    # Should show detailed channel table (not simple listing)
    assert "Dtype" in output
    assert "Rate (Hz)" in output
    assert "Samples" in output
    assert "Unit" in output
    # Should show channel names
    assert "L1:CHAN1" in output
    # Should show frame table
    assert "GPS Start" in output

test_inspect_verbose_3_valid_column

test_inspect_verbose_3_valid_column(invalid_adc_file)

-vvv shows a Valid column marking invalid ADC channels.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_3_valid_column(self, invalid_adc_file):
    """-vvv shows a Valid column marking invalid ADC channels."""
    result = runner.invoke(app, ["inspect", "-vvv", str(invalid_adc_file)])
    assert result.exit_code == 0, result.stdout

    lines = result.stdout.splitlines()
    assert any("Valid" in line for line in lines)
    bad_line = next(line for line in lines if "L1:BAD_ADC" in line)
    assert " no" in bad_line
    good_line = next(line for line in lines if "L1:GOOD_ADC" in line)
    assert "yes" in good_line
    # Proc/sim channels carry no dataValid flag and are always valid
    proc_line = next(line for line in lines if "L1:PROC" in line)
    assert "yes" in proc_line

test_inspect_verbose_3_valid_column_all_frames

test_inspect_verbose_3_valid_column_all_frames(tmp_path)

-vvv Valid column reflects all frames, not just frame 0.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_3_valid_column_all_frames(self, tmp_path):
    """-vvv Valid column reflects all frames, not just frame 0."""
    path = tmp_path / "later_invalid.gwf"
    data = np.arange(64, dtype=np.float64)

    with gwframe.FrameWriter(str(path)) as writer:
        for i in range(2):
            frame = gwframe.Frame(
                start=1234567890.0 + i, duration=1.0, name="L1", run=1
            )
            # Valid in frame 0, invalid only in frame 1
            channel_data = np.ma.MaskedArray(data, mask=True) if i == 1 else data
            frame.add_channel(
                "L1:BAD_ADC",
                channel_data,
                sample_rate=64,
                channel_type="adc",
                on_mask_loss="ignore",
            )
            writer.write_frame(frame)

    result = runner.invoke(app, ["inspect", "-vvv", str(path)])
    assert result.exit_code == 0, result.stdout

    lines = result.stdout.splitlines()
    bad_line = next(line for line in lines if "L1:BAD_ADC" in line)
    assert " no" in bad_line

test_inspect_verbose_4_all_valid

test_inspect_verbose_4_all_valid(tmp_path)

-vvvv reports when all ADC channels are valid.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_4_all_valid(self, tmp_path):
    """-vvvv reports when all ADC channels are valid."""
    path = tmp_path / "all_valid.gwf"
    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.add_channel(
        "L1:ADC",
        np.arange(64, dtype=np.float64),
        sample_rate=64,
        channel_type="adc",
    )
    frame.write(str(path))

    result = runner.invoke(app, ["inspect", "-vvvv", str(path)])
    assert result.exit_code == 0, result.stdout
    assert "All ADC channels valid" in result.stdout

test_inspect_verbose_4_invalid_section

test_inspect_verbose_4_invalid_section(invalid_adc_file)

-vvvv adds the invalid-channel report.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_4_invalid_section(self, invalid_adc_file):
    """-vvvv adds the invalid-channel report."""
    result = runner.invoke(app, ["inspect", "-vvvv", str(invalid_adc_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "Invalid channels" in output
    lines = output.splitlines()
    # The invalid table row carries the raw dataValid flag as hex
    flag_line = next(
        line for line in lines if "L1:BAD_ADC" in line and "0x" in line
    )
    assert "all" in flag_line
    # Valid channels don't appear in the invalid section
    assert not any("L1:GOOD_ADC" in line and "0x" in line for line in lines)

test_inspect_verbose_4_multiframe

test_inspect_verbose_4_multiframe(multi_frame_invalid_file)

-vvvv lists the frames each invalid channel is flagged in.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_4_multiframe(self, multi_frame_invalid_file):
    """-vvvv lists the frames each invalid channel is flagged in."""
    result = runner.invoke(app, ["inspect", "-vvvv", str(multi_frame_invalid_file)])
    assert result.exit_code == 0, result.stdout

    lines = result.stdout.splitlines()
    flag_line = next(
        line for line in lines if "L1:BAD_ADC" in line and "0x" in line
    )
    assert "0, 2" in flag_line

test_inspect_verbose_4_no_adc

test_inspect_verbose_4_no_adc(single_frame_file)

-vvvv on a proc-only file explains dataValid is ADC-only.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_4_no_adc(self, single_frame_file):
    """-vvvv on a proc-only file explains dataValid is ADC-only."""
    result = runner.invoke(app, ["inspect", "-vvvv", str(single_frame_file)])
    assert result.exit_code == 0, result.stdout
    assert "No ADC channels" in result.stdout

test_inspect_verbose_5_empty_frame

test_inspect_verbose_5_empty_frame(tmp_path)

-vvvvv on a channel-less frame.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_5_empty_frame(self, tmp_path):
    """-vvvvv on a channel-less frame."""
    test_file = tmp_path / "empty.gwf"
    frame = gwframe.Frame(start=1234567890.0, duration=1.0, name="L1", run=1)
    frame.write(str(test_file))

    result = runner.invoke(app, ["inspect", "-vvvvv", str(test_file)])
    assert result.exit_code == 0, result.stdout
    assert "(no channels)" in result.stdout

test_inspect_verbose_5_preview

test_inspect_verbose_5_preview(invalid_adc_file)

-vvvvv adds a per-frame data preview.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_5_preview(self, invalid_adc_file):
    """-vvvvv adds a per-frame data preview."""
    result = runner.invoke(app, ["inspect", "-vvvvv", str(invalid_adc_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "Data preview" in output
    assert "Frame 0" in output
    # Abbreviated array: first value, ellipsis, last value
    assert "..." in output
    assert "63.0" in output

test_inspect_verbose_5_truncation

test_inspect_verbose_5_truncation(tmp_path)

-vvvvv truncates the preview to first/last 5 frames.

Source code in gwframe/tests/test_cli.py

def test_inspect_verbose_5_truncation(self, tmp_path):
    """-vvvvv truncates the preview to first/last 5 frames."""
    test_file = tmp_path / "many_frames.gwf"

    with gwframe.FrameWriter(str(test_file)) as writer:
        for i in range(50):
            data = np.zeros(16, dtype=np.float32)
            writer.write(
                data,
                start=1234567890.0 + i,
                sample_rate=16,
                name="L1:TEST",
                unit="counts",
            )

    result = runner.invoke(app, ["inspect", "-vvvvv", str(test_file)])
    assert result.exit_code == 0, result.stdout

    output = result.stdout
    assert "frames omitted" in output
    assert "Frame 49" in output
    assert "Frame 25" not in output

TestOutputOptionValidation

CLI validation for mutually-exclusive --in-place and --output-dir.

test_rename_rejects_in_place_and_output_dir

test_rename_rejects_in_place_and_output_dir(single_frame_file, tmp_path)

Passing both --in-place and --output-dir should exit non-zero.

Source code in gwframe/tests/test_cli.py

def test_rename_rejects_in_place_and_output_dir(self, single_frame_file, tmp_path):
    """Passing both --in-place and --output-dir should exit non-zero."""
    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "--in-place",
            "--output-dir",
            str(tmp_path / "out"),
            "-m",
            "L1:CHAN1=>L1:RENAMED",
        ],
    )
    assert result.exit_code == 1
    assert "mutually exclusive" in result.stdout

TestRecompressCommand

Tests for gwframe recompress command.

test_recompress_basic

test_recompress_basic(single_frame_file, tmp_path)

Test basic recompress command.

Source code in gwframe/tests/test_cli.py

def test_recompress_basic(self, single_frame_file, tmp_path):
    """Test basic recompress command."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "recompress",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-c",
            "GZIP",
            "-l",
            "9",
        ],
    )

    assert result.exit_code == 0, result.stdout
    assert (output_dir / single_frame_file.name).exists()

    # Verify data integrity after recompression
    output_file = output_dir / single_frame_file.name
    original_data = gwframe.read(str(single_frame_file), "L1:CHAN1")
    recompressed_data = gwframe.read(str(output_file), "L1:CHAN1")
    assert np.allclose(original_data.array, recompressed_data.array)

test_recompress_in_place

test_recompress_in_place(single_frame_file, tmp_path)

Test recompress with --in-place.

Source code in gwframe/tests/test_cli.py

def test_recompress_in_place(self, single_frame_file, tmp_path):
    """Test recompress with --in-place."""
    test_file = tmp_path / "test.gwf"
    shutil.copy(single_frame_file, test_file)

    # Read original data
    original_data = gwframe.read(str(test_file), "L1:CHAN1")

    result = runner.invoke(app, ["recompress", str(test_file), "-i", "-c", "RAW"])

    assert result.exit_code == 0, result.stdout

    # Verify data integrity after in-place recompression
    recompressed_data = gwframe.read(str(test_file), "L1:CHAN1")
    assert np.allclose(original_data.array, recompressed_data.array)

TestRecompressInvalidCompression

cli.py:789-794 — invalid compression string should exit cleanly.

TestRenameCommand

Tests for gwframe rename command.

test_rename_basic

test_rename_basic(single_frame_file, tmp_path)

Test basic rename command.

Source code in gwframe/tests/test_cli.py

def test_rename_basic(self, single_frame_file, tmp_path):
    """Test basic rename command."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-m",
            "L1:CHAN1=>L1:RENAMED1",
            "-m",
            "L1:CHAN2=>L1:RENAMED2",
        ],
    )

    assert result.exit_code == 0, result.stdout
    assert output_dir.exists()

    # Verify channels were renamed
    output_file = output_dir / single_frame_file.name
    channels = get_channels(str(output_file))
    assert "L1:RENAMED1" in channels
    assert "L1:RENAMED2" in channels
    assert "L1:CHAN1" not in channels
    assert "L1:CHAN2" not in channels

test_rename_in_place

test_rename_in_place(single_frame_file, tmp_path)

Test rename with --in-place.

Source code in gwframe/tests/test_cli.py

def test_rename_in_place(self, single_frame_file, tmp_path):
    """Test rename with --in-place."""
    # Copy file to tmp_path so we can modify it
    test_file = tmp_path / "test.gwf"
    shutil.copy(single_frame_file, test_file)

    result = runner.invoke(
        app,
        [
            "rename",
            str(test_file),
            "-i",
            "-m",
            "L1:CHAN1=>L1:RENAMED",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify file was modified in place
    channels = get_channels(str(test_file))
    assert "L1:RENAMED" in channels
    assert "L1:CHAN1" not in channels

test_rename_mutual_exclusivity

test_rename_mutual_exclusivity(single_frame_file, tmp_path)

Test that --in-place and -o are mutually exclusive.

Source code in gwframe/tests/test_cli.py

def test_rename_mutual_exclusivity(self, single_frame_file, tmp_path):
    """Test that --in-place and -o are mutually exclusive."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-i",
            "-m",
            "L1:CHAN1=>TEST:RENAMED",
        ],
    )

    assert result.exit_code != 0
    assert "mutually exclusive" in result.stdout.lower()

test_rename_single_file_output

test_rename_single_file_output(single_frame_file, tmp_path)

Test rename with single file output.

Source code in gwframe/tests/test_cli.py

def test_rename_single_file_output(self, single_frame_file, tmp_path):
    """Test rename with single file output."""
    output_file = tmp_path / "output.gwf"

    result = runner.invoke(
        app,
        [
            "rename",
            str(single_frame_file),
            "-o",
            str(output_file),
            "-m",
            "L1:CHAN1=>L1:RENAMED",
        ],
    )

    assert result.exit_code == 0, result.stdout
    assert output_file.exists()

    # Verify channels were renamed
    channels = get_channels(str(output_file))
    assert "L1:RENAMED" in channels

TestReplaceCommand

Tests for gwframe replace command.

test_replace_basic

test_replace_basic(tmp_path, sample_data, create_test_frame)

Replace a channel's data from an update file.

Source code in gwframe/tests/test_cli.py

def test_replace_basic(self, tmp_path, sample_data, create_test_frame):
    """Replace a channel's data from an update file."""
    n_samples = sample_data["n_samples"]
    base_file = tmp_path / "base.gwf"
    update_file = tmp_path / "update.gwf"
    output_dir = tmp_path / "output"

    base_data = np.zeros(n_samples, dtype=np.float32)
    update_data = np.ones(n_samples, dtype=np.float32)

    with gwframe.FrameWriter(str(base_file)) as writer:
        frame = create_test_frame(
            sample_data["t0"],
            sample_data["duration"],
            {"L1:CHAN1": base_data, "L1:CHAN2": base_data * 2},
        )
        writer.write_frame(frame)

    with gwframe.FrameWriter(str(update_file)) as writer:
        frame = create_test_frame(
            sample_data["t0"],
            sample_data["duration"],
            {"L1:CHAN1": update_data},
        )
        writer.write_frame(frame)

    result = runner.invoke(
        app,
        [
            "replace",
            str(base_file),
            "--update",
            str(update_file),
            "-o",
            str(output_dir),
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code == 0, result.stdout

    output_file = output_dir / base_file.name
    assert output_file.exists()

    # Replaced channel should match the update file
    replaced = gwframe.read(str(output_file), "L1:CHAN1")
    np.testing.assert_array_equal(update_data, replaced.array)

    # Untouched channel should still match the base
    untouched = gwframe.read(str(output_file), "L1:CHAN2")
    np.testing.assert_array_equal(base_data * 2, untouched.array)

test_replace_empty_base_directory

test_replace_empty_base_directory(tmp_path, sample_data, create_test_frame)

An empty base directory should exit 1 with 'No base files found'.

Source code in gwframe/tests/test_cli.py

def test_replace_empty_base_directory(
    self, tmp_path, sample_data, create_test_frame
):
    """An empty base directory should exit 1 with 'No base files found'."""
    empty_base = tmp_path / "empty_base"
    empty_base.mkdir()

    update_file = tmp_path / "update.gwf"
    with gwframe.FrameWriter(str(update_file)) as writer:
        frame = create_test_frame(
            sample_data["t0"],
            sample_data["duration"],
            {"L1:CHAN1": np.ones(sample_data["n_samples"], dtype=np.float32)},
        )
        writer.write_frame(frame)

    result = runner.invoke(
        app,
        [
            "replace",
            str(empty_base),
            "--update",
            str(update_file),
            "-o",
            str(tmp_path / "out"),
        ],
    )
    assert result.exit_code == 1
    assert "No base files found" in result.stdout

test_replace_empty_update_directory

test_replace_empty_update_directory(tmp_path, sample_data, create_test_frame)

An empty update directory should exit 1 with 'No update files found'.

Source code in gwframe/tests/test_cli.py

def test_replace_empty_update_directory(
    self, tmp_path, sample_data, create_test_frame
):
    """An empty update directory should exit 1 with 'No update files found'."""
    base_file = tmp_path / "base.gwf"
    with gwframe.FrameWriter(str(base_file)) as writer:
        frame = create_test_frame(
            sample_data["t0"],
            sample_data["duration"],
            {"L1:CHAN1": np.zeros(sample_data["n_samples"], dtype=np.float32)},
        )
        writer.write_frame(frame)

    empty_update = tmp_path / "empty_update"
    empty_update.mkdir()

    result = runner.invoke(
        app,
        [
            "replace",
            str(base_file),
            "--update",
            str(empty_update),
            "-o",
            str(tmp_path / "out"),
        ],
    )
    assert result.exit_code == 1
    assert "No update files found" in result.stdout

TestResizeCommand

Tests for gwframe resize command.

test_resize_basic

test_resize_basic(multi_frame_file, tmp_path, sample_data)

Test basic resize command.

Source code in gwframe/tests/test_cli.py

def test_resize_basic(self, multi_frame_file, tmp_path, sample_data):
    """Test basic resize command."""
    output_dir = tmp_path / "output"
    target_duration = sample_data["duration"] / 2

    result = runner.invoke(
        app,
        [
            "resize",
            str(multi_frame_file),
            "-o",
            str(output_dir),
            "-d",
            str(target_duration),
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify frames were resized
    output_file = output_dir / multi_frame_file.name
    frames = list(gwframe.read_frames(str(output_file)))
    # Each of 3 original frames split in 2 = 6 frames
    assert len(frames) == 6
    for frame in frames:
        assert abs(frame.duration - target_duration) < 1e-9

test_resize_in_place

test_resize_in_place(multi_frame_file, tmp_path, sample_data)

Test resize with --in-place.

Source code in gwframe/tests/test_cli.py

def test_resize_in_place(self, multi_frame_file, tmp_path, sample_data):
    """Test resize with --in-place."""
    test_file = tmp_path / "test.gwf"
    shutil.copy(multi_frame_file, test_file)
    target_duration = sample_data["duration"] / 2

    result = runner.invoke(
        app, ["resize", str(test_file), "-i", "-d", str(target_duration)]
    )

    assert result.exit_code == 0, result.stdout

    # Verify frames were resized in place
    frames = list(gwframe.read_frames(str(test_file)))
    assert len(frames) == 6

TestSelectCommand

Tests for gwframe select command.

test_select_basic

test_select_basic(single_frame_file, tmp_path)

Test basic select command.

Source code in gwframe/tests/test_cli.py

def test_select_basic(self, single_frame_file, tmp_path):
    """Test basic select command."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "select",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify only selected channel remains
    output_file = output_dir / single_frame_file.name
    channels = get_channels(str(output_file))
    assert "L1:CHAN1" in channels
    assert "L1:CHAN2" not in channels
    assert "L1:CHAN3" not in channels

test_select_in_place

test_select_in_place(single_frame_file, tmp_path)

Test select with --in-place.

Source code in gwframe/tests/test_cli.py

def test_select_in_place(self, single_frame_file, tmp_path):
    """Test select with --in-place."""
    test_file = tmp_path / "test.gwf"
    shutil.copy(single_frame_file, test_file)

    result = runner.invoke(app, ["select", str(test_file), "-i", "-c", "L1:CHAN1"])

    assert result.exit_code == 0, result.stdout

    # Verify only selected channel remains
    channels = get_channels(str(test_file))
    assert "L1:CHAN1" in channels
    assert "L1:CHAN2" not in channels

test_select_multiple_channels

test_select_multiple_channels(single_frame_file, tmp_path)

Test selecting multiple channels.

Source code in gwframe/tests/test_cli.py

def test_select_multiple_channels(self, single_frame_file, tmp_path):
    """Test selecting multiple channels."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "select",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-c",
            "L1:CHAN1",
            "-c",
            "L1:CHAN3",
        ],
    )

    assert result.exit_code == 0, result.stdout

    # Verify only selected channels remain
    output_file = output_dir / single_frame_file.name
    channels = get_channels(str(output_file))
    assert "L1:CHAN1" in channels
    assert "L1:CHAN3" in channels
    assert "L1:CHAN2" not in channels

test_select_mutual_exclusivity

test_select_mutual_exclusivity(single_frame_file, tmp_path)

Test that --in-place and -o are mutually exclusive.

Source code in gwframe/tests/test_cli.py

def test_select_mutual_exclusivity(self, single_frame_file, tmp_path):
    """Test that --in-place and -o are mutually exclusive."""
    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "select",
            str(single_frame_file),
            "-o",
            str(output_dir),
            "-i",
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code != 0
    assert "mutually exclusive" in result.stdout.lower()

test_select_recursive

test_select_recursive(tmp_path, sample_data, create_test_frame)

Test select with --recursive flag.

Source code in gwframe/tests/test_cli.py

def test_select_recursive(self, tmp_path, sample_data, create_test_frame):
    """Test select with --recursive flag."""
    # Create nested directory structure
    sub_dir = tmp_path / "data" / "sub"
    sub_dir.mkdir(parents=True)
    n_samples = sample_data["n_samples"]

    file_path = sub_dir / "nested.gwf"
    with gwframe.FrameWriter(str(file_path)) as writer:
        channels_data = {
            "L1:CHAN1": np.arange(n_samples, dtype=np.float32),
            "L1:CHAN2": np.arange(n_samples, dtype=np.float32) * 2,
        }
        frame = create_test_frame(
            sample_data["t0"], sample_data["duration"], channels_data
        )
        writer.write_frame(frame)

    output_dir = tmp_path / "output"

    result = runner.invoke(
        app,
        [
            "select",
            str(tmp_path / "data"),
            "-o",
            str(output_dir),
            "-r",
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code == 0, result.stdout

    output_files = list(output_dir.glob("*.gwf"))
    assert len(output_files) == 1
    channels = get_channels(str(output_files[0]))
    assert "L1:CHAN1" in channels
    assert "L1:CHAN2" not in channels

test_select_single_file_output

test_select_single_file_output(single_frame_file, tmp_path)

Test select with single file output.

Source code in gwframe/tests/test_cli.py

def test_select_single_file_output(self, single_frame_file, tmp_path):
    """Test select with single file output."""
    output_file = tmp_path / "output.gwf"

    result = runner.invoke(
        app,
        [
            "select",
            str(single_frame_file),
            "-o",
            str(output_file),
            "-c",
            "L1:CHAN1",
        ],
    )

    assert result.exit_code == 0, result.stdout
    assert output_file.exists()

    channels = get_channels(str(output_file))
    assert "L1:CHAN1" in channels
    assert "L1:CHAN2" not in channels

TestTransformCommand

Tests for gwframe transform command (chained stages).

raw_dir

raw_dir(tmp_path, sample_data, create_test_frame)

Three conventionally named files, two contiguous 1 s frames each.

Source code in gwframe/tests/test_cli.py

@pytest.fixture
def raw_dir(self, tmp_path, sample_data, create_test_frame):
    """Three conventionally named files, two contiguous 1 s frames each."""
    raw = tmp_path / "raw"
    raw.mkdir()
    n = sample_data["n_samples"]
    t0 = int(sample_data["t0"])
    for i in range(3):
        start = t0 + i * 2
        with FrameWriter(raw / f"L-L1_RAW-{start}-2.gwf") as writer:
            for j in range(2):
                data = np.arange(n, dtype=np.float32)
                data[0] = np.nan
                frame = create_test_frame(
                    start + j,
                    1.0,
                    {"L1:CHAN1": data, "L1:CHAN2": data * 2},
                    frame_number=i * 2 + j,
                )
                writer.write_frame(frame)
    return raw

test_chain

test_chain(raw_dir, tmp_path)

Stages apply in order, each with its own options.

Source code in gwframe/tests/test_cli.py

def test_chain(self, raw_dir, tmp_path):
    """Stages apply in order, each with its own options."""
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        [
            "transform",
            str(raw_dir),
            "-o",
            str(out),
            "--compression",
            "GZIP",
            "rename",
            "-m",
            "L1:CHAN1=>L1:STRAIN",
            "drop",
            "-c",
            "L1:CHAN2",
            "impute",
            "-c",
            "L1:STRAIN",
            "resize",
            "-d",
            "0.5",
        ],
    )
    assert result.exit_code == 0, result.stdout
    assert "Wrote 3 file(s)" in result.stdout

    files = sorted(out.glob("*.gwf"))
    assert [f.name for f in files] == sorted(p.name for p in raw_dir.iterdir())
    info = get_info(files[0])
    assert info.channels == ["L1:STRAIN"]
    assert info.num_frames == 4
    assert info.compression == Compression.GZIP
    assert gwframe.read(files[0], "L1:STRAIN").array[0] == 0.0

test_dry_run

test_dry_run(raw_dir, tmp_path)

--dry-run prints the plan and warnings and writes nothing.

Source code in gwframe/tests/test_cli.py

def test_dry_run(self, raw_dir, tmp_path):
    """--dry-run prints the plan and warnings and writes nothing."""
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        [
            "transform",
            str(raw_dir),
            "-o",
            str(out),
            "--dry-run",
            "drop",
            "-c",
            "L1:NOPE",
            "resize",
            "-d",
            "2",
        ],
    )
    assert result.exit_code == 0, result.stdout
    assert "1. drop L1:NOPE" in result.stdout
    assert "2. resize 2s" in result.stdout
    assert "Channels: 2 in -> 2 out" in result.stdout
    assert "Warning: stage 1 (drop)" in result.stdout
    assert "L1:NOPE" in result.stdout
    assert "nothing written" in result.stdout
    assert not out.exists()

test_file_duration_and_prefix

test_file_duration_and_prefix(raw_dir, tmp_path)

--file-duration re-chunks the stream; --prefix names the files.

Source code in gwframe/tests/test_cli.py

def test_file_duration_and_prefix(self, raw_dir, tmp_path):
    """--file-duration re-chunks the stream; --prefix names the files."""
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        [
            "transform",
            str(raw_dir),
            "-o",
            str(out),
            "--file-duration",
            "4",
            "--prefix",
            "L-L1_CURATED",
            "resize",
            "-d",
            "2",
        ],
    )
    assert result.exit_code == 0, result.stdout
    assert "files of 4s" in result.stdout
    names = sorted(p.name for p in out.glob("*.gwf"))
    assert names == [
        "L-L1_CURATED-1234567890-4.gwf",
        "L-L1_CURATED-1234567894-2.gwf",
    ]
    info = get_info(out / names[0])
    assert [f.duration for f in info.frames] == [2.0, 2.0]

test_file_duration_default_prefix

test_file_duration_default_prefix(raw_dir, tmp_path)

The prefix defaults to that of the first input file.

Source code in gwframe/tests/test_cli.py

def test_file_duration_default_prefix(self, raw_dir, tmp_path):
    """The prefix defaults to that of the first input file."""
    out = tmp_path / "out"
    result = runner.invoke(
        app, ["transform", str(raw_dir), "-o", str(out), "--file-duration", "6"]
    )
    assert result.exit_code == 0, result.stdout
    assert [p.name for p in out.glob("*.gwf")] == ["L-L1_RAW-1234567890-6.gwf"]

test_fill_gaps

test_fill_gaps(raw_dir, tmp_path)

fill-gaps bridges a gap so re-chunked output stays contiguous.

Source code in gwframe/tests/test_cli.py

def test_fill_gaps(self, raw_dir, tmp_path):
    """fill-gaps bridges a gap so re-chunked output stays contiguous."""
    (raw_dir / "L-L1_RAW-1234567892-2.gwf").unlink()  # 2 s hole in the middle
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        [
            "transform",
            str(raw_dir),
            "-o",
            str(out),
            "--file-duration",
            "6",
            "fill-gaps",
            "-f",
            "-1",
        ],
    )
    assert result.exit_code == 0, result.stdout
    assert "1. fill-gaps with -1" in result.stdout
    assert [p.name for p in out.glob("*.gwf")] == ["L-L1_RAW-1234567890-6.gwf"]
    filled = gwframe.read(
        out / "L-L1_RAW-1234567890-6.gwf",
        "L1:CHAN1",
        start=1234567892,
        end=1234567894,
    )
    assert np.all(filled.array == -1)

test_matches_standalone_commands

test_matches_standalone_commands(raw_dir, tmp_path)

One transform pass equals the standalone commands chained via dirs.

Source code in gwframe/tests/test_cli.py

def test_matches_standalone_commands(self, raw_dir, tmp_path):
    """One transform pass equals the standalone commands chained via dirs."""
    chained = tmp_path / "chained"
    for step, args in enumerate(
        [
            ["rename", "-m", "L1:CHAN1=>L1:STRAIN"],
            ["drop", "-c", "L1:CHAN2"],
            ["resize", "-d", "0.5"],
        ]
    ):
        src = raw_dir if step == 0 else chained / str(step - 1)
        result = runner.invoke(
            app, [args[0], str(src), "-o", str(chained / str(step)), *args[1:]]
        )
        assert result.exit_code == 0, result.stdout

    out = tmp_path / "out"
    result = runner.invoke(
        app,
        [
            "transform",
            str(raw_dir),
            "-o",
            str(out),
            "rename",
            "-m",
            "L1:CHAN1=>L1:STRAIN",
            "drop",
            "-c",
            "L1:CHAN2",
            "resize",
            "-d",
            "0.5",
        ],
    )
    assert result.exit_code == 0, result.stdout

    result = runner.invoke(app, ["validate", str(chained / "2"), str(out)])
    assert result.exit_code == 0, result.stdout

test_no_stages_rewrites

test_no_stages_rewrites(raw_dir, tmp_path)

Without stages the files are rewritten (recompressed).

Source code in gwframe/tests/test_cli.py

def test_no_stages_rewrites(self, raw_dir, tmp_path):
    """Without stages the files are rewritten (recompressed)."""
    out = tmp_path / "out"
    result = runner.invoke(
        app, ["transform", str(raw_dir), "-o", str(out), "--compression", "RAW"]
    )
    assert result.exit_code == 0, result.stdout
    assert "no stages" in result.stdout
    assert get_info(next(out.glob("*.gwf"))).compression == Compression.RAW

test_options_after_inputs

test_options_after_inputs(raw_dir, tmp_path)

Group options may follow the input paths.

Source code in gwframe/tests/test_cli.py

def test_options_after_inputs(self, raw_dir, tmp_path):
    """Group options may follow the input paths."""
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        ["transform", str(raw_dir), "-r", "-o", str(out), "drop", "-c", "L1:CHAN2"],
    )
    assert result.exit_code == 0, result.stdout
    assert get_channels(next(out.glob("*.gwf"))) == ["L1:CHAN1"]

test_preflight_error

test_preflight_error(raw_dir, tmp_path)

A chain that leaves no channels is rejected before any write.

Source code in gwframe/tests/test_cli.py

def test_preflight_error(self, raw_dir, tmp_path):
    """A chain that leaves no channels is rejected before any write."""
    out = tmp_path / "out"
    result = runner.invoke(
        app,
        ["transform", str(raw_dir), "-o", str(out), "select", "-c", "L1:NOPE"],
    )
    assert result.exit_code == 1
    assert "no channels would remain" in result.stdout
    assert not out.exists()

test_recipe_file

test_recipe_file(raw_dir, tmp_path)

@FILE splices a recipe of stages into the command line.

Source code in gwframe/tests/test_cli.py

def test_recipe_file(self, raw_dir, tmp_path):
    """@FILE splices a recipe of stages into the command line."""
    recipe = tmp_path / "curated.recipe"
    recipe.write_text(
        "# curated dataset\n"
        "rename \\\n"
        '    -m "L1:CHAN1=>L1:STRAIN"\n'
        "drop -c L1:CHAN2   # trailing comment\n"
        "\n"
        "resize -d 0.5\n"
    )
    out = tmp_path / "out"
    result = runner.invoke(
        app, ["transform", str(raw_dir), "-o", str(out), f"@{recipe}"]
    )
    assert result.exit_code == 0, result.stdout
    assert "1. rename L1:CHAN1=>L1:STRAIN" in result.stdout
    assert "3. resize 0.5s" in result.stdout
    info = get_info(next(out.glob("*.gwf")))
    assert info.channels == ["L1:STRAIN"]
    assert info.num_frames == 4

test_stage_option_before_stage_hint

test_stage_option_before_stage_hint(raw_dir, tmp_path)

An option given before its stage name gets a pointer to the stages.

Source code in gwframe/tests/test_cli.py

def test_stage_option_before_stage_hint(self, raw_dir, tmp_path):
    """An option given before its stage name gets a pointer to the stages."""
    result = runner.invoke(
        app,
        ["transform", str(raw_dir), "-o", str(tmp_path / "o"), "-c", "X", "drop"],
    )
    assert result.exit_code == 2
    assert "must follow its name" in result.output
    assert "rename" in result.output

TestValidateCommand

Tests for gwframe validate command.

differing_file

differing_file(single_frame_file, tmp_path, sample_data)

A copy of single_frame_file with one channel's data perturbed.

Source code in gwframe/tests/test_cli.py

@pytest.fixture
def differing_file(self, single_frame_file, tmp_path, sample_data):
    """A copy of single_frame_file with one channel's data perturbed."""
    file_path = tmp_path / "differing.gwf"
    data = gwframe.read(str(single_frame_file), "L1:CHAN1")
    perturbed = data.array.copy()
    perturbed[0] += 1.0

    frame = gwframe.Frame(
        start=sample_data["t0"],
        duration=sample_data["duration"],
        name="TEST",
        run=1,
        frame_number=0,
    )
    frame.add_channel(
        "L1:CHAN1", perturbed, sample_rate=data.sample_rate, unit="counts"
    )
    for name in ("L1:CHAN2", "L1:CHAN3"):
        other = gwframe.read(str(single_frame_file), name)
        frame.add_channel(
            name, other.array, sample_rate=other.sample_rate, unit="counts"
        )
    with gwframe.FrameWriter(str(file_path)) as writer:
        writer.write_frame(frame)
    return file_path

test_common_shorthand

test_common_shorthand(single_frame_file, tmp_path)

--common behaves like --common-channels --common-time-spans.

Source code in gwframe/tests/test_cli.py

def test_common_shorthand(self, single_frame_file, tmp_path):
    """--common behaves like --common-channels --common-time-spans."""
    left_dir = tmp_path / "dir1"
    right_dir = tmp_path / "dir2"
    left_dir.mkdir()
    right_dir.mkdir()
    shutil.copy(single_frame_file, left_dir / "a.gwf")
    shutil.copy(single_frame_file, left_dir / "extra.gwf")
    resize_frames([str(left_dir / "extra.gwf")], None, 0.5, in_place=True)
    # Right side has a channel dropped and no "extra" file
    drop_channels([str(single_frame_file)], str(right_dir), ["L1:CHAN3"])

    result = runner.invoke(app, ["validate", str(left_dir), str(right_dir)])
    assert result.exit_code == 1

    result = runner.invoke(
        app, ["validate", str(left_dir), str(right_dir), "--common"]
    )
    assert result.exit_code == 0, result.stdout