diff --git a/dascore/io/core.py b/dascore/io/core.py index 57393c1f3..0f94a880a 100644 --- a/dascore/io/core.py +++ b/dascore/io/core.py @@ -720,7 +720,7 @@ def _get_format( # raise, in which case the format doesn't belong. func_input = man.get_resource(required_type) format_version = func(func_input, _pre_cast=True) - except RemoteCacheError: # pragma: no cover -- remote fetch only + except RemoteCacheError: # A remote fetch failure is a real error, not a "wrong # format" signal, so it must propagate rather than be # swallowed by the robustness handler below. diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index 57952f975..ea4acdf01 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -425,12 +425,12 @@ def _build_members(sub: pd.DataFrame, outputs: pd.DataFrame, name) -> pd.DataFra continue lo = max(src1[src_num], chu1[out_num]) hi = min(src2[src_num], chu2[out_num]) - if lo > hi: # pragma: no cover -- searchsorted boundary guard - # Sources within a partition are continuous (partitioning - # splits on gaps) and start-corrected, so searchsorted does - # not offer a non-overlapping source in practice; this guards - # against a boundary off-by-one rather than a reachable state. - continue + # Sources within a partition are continuous (partitioning + # splits on gaps) and start-corrected, so searchsorted never + # offers a source which does not overlap the output. Assert it + # rather than skipping: silently dropping a source would lose + # data, and the state cannot be reached from the public API. + assert lo <= hi, f"source {src_num} does not overlap output {out_num}" unchanged = ( lo == orig_min[src_num] and hi == orig_max[src_num] diff --git a/dascore/utils/hdf5.py b/dascore/utils/hdf5.py index ea8d322fd..84602cc73 100644 --- a/dascore/utils/hdf5.py +++ b/dascore/utils/hdf5.py @@ -240,7 +240,7 @@ class H5Writer(H5Reader): class _RemoteH5Writer: """Wrap a local h5py file and upload it back to the remote resource.""" - def __init__(self, resource: UPath, mode: str): # pragma: no cover + def __init__(self, resource: UPath, mode: str): self._resource = resource suffix = resource.suffix or ".h5" fd, temp_name = tempfile.mkstemp(suffix=suffix) @@ -264,10 +264,10 @@ def __init__(self, resource: UPath, mode: str): # pragma: no cover def __getitem__(self, item): return self._handle[item] - def __setitem__(self, key, value): # pragma: no cover + def __setitem__(self, key, value): self._handle[key] = value - def __contains__(self, item): # pragma: no cover + def __contains__(self, item): return item in self._handle def commit(self): @@ -299,10 +299,10 @@ def _abort(self): """Backward-compatible alias for abort().""" self.abort() - def __enter__(self): # pragma: no cover + def __enter__(self): return self - def __exit__(self, exc_type, exc, tb): # pragma: no cover + def __exit__(self, exc_type, exc, tb): if exc_type is None: self.commit() else: diff --git a/tests/test_io/test_io_core.py b/tests/test_io/test_io_core.py index 15e441c75..a4fb1a35f 100644 --- a/tests/test_io/test_io_core.py +++ b/tests/test_io/test_io_core.py @@ -514,6 +514,27 @@ def test_other_formats_still_usable(self, broken_ep_manager): assert len(list(broken_ep_manager.yield_fiberio())) +class TestGetFormatErrors: + """Errors which must not be mistaken for a format mismatch.""" + + def test_remote_cache_error_propagates(self, monkeypatch, tmp_path): + """A remote fetch failure is a real error, not a wrong-format signal. + + The loop over FiberIOs swallows exceptions so a reader which does + not recognize a file can be skipped; a cache failure has to escape + that handler instead of being reported as an unknown format. + """ + path = tmp_path / "unfetchable.h5" + path.write_bytes(b"not really an h5 file") + + def _raise(*args, **kwargs): + raise RemoteCacheError("cannot fetch this resource") + + monkeypatch.setattr(IOResourceManager, "get_resource", _raise) + with pytest.raises(RemoteCacheError, match="cannot fetch this resource"): + dc.get_format(path) + + class TestFormatManagerConcurrency: """Concurrent plugin loading must never expose a partial registry.""" diff --git a/tests/test_utils/test_io_utils.py b/tests/test_utils/test_io_utils.py index c241c9061..399b379a3 100644 --- a/tests/test_utils/test_io_utils.py +++ b/tests/test_utils/test_io_utils.py @@ -11,6 +11,7 @@ from upath import UPath import dascore as dc +import dascore.utils.hdf5 as hdf5_module import dascore.utils.remote_io as remote_io from dascore.config import config_context from dascore.exceptions import PatchConversionError, RemoteCacheError @@ -427,6 +428,54 @@ def test_h5_writer_to_remote_upath_aborts_on_context_error(self): raise RuntimeError("boom") assert not path.exists() + def test_h5_writer_remote_context_commits(self): + """Leaving the context without an error uploads the file.""" + path = UPath("memory://dascore/upath-write-commit.h5") + with H5Writer.get_handle(path) as handle: + handle.create_dataset("data", data=[1, 2, 3]) + assert path.exists() + with path.open("rb") as raw, h5py.File(raw, "r", driver="fileobj") as reopened: + assert list(reopened["data"][:]) == [1, 2, 3] + + def test_h5_writer_remote_append_keeps_existing(self): + """Reopening an existing remote file downloads it before writing.""" + path = UPath("memory://dascore/upath-write-append.h5") + with H5Writer.get_handle(path) as handle: + handle.create_dataset("first", data=[1, 2, 3]) + with H5Writer.get_handle(path) as handle: + handle.create_dataset("second", data=[4, 5, 6]) + with path.open("rb") as raw, h5py.File(raw, "r", driver="fileobj") as reopened: + assert list(reopened["first"][:]) == [1, 2, 3] + assert list(reopened["second"][:]) == [4, 5, 6] + + def test_h5_writer_remote_setitem_and_contains(self): + """The remote writer proxies item assignment and membership.""" + path = UPath("memory://dascore/upath-write-setitem.h5") + with H5Writer.get_handle(path) as handle: + handle["data"] = [1, 2, 3] + assert "data" in handle + assert "missing" not in handle + + def test_h5_writer_remote_cleans_up_after_open_failure(self, monkeypatch): + """A failed local open removes the temp file and raises.""" + created = [] + real_mkstemp = hdf5_module.tempfile.mkstemp + + def _tracking_mkstemp(*args, **kwargs): + file_descriptor, name = real_mkstemp(*args, **kwargs) + created.append(Path(name)) + return file_descriptor, name + + def _raise(*args, **kwargs): + raise RuntimeError("boom") + + monkeypatch.setattr(hdf5_module.tempfile, "mkstemp", _tracking_mkstemp) + monkeypatch.setattr(hdf5_module, "H5pyFile", _raise) + path = UPath("memory://dascore/upath-write-open-failure.h5") + with pytest.raises(RuntimeError, match="boom"): + H5Writer.get_handle(path) + assert created and not created[0].exists() + def test_h5_writer_remote_abort_is_idempotent(self): """Remote writer aborts should be safe to call more than once.""" path = UPath("memory://dascore/upath-write-abort-twice.h5")