test_compare

Tests for the gwframe.compare module.

TestCompareChannels

Channel set handling.

TestCompareData

Sample data comparison.

TestCompareFilesIdentical

Comparison of identical files.

TestCompareMasks

Mask (dataValid) comparison.

TestCompareMetadata

Per-channel metadata comparison.

TestComparePaths

Directory and path-level comparison.

dir_pair

dir_pair(tmp_path, base_data)

Two directories with matching files (different names).

Source code in gwframe/tests/test_compare.py

@pytest.fixture
def dir_pair(self, tmp_path, base_data):
    """Two directories with matching files (different names)."""
    left_dir = tmp_path / "dir1"
    right_dir = tmp_path / "dir2"
    left_dir.mkdir()
    right_dir.mkdir()
    for i in range(3):
        frames = [{"L1:DATA": base_data + i}]
        write_file(left_dir / f"left-{i}.gwf", frames, t0=T0 + i)
        write_file(right_dir / f"right-{i}.gwf", frames, t0=T0 + i)
    return left_dir, right_dir

shifted_pair

shifted_pair(tmp_path, base_data)

The same six 1 s frames, cut into files at different boundaries.

Source code in gwframe/tests/test_compare.py

@pytest.fixture
def shifted_pair(self, tmp_path, base_data):
    """The same six 1 s frames, cut into files at different boundaries."""
    left_dir = tmp_path / "left"
    right_dir = tmp_path / "right"
    left_dir.mkdir()
    right_dir.mkdir()
    frames = [{"L1:DATA": base_data + i} for i in range(6)]
    # left: [0,2) [2,4) [4,6)   right: [0,3) [3,6)
    for i in range(3):
        write_file(left_dir / f"l{i}.gwf", frames[2 * i : 2 * i + 2], t0=T0 + 2 * i)
    for i in range(2):
        write_file(
            right_dir / f"r{i}.gwf", frames[3 * i : 3 * i + 3], t0=T0 + 3 * i
        )
    return left_dir, right_dir

test_file_vanished_during_compare

test_file_vanished_during_compare(dir_pair, base_data, monkeypatch)

File deleted from retention between pairing and reading.

Source code in gwframe/tests/test_compare.py

def test_file_vanished_during_compare(self, dir_pair, base_data, monkeypatch):
    """File deleted from retention between pairing and reading."""
    import gwframe.compare as compare_mod

    left_dir, right_dir = dir_pair
    victim = right_dir / "right-1.gwf"
    real_reader = compare_mod.FrameReader

    def racy_reader(source, **kwargs):
        if Path(source) == victim:
            victim.unlink(missing_ok=True)
            msg = f"Unable to open file: {victim}"
            raise OSError(msg)
        return real_reader(source, **kwargs)

    monkeypatch.setattr(compare_mod, "FrameReader", racy_reader)

    result = compare_paths(left_dir, right_dir, common_time_spans=True)
    assert result.consistent
    assert len(result.file_results) == 2

    # Without the flag, the same race is reported as an error
    write_file(victim, [{"L1:DATA": base_data + 1}], t0=T0 + 1)
    result = compare_paths(left_dir, right_dir)
    assert not result.consistent
    fields = {d.field for r in result.file_results for d in r.differences}
    assert "open_failed" in fields

test_file_vanished_during_keying

test_file_vanished_during_keying(dir_pair, monkeypatch)

File deleted from retention between directory listing and keying.

Source code in gwframe/tests/test_compare.py

def test_file_vanished_during_keying(self, dir_pair, monkeypatch):
    """File deleted from retention between directory listing and keying."""
    import gwframe.compare as compare_mod

    left_dir, right_dir = dir_pair
    victim = right_dir / "right-1.gwf"
    real_get_info = compare_mod.get_info

    def racy_get_info(path):
        if path == victim:
            victim.unlink(missing_ok=True)
            raise FileNotFoundError(2, "No such file or directory", str(victim))
        return real_get_info(path)

    monkeypatch.setattr(compare_mod, "get_info", racy_get_info)

    result = compare_paths(left_dir, right_dir, common_time_spans=True)
    assert result.consistent
    assert len(result.file_results) == 2

test_kwargs_forwarded

test_kwargs_forwarded(dir_pair, base_data)

Tolerances propagate to per-file comparison in directory mode.

Source code in gwframe/tests/test_compare.py

def test_kwargs_forwarded(self, dir_pair, base_data):
    """Tolerances propagate to per-file comparison in directory mode."""
    left_dir, right_dir = dir_pair
    write_file(right_dir / "right-0.gwf", [{"L1:DATA": base_data + 1e-10}], t0=T0)

    assert not compare_paths(left_dir, right_dir).consistent
    assert compare_paths(left_dir, right_dir, atol=1e-6).consistent

test_pairing_is_content_based

test_pairing_is_content_based(dir_pair, base_data)

A data mismatch is attributed to the time-matched pair.

Source code in gwframe/tests/test_compare.py

def test_pairing_is_content_based(self, dir_pair, base_data):
    """A data mismatch is attributed to the time-matched pair."""
    left_dir, right_dir = dir_pair
    write_file(right_dir / "right-1.gwf", [{"L1:DATA": base_data * 7}], t0=T0 + 1)

    result = compare_paths(left_dir, right_dir)
    inconsistent = [r for r in result.file_results if not r.consistent]
    assert len(inconsistent) == 1
    left, right = inconsistent[0].left, inconsistent[0].right
    assert left is not None and left.name == "left-1.gwf"
    assert right is not None and right.name == "right-1.gwf"

TestCompareStructure

Frame structure comparison.

test_offset_frames_paired_by_span

test_offset_frames_paired_by_span(tmp_path, base_data)

Files sharing one span pair correctly even at different indices.

Source code in gwframe/tests/test_compare.py

def test_offset_frames_paired_by_span(self, tmp_path, base_data):
    """Files sharing one span pair correctly even at different indices."""
    frames = [{"L1:CHAN": base_data}] * 3
    left = write_file(tmp_path / "a.gwf", frames, t0=T0)
    # Right side starts one frame later: spans T0+1, T0+2, T0+3
    right = write_file(tmp_path / "b.gwf", frames, t0=T0 + DURATION)

    result = compare_files(left, right, common_time_spans=True)
    assert result.consistent
    assert result.frames_compared == 2

base_data

base_data()

Deterministic float64 test data.

Source code in gwframe/tests/test_compare.py

@pytest.fixture
def base_data():
    """Deterministic float64 test data."""
    return np.arange(16, dtype=np.float64)

write_file

write_file(path, frames, *, t0=T0, duration=DURATION)

Write a GWF file from a list of {channel: array-or-(array, kwargs)}.

Source code in gwframe/tests/test_compare.py

def write_file(path, frames, *, t0=T0, duration=DURATION):
    """Write a GWF file from a list of {channel: array-or-(array, kwargs)}."""
    with FrameWriter(str(path)) as writer:
        for i, channels in enumerate(frames):
            frame = Frame(
                start=t0 + i * duration,
                duration=duration,
                name="TEST",
                run=1,
                frame_number=i,
            )
            for name, spec in channels.items():
                data, kwargs = spec if isinstance(spec, tuple) else (spec, {})
                kwargs.setdefault("sample_rate", len(data) / duration)
                kwargs.setdefault("unit", "counts")
                frame.add_channel(name, data, **kwargs)
            writer.write_frame(frame)
    return path