Memory leak regression tests.
These tests verify that repeated read/write operations do not leak memory,
particularly around masked array handling, Frame lifecycle, and C++ object
prevent premature garbage collection patterns.
TestFileIOMemory
Verify file-based read/write does not leak.
test_frame_writer_multiple_frames
test_frame_writer_multiple_frames(tmp_path)
FrameWriter writing many frames should not leak.
Source code in gwframe/tests/test_memory.py
| def test_frame_writer_multiple_frames(self, tmp_path):
"""FrameWriter writing many frames should not leak."""
data = np.random.randn(N_SAMPLES)
path = str(tmp_path / "multi.gwf")
def cycle():
with gwframe.FrameWriter(path) as writer:
for i in range(10):
writer.write(
data,
start=T0 + i,
sample_rate=SAMPLE_RATE,
name="L1:TEST",
)
avg = _measure_memory_growth(cycle, iterations=20)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_read_frames_generator
test_read_frames_generator(tmp_path)
read_frames() generator should not leak.
Source code in gwframe/tests/test_memory.py
| def test_read_frames_generator(self, tmp_path):
"""read_frames() generator should not leak."""
data = np.random.randn(N_SAMPLES)
path = str(tmp_path / "test.gwf")
gwframe.write(path, data, start=T0, sample_rate=SAMPLE_RATE, name="L1:TEST")
def cycle():
for frame in gwframe.read_frames(path):
assert len(frame) == 1
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_write_read_file
test_write_read_file(tmp_path)
File write -> read cycle should not leak.
Source code in gwframe/tests/test_memory.py
| def test_write_read_file(self, tmp_path):
"""File write -> read cycle should not leak."""
data = np.random.randn(N_SAMPLES)
path = str(tmp_path / "test.gwf")
def cycle():
gwframe.write(path, data, start=T0, sample_rate=SAMPLE_RATE, name="L1:TEST")
ts = gwframe.read(path, "L1:TEST")
assert len(ts.array) == N_SAMPLES
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
TestFrameRebuildMemory
Verify Frame._rebuild_frame() does not leak.
test_channel_delete_rebuild
test_channel_delete_rebuild()
Deleting channels and writing should not leak.
Source code in gwframe/tests/test_memory.py
| def test_channel_delete_rebuild(self):
"""Deleting channels and writing should not leak."""
data = np.random.randn(N_SAMPLES)
def cycle():
frame = gwframe.Frame(start=T0, duration=DURATION, name="L1")
frame.add_channel("L1:KEEP", data, sample_rate=SAMPLE_RATE)
frame.add_channel("L1:DROP", data.copy(), sample_rate=SAMPLE_RATE)
del frame["L1:DROP"]
_ = frame.write_bytes()
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_channel_replace_rebuild
test_channel_replace_rebuild()
Replacing channel data via dict interface should not leak.
Source code in gwframe/tests/test_memory.py
| def test_channel_replace_rebuild(self):
"""Replacing channel data via dict interface should not leak."""
data = np.random.randn(N_SAMPLES)
def cycle():
frame = gwframe.Frame(start=T0, duration=DURATION, name="L1")
frame.add_channel("L1:TEST", data, sample_rate=SAMPLE_RATE)
# Replace with new TimeSeries
ts = frame["L1:TEST"]
new_data = np.zeros_like(data)
frame["L1:TEST"] = gwframe.TimeSeries(
array=new_data,
name=ts.name,
dtype=ts.dtype,
start=ts.start,
dt=ts.dt,
duration=ts.duration,
sample_rate=ts.sample_rate,
unit=ts.unit,
type=ts.type,
)
_ = frame.write_bytes()
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
TestMaskedArrayMemory
Verify no leaks in masked array paths.
test_masked_adc_roundtrip
test_masked_adc_roundtrip()
Masked ADC write/read cycle should not leak.
Source code in gwframe/tests/test_memory.py
| def test_masked_adc_roundtrip(self):
"""Masked ADC write/read cycle should not leak."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
masked = np.ma.MaskedArray(data, mask=mask)
def cycle():
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST", masked, sample_rate=SAMPLE_RATE, channel_type="adc"
)
gwf_bytes = frame.write_bytes()
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
assert ts.mask is not None
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_mixed_masked_plain_channels
test_mixed_masked_plain_channels()
Frame with both masked and plain channels should not leak.
Source code in gwframe/tests/test_memory.py
| def test_mixed_masked_plain_channels(self):
"""Frame with both masked and plain channels should not leak."""
adc_data = np.random.randn(N_SAMPLES).astype(np.float64)
proc_data = np.random.randn(N_SAMPLES).astype(np.float64)
masked_adc = np.ma.MaskedArray(adc_data, mask=np.ones(N_SAMPLES, dtype=bool))
def cycle():
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:ADC", masked_adc, sample_rate=SAMPLE_RATE, channel_type="adc"
)
frame.add_channel(
"H1:PROC", proc_data, sample_rate=SAMPLE_RATE, channel_type="proc"
)
gwf_bytes = frame.write_bytes()
result = gwframe.read_bytes(gwf_bytes, channels=None, allow_invalid=True)
assert result["H1:ADC"].mask is not None
assert result["H1:PROC"].mask is None
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_to_masked_does_not_leak
test_to_masked_does_not_leak()
Repeated to_masked() calls should not leak.
Source code in gwframe/tests/test_memory.py
| def test_to_masked_does_not_leak(self):
"""Repeated to_masked() calls should not leak."""
data = np.random.randn(N_SAMPLES).astype(np.float64)
mask = np.ones(N_SAMPLES, dtype=bool)
frame = gwframe.Frame(start=T0, duration=DURATION, name="H1")
frame.add_channel(
"H1:TEST",
np.ma.MaskedArray(data, mask=mask),
sample_rate=SAMPLE_RATE,
channel_type="adc",
)
gwf_bytes = frame.write_bytes()
ts = gwframe.read_bytes(gwf_bytes, "H1:TEST", allow_invalid=True)
def cycle():
ma = ts.to_masked()
assert isinstance(ma, np.ma.MaskedArray)
avg = _measure_memory_growth(cycle, iterations=500)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
TestPlainArrayMemory
Verify no leaks in standard write/read cycle.
test_frame_lifecycle
Creating and discarding Frame objects should not leak.
Source code in gwframe/tests/test_memory.py
| def test_frame_lifecycle(self):
"""Creating and discarding Frame objects should not leak."""
data = np.random.randn(N_SAMPLES)
def cycle():
frame = gwframe.Frame(start=T0, duration=DURATION, name="L1")
frame.add_channel("L1:CH1", data, sample_rate=SAMPLE_RATE)
frame.add_channel("L1:CH2", data.copy(), sample_rate=SAMPLE_RATE)
_ = frame.write_bytes()
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_multi_channel_read
test_multi_channel_read()
Multi-channel read should not leak.
Source code in gwframe/tests/test_memory.py
| def test_multi_channel_read(self):
"""Multi-channel read should not leak."""
data1 = np.random.randn(N_SAMPLES)
data2 = np.random.randn(N_SAMPLES).astype(np.float32)
frame = gwframe.Frame(start=T0, duration=DURATION, name="L1")
frame.add_channel("L1:CH1", data1, sample_rate=SAMPLE_RATE)
frame.add_channel("L1:CH2", data2, sample_rate=SAMPLE_RATE)
gwf_bytes = frame.write_bytes()
def cycle():
result = gwframe.read_bytes(gwf_bytes, channels=None)
assert len(result) == 2
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|
test_write_bytes_read_bytes
test_write_bytes_read_bytes()
write_bytes -> read_bytes round-trip should not leak.
Source code in gwframe/tests/test_memory.py
| def test_write_bytes_read_bytes(self):
"""write_bytes -> read_bytes round-trip should not leak."""
data = np.random.randn(N_SAMPLES)
def cycle():
gwf_bytes = gwframe.write_bytes(
data, start=T0, sample_rate=SAMPLE_RATE, name="L1:TEST"
)
ts = gwframe.read_bytes(gwf_bytes, "L1:TEST")
assert len(ts.array) == N_SAMPLES
avg = _measure_memory_growth(cycle)
assert avg < LEAK_THRESHOLD_BYTES_PER_ITER, (
f"Possible leak: {avg:.0f} bytes/iter"
)
|