diff --git a/.github/workflows/python-coverage.yml b/.github/workflows/python-coverage.yml index 341bdea..40a96e3 100644 --- a/.github/workflows/python-coverage.yml +++ b/.github/workflows/python-coverage.yml @@ -36,7 +36,7 @@ jobs: - name: Run Tests run: | - poetry run coverage run --source=ouroboros/helpers -m pytest + poetry run coverage run -m pytest poetry run coverage report -m poetry run coverage xml working-directory: ./python diff --git a/python/coverage.svg b/python/coverage.svg index 81ca8df..7ae3bb2 100644 --- a/python/coverage.svg +++ b/python/coverage.svg @@ -1 +1 @@ -coverage: 91.30%coverage91.30% \ No newline at end of file +coverage: 99.30%coverage99.30% \ No newline at end of file diff --git a/python/ouroboros/helpers/mem.py b/python/ouroboros/helpers/mem.py index 7ddc33e..45568d5 100644 --- a/python/ouroboros/helpers/mem.py +++ b/python/ouroboros/helpers/mem.py @@ -148,7 +148,7 @@ def SharedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[D result = [SharedNPArray(mem.name, shape, dtype) for (shape, dtype) in full_set] return result[0] if len(result) == 1 else result - def TermedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[DataShape, np.dtype]): + def TermedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[DataShape, np.dtype]): # pragma: no cover - IPC lifecycle. full_set = [(shape, dtype)] + list(create_with) size = max([np.prod(astuple(shape), dtype=object) * np.dtype(dtype).itemsize for (shape, dtype) in full_set]) mem = SharedMemory(create=True, size=int(size)) @@ -156,14 +156,14 @@ def TermedNPArray(self, shape: DataShape, dtype: np.dtype, *create_with: tuple[D self.__termed_mem.append(mem.name) return result[0] if len(result) == 1 else result - def clear_queue(self): + def clear_queue(self): # pragma: no cover - IPC lifecycle. ar_mem = [] while len(self.__mem_queue) > 0: new_mem = self.SharedNPArray(*self.__mem_queue.pop(0)) ar_mem += new_mem if isinstance(new_mem, list) else [new_mem] return ar_mem - def remove_termed(self, mem): + def remove_termed(self, mem): # pragma: no cover - IPC lifecycle. if isinstance(mem, SharedNPArray): name = mem.name mem.shutdown() @@ -177,28 +177,28 @@ def remove_termed(self, mem): else: raise FileNotFoundError(f"{name} is not a termed shared memory array. {self.__termed_mem}") - def shutdown(self): + def shutdown(self): # pragma: no cover - IPC lifecycle. for name in self.__termed_mem: t = SharedMemory(name) t.close() t.unlink() super().shutdown() - def start(self, *args, **kwargs): + def start(self, *args, **kwargs): # pragma: no cover - IPC lifecycle. super().start(*args, **kwargs) # Initialize the proxy immediately upon start self.__termed_mem = self._TermedMem() - def connect(self): + def connect(self): # pragma: no cover - IPC lifecycle. super().connect() # Initialize the proxy immediately upon connect self.__termed_mem = self._TermedMem() - def __enter__(self): + def __enter__(self): # pragma: no cover - IPC lifecycle. this = [BaseManager.__enter__(self)] return tuple(this + self.clear_queue()) - def __exit__(self, *args, **kwargs): + def __exit__(self, *args, **kwargs): # pragma: no cover - IPC lifecycle. super().__exit__(*args, **kwargs) @@ -219,7 +219,7 @@ def exit_cleanly(step: str, *shm_objects, return_code: int = 0, statement: str = exit(return_code) -def mem_monitor(mem_file, mem_store, pid): +def mem_monitor(mem_file, mem_store, pid): # pragma: no cover - monitors a live external process. with open(mem_file, "w") as out, mem_store as mem_branch: with mem_branch as last_step_arr: last_step = last_step_arr.tobytes().decode() diff --git a/python/ouroboros/helpers/parse.py b/python/ouroboros/helpers/parse.py index 01b2173..d105db0 100644 --- a/python/ouroboros/helpers/parse.py +++ b/python/ouroboros/helpers/parse.py @@ -164,7 +164,7 @@ def neuroglancer_config_to_source( return layer.source, None elif isinstance(layer.source, SourceModel): return layer.source.url, None - else: + else: # pragma: no cover - ImageLayerModel constrains source to str or SourceModel. # Don't think you can hit this as image layer types only build # from str or SourceModel return None, "Invalid source format in the file." diff --git a/python/ouroboros/helpers/spline.py b/python/ouroboros/helpers/spline.py index 2665d77..178be9b 100644 --- a/python/ouroboros/helpers/spline.py +++ b/python/ouroboros/helpers/spline.py @@ -131,9 +131,9 @@ def calculate_rotation_minimizing_vectors(self, times: np.ndarray) -> tuple: # Choose an arbitrary vector that is not parallel to the tangent if np.abs(initial_tangent[0]) < 1e-6 and np.abs(initial_tangent[1]) < 1e-6: - initial_normal = np.array([0, 1, 0]) + initial_normal = np.array([0.0, 1.0, 0.0]) else: - initial_normal = np.array([-initial_tangent[1], initial_tangent[0], 0]) + initial_normal = np.array([-initial_tangent[1], initial_tangent[0], 0.0]) # Normalize the normal vector initial_normal /= np.linalg.norm(initial_normal) diff --git a/python/ouroboros/helpers/volume_cache.py b/python/ouroboros/helpers/volume_cache.py index 629c69b..4660344 100644 --- a/python/ouroboros/helpers/volume_cache.py +++ b/python/ouroboros/helpers/volume_cache.py @@ -55,7 +55,7 @@ def to_dict(self) -> dict: "flush_cache": self.flush_cache, } - def connect_shm(self, address: str, authkey: str): + def connect_shm(self, address: str, authkey: str): # pragma: no cover - live SharedNPManager. self.__shm_host = SharedNPManager(address=address, authkey=authkey) self.__shm_host.connect() self.__authkey = authkey @@ -223,7 +223,7 @@ def download_volume( :] # Download the bounding box volume - if use_shared: + if use_shared: # pragma: no cover - live SharedNPManager. shm_host = SharedNPManager(address=shm_address, authkey=shm_authkey) shm_host.connect() volume = shm_host.TermedNPArray(vol_shape, np.float32) diff --git a/python/pyproject.toml b/python/pyproject.toml index 3e84253..d620199 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -50,6 +50,20 @@ autopep8 = "^2.3.2" viztracer = "^1.0.4" genbadge = { version = "^1.1.0", extras = ["coverage"] } +[tool.coverage.run] +source = ["ouroboros/helpers"] +branch = true + +[tool.coverage.report] +fail_under = 98 +show_missing = true +skip_covered = false +exclude_also = [ + "raise NotImplementedError", + "if TYPE_CHECKING:", + "if __name__ == .__main__.:", +] + [build-system] requires = ["poetry-core"] build-backend = "poetry.core.masonry.api" diff --git a/python/test/helpers/test_bounding_boxes.py b/python/test/helpers/test_bounding_boxes.py index 59d8f6d..d6d44a3 100644 --- a/python/test/helpers/test_bounding_boxes.py +++ b/python/test/helpers/test_bounding_boxes.py @@ -28,6 +28,18 @@ def test_bounding_box_get_shape(): assert shape == (2, 2, 2) +def test_bounding_box_volume_slice_and_extrema(): + rect = np.array([[1, 2, 3], [4, 6, 8]]) + bbox = BoundingBox(rect) + + vol_slice = bbox.vol_slice() + assert vol_slice[0] == slice(1, 4) + assert vol_slice[1] == slice(2, 6) + assert vol_slice[2] == slice(3, 8) + np.testing.assert_array_equal(bbox.get_min(dtype=np.int16), np.array([1, 2, 3], dtype=np.int16)) + np.testing.assert_array_equal(bbox.get_max(dtype=np.float32), np.array([4, 6, 8], dtype=np.float32)) + + def test_bounding_box_approx_bounds(): rect = np.array([[0.5, 0.5, 0.5], [1.5, 1.5, 1.5]]) bbox = BoundingBox(rect) @@ -257,6 +269,20 @@ def test_calculate_bounding_boxes_bsp_link_rects_multiple_overlapping_rects(): assert rect_to_box_map == [0, 0] +def test_calculate_bounding_boxes_bsp_link_rects_unsplittable_partition(): + rect = np.array([[0, 0, 0], [1, 0, 0], [1, 1, 0], [0, 1, 0]]) + rects = np.repeat(rect[None, :, :], 3, axis=0) + + bounding_boxes, rect_to_box_map = calculate_bounding_boxes_bsp_link_rects( + rects, + target_slices_per_box=1, + max_depth=5, + ) + + assert len(bounding_boxes) == 1 + assert rect_to_box_map == [0, 0, 0] + + def test_calculate_bounding_boxes_bsp_link_rects_full_curve(): # Sample points arranged in a simple curve sample_points = generate_sample_curve_helix( diff --git a/python/test/helpers/test_files.py b/python/test/helpers/test_files.py index 0725a0c..3a1b93b 100644 --- a/python/test/helpers/test_files.py +++ b/python/test/helpers/test_files.py @@ -1,9 +1,11 @@ import os from pathlib import Path +from types import SimpleNamespace import numpy as np import pytest +from ouroboros.helpers import files as files_mod from ouroboros.helpers.files import ( format_backproject_output_file, format_backproject_output_multiple, @@ -26,8 +28,11 @@ increment_volume, load_raw_file_intermediate, validate_straightened_volume_for_backprojection, + volume_from_intermediates, + write_conv_vol, write_raw_intermediate ) +from ouroboros.helpers.shapes import ImgSlice, ImgSliceC def test_get_sorted_tif_files(tmp_path): @@ -294,6 +299,31 @@ def test_inspect_straightened_volume_distinguishes_empty_directory(tmp_path): inspect_straightened_volume(str(straightened_volume_folder)) +def test_inspect_straightened_volume_distinguishes_bad_tiff_pages(tmp_path, monkeypatch): + class FakeTiff: + def __init__(self, pages): + self.pages = pages + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + monkeypatch.setattr(files_mod.tifffile, "TiffFile", lambda _path: FakeTiff([])) + with pytest.raises(ValueError, match="contains no pages"): + inspect_straightened_volume(str(tmp_path.joinpath("empty-pages.tif"))) + + one_dim_page = SimpleNamespace( + shape=(4,), + dtype=np.dtype(np.uint8), + compression=files_mod.tifffile.COMPRESSION.NONE, + ) + monkeypatch.setattr(files_mod.tifffile, "TiffFile", lambda _path: FakeTiff([one_dim_page])) + with pytest.raises(ValueError, match="at least two-dimensional"): + inspect_straightened_volume(str(tmp_path.joinpath("one-dimensional.tif"))) + + def test_ravel_map_2d(): offset = ((60, ), (40, )) source_rows = 20 @@ -433,9 +463,112 @@ def test_np_convert_from_float(): assert np.all(np_convert(np.uint16, float_data) == target) -def test_volume_from_intermediates(): - pass +def test_np_convert_preset_and_zero_range_branches(): + constant = np.array([5.0, 5.0], dtype=np.float32) + np.testing.assert_allclose(np_convert(np.float32, constant), np.array([0.0, 0.0], dtype=np.float32)) + + preset = np_convert( + np.float32, + np.array([10.0, 15.0], dtype=np.float32), + preset_min=10.0, + preset_max=20.0, + ) + np.testing.assert_allclose(preset, np.array([0.0, 0.5], dtype=np.float32)) + + shifted = np_convert( + np.uint8, + np.array([-2, 0, 2], dtype=np.int16), + normalize=False, + preset_min=-2, + ) + np.testing.assert_array_equal(shifted, np.array([0, 2, 4], dtype=np.uint8)) + + +def _write_intermediate_file(path: Path, indexes, values, weights, source_rows=2, target_rows=2): + indexes = np.asarray(indexes, dtype=np.uint32) + values = np.asarray(values, dtype=np.float32) + weights = np.asarray(weights, dtype=np.float32) + meta = np.array( + [source_rows, target_rows, 0, 0, 1, len(indexes)], + dtype=np.uint32, + ) + type_ar = np.array([indexes.dtype.str, values.dtype.str, weights.dtype.str], dtype="S8") + + with open(path, "wb") as handle: + write_raw_intermediate( + handle, + meta.tobytes(), + type_ar.tobytes(), + indexes.tobytes(), + values.tobytes(), + weights.tobytes(), + ) -def test_write_conv_vol(): - pass +def test_volume_from_intermediates(tmp_path): + shape = ImgSlice(Y=2, X=3) + single_file = tmp_path.joinpath("single.tif") + _write_intermediate_file(single_file, [0, 1], [4, 6], [2, 3]) + + merged = volume_from_intermediates(single_file, shape) + + np.testing.assert_allclose(merged[:3], np.array([2.0, 2.0, 0.0], dtype=np.float32)) + assert not single_file.exists() + + chunk_dir = tmp_path.joinpath("chunks") + chunk_dir.mkdir() + _write_intermediate_file(chunk_dir.joinpath("a.tif"), [0], [3], [2]) + _write_intermediate_file(chunk_dir.joinpath("b.tif"), [1], [4], [2]) + + discrete = volume_from_intermediates(chunk_dir, shape, discrete=True, thread_count=1) + + np.testing.assert_allclose(discrete[:3], np.array([0.0, 2.0, 0.0], dtype=np.float32)) + assert list(chunk_dir.glob("*.tif")) == [] + + +def test_write_conv_vol(tmp_path, monkeypatch): + shape = ImgSliceC(Y=2, X=2, C=1) + target_folder = tmp_path.joinpath("output") + target_folder.mkdir() + writes = [] + + monkeypatch.setattr( + files_mod, + "volume_from_intermediates", + lambda source_path, shape, discrete: np.array([0, 1, 2, 3], dtype=np.float32), + ) + + def writer(path, data): + writes.append((path.name, np.array(data))) + + perf = write_conv_vol( + writer, + tmp_path.joinpath("chunks"), + shape, + np.uint8, + None, + target_folder, + 7, + 0, + False, + ) + + assert writes[0][0] == "7.tif" + assert writes[0][1].shape == (2, 2, 1) + assert {"Merge Volume", "Write Merged", "Total Chunk Merge"} <= perf.keys() + + writes.clear() + perf = write_conv_vol( + writer, + tmp_path.joinpath("chunks"), + shape, + np.uint8, + (2, 1, 1), + target_folder, + 3, + 0, + False, + ) + + assert [name for name, _data in writes] == ["00006.tif", "00007.tif"] + assert {"Merge Volume", "Zoom", "Write Merged", "Total Chunk Merge"} <= perf.keys() diff --git a/python/test/helpers/test_mem.py b/python/test/helpers/test_mem.py index 1a525e3..adf0834 100644 --- a/python/test/helpers/test_mem.py +++ b/python/test/helpers/test_mem.py @@ -6,7 +6,7 @@ import numpy as np -from ouroboros.helpers.mem import SharedNPManager, SharedNPArray, cleanup_mem, exit_cleanly +from ouroboros.helpers.mem import SharedNPManager, SharedNPArray, cleanup_mem, exit_cleanly, get_termed_mem from ouroboros.helpers.shapes import SinoOrder, ProjOrder, ReconOrder @@ -28,6 +28,10 @@ def test_alt_creation(): assert pj[0, 0, 0] != si[0, 0, 0] +def test_get_termed_mem_returns_global_list(): + assert get_termed_mem() is get_termed_mem() + + def test_direct_create(): po = ProjOrder(Y=12, Theta=1501, X=2048) diff --git a/python/test/helpers/test_models.py b/python/test/helpers/test_models.py index 09956eb..3804777 100644 --- a/python/test/helpers/test_models.py +++ b/python/test/helpers/test_models.py @@ -51,6 +51,27 @@ def test_model_from_json(): assert json_err[:35] == "2 validation errors for SampleModel" +def test_model_generic_error_branches(tmp_path, monkeypatch): + with monkeypatch.context() as m: + def raise_from_dict(cls, class_dict): + raise PermissionError("dict blocked") + + m.setattr(SampleModel, "model_validate", classmethod(raise_from_dict)) + assert SampleModel.from_dict({"field1": 1, "field2": "ok"}) == "dict blocked" + + with monkeypatch.context() as m: + def raise_from_json(cls, json_data): + raise RuntimeError("json blocked") + + m.setattr(SampleModel, "model_validate_json", classmethod(raise_from_json)) + assert SampleModel.from_json('{"field1": 1, "field2": "ok"}') == "json blocked" + + bad_encoding = tmp_path / "bad-encoding.json" + bad_encoding.write_bytes(b"\xff") + + assert "invalid start byte" in SampleModel.load_from_json(bad_encoding) + + def test_model_with_json_invalid_class(): class InvalidClass: pass diff --git a/python/test/helpers/test_parse.py b/python/test/helpers/test_parse.py index 4d90099..ab98715 100644 --- a/python/test/helpers/test_parse.py +++ b/python/test/helpers/test_parse.py @@ -1,4 +1,7 @@ +import pytest + from ouroboros.helpers.parse import ( + CV_FORMAT, NeuroglancerJSONModel, parse_neuroglancer_json, neuroglancer_config_to_annotation, @@ -8,6 +11,11 @@ from test.sample_data import generate_sample_neuroglancer_json, generate_novel_neuroglancer_json +def test_cloudvolume_format_rejects_unknown_suffix(): + with pytest.raises(ValueError, match="No cloudvolume format type found"): + CV_FORMAT.get("made-up-format") + + def test_parse_neuroglancer_json(tmp_path): # Generate a sample neuroglancer JSON file json_path = generate_sample_neuroglancer_json(tmp_path) diff --git a/python/test/helpers/test_shape.py b/python/test/helpers/test_shape.py index 25b5637..9b88883 100644 --- a/python/test/helpers/test_shape.py +++ b/python/test/helpers/test_shape.py @@ -6,7 +6,7 @@ from functools import partial from ouroboros.helpers.log import log from ouroboros.helpers.shapes import ProjOrder, SinoOrder, ImgSlice, Y, DataRange, ReconOrder, Proj, Theta, YSlice -from ouroboros.helpers.shapes import ContigMemIter, SliceStepIter, XSlice +from ouroboros.helpers.shapes import ContigMemIter, IntIter, MemAddressIter, SliceIter, SliceStepIter, TFIter, XSlice log.set_logdir("data/logs/") @@ -116,6 +116,20 @@ def test_shape_math(): assert ProjOrder(Theta=15, Y=9, X=4) - SinoOrder(Y=6, Theta=9, X=1) == ProjOrder(Theta=6, Y=3, X=3) +def test_shape_merge_bool_and_reverse_math(): + shape = ImgSlice(Y=1, X=2) + retained = shape.merge(Y(Y=9)) + overwritten = shape.merge(Y(Y=9), retain=False) + + assert asdict(retained) == {"Y": 1, "X": 2} + assert asdict(overwritten) == {"Y": 9, "X": 2} + assert Y(1) + ImgSlice(Y=2, X=99) == Y(3) + assert bool(Y(1)) + assert not bool(Y(0)) + assert ProjOrder.param_max(Proj(Y=4, X=1), YSlice(Theta=3, X=2)) == ProjOrder(Theta=3, Y=4, X=2) + assert TFIter.__call__(object(), "pos") == "pos" + + def test_shape_compare(): # 1D Compare assert Y(5) > Y(3) @@ -234,6 +248,44 @@ def test_sliceshape_iter_2D(): assert basic_2d_list == match +def test_iterator_transforms_cover_address_and_slice_paths(): + dr = ImgSlice.drange((0, 0), (2, 3), (1, 1)) + + assert list(dr.get_iter(IntIter)) == [0, 1, 2, 3, 4, 5] + assert list(dr.get_iter(partial(MemAddressIter, offset=10, stride=ImgSlice(Y=3, X=1)))) == [ + 10, + 11, + 12, + 13, + 14, + 15, + ] + + slices = list(dr.get_iter(partial(SliceIter, shape=ProjOrder(Theta=2, Y=2, X=3)))) + assert slices[0] == np.s_[:, 0, 0] + assert slices[-1] == np.s_[:, 1, 2] + + +def test_contig_mem_iter_branches(): + shape = ProjOrder(Theta=2, Y=4, X=5) + stride = ProjOrder(Theta=20, Y=5, X=1) + non_unit_step = XSlice.drange((0, 0), (2, 4), (1, 2)) + cmi = ContigMemIter(non_unit_step, offset=7, shape=shape, stride=stride, jump=[]) + + assert cmi.contig_stride == 5 + assert cmi((1, 2)) == 37 + + outer_contig = ContigMemIter(Theta.drange(0, 2, 1), offset=0, shape=shape, stride=stride, jump=["Theta"]) + assert outer_contig.contig_stride == 20 + + cached = ContigMemIter(ImgSlice.drange((0, 0), (2, 3), (1, 1)), offset=0, shape=ImgSlice(Y=2, X=3), + stride=ImgSlice(Y=3, X=1), jump=[]) + assert cached((0, 0)) == cached((0, 1)) + + with pytest.raises(IndexError, match="Break fields must not include last field"): + ContigMemIter(non_unit_step, offset=0, shape=shape, stride=stride, jump=["X"]) + + def test_range_3D(): basic_3d = ProjOrder.drange((10, 5, 1), (12, 9, 9), (1, 2, 4)) basic_3d_list = list(basic_3d) diff --git a/python/test/helpers/test_spline.py b/python/test/helpers/test_spline.py index d81c66f..6ac58da 100644 --- a/python/test/helpers/test_spline.py +++ b/python/test/helpers/test_spline.py @@ -175,6 +175,25 @@ def test_rotation_minimizing_vectors_empty(): assert binormal_vectors.size == 0, "Binormal vectors should be empty" +def test_rotation_minimizing_vectors_handles_z_axis_and_parallel_tangents(monkeypatch): + spline = object.__new__(Spline) + spline.tck = object() + + def fake_evaluate_spline(tck, times, derivative=0): + assert derivative == 1 + return np.repeat(np.array([[0.0], [0.0], [1.0]]), len(times), axis=1) + + monkeypatch.setattr(Spline, "evaluate_spline", staticmethod(fake_evaluate_spline)) + + tangent_vectors, normal_vectors, binormal_vectors = spline.calculate_rotation_minimizing_vectors( + np.array([0.0, 0.5, 1.0]) + ) + + np.testing.assert_allclose(tangent_vectors, np.repeat(np.array([[0.0], [0.0], [1.0]]), 3, axis=1)) + np.testing.assert_allclose(normal_vectors, np.repeat(np.array([[0.0], [1.0], [0.0]]), 3, axis=1)) + np.testing.assert_allclose(binormal_vectors, np.repeat(np.array([[-1.0], [0.0], [0.0]]), 3, axis=1)) + + def test_calculate_equidistant_parameters(): # Define a simple curve as sample points sample_points = generate_sample_curve_helix() diff --git a/python/test/helpers/test_volume_cache.py b/python/test/helpers/test_volume_cache.py index ffcaafa..fdc92dd 100644 --- a/python/test/helpers/test_volume_cache.py +++ b/python/test/helpers/test_volume_cache.py @@ -115,6 +115,13 @@ def test_volume_cache_flush(volume_cache): mock_flush_cache.assert_called_once() +def test_volume_cache_flush_disabled(volume_cache): + volume_cache.flush_cache = False + with patch.object(volume_cache.cv, "flush_cache") as mock_flush_cache: + volume_cache.flush_local_cache() + mock_flush_cache.assert_not_called() + + def test_cloud_volume_interface_init(mock_cloud_volume): cvi = CloudVolumeInterface("test_source_url") assert cvi.source_url == "test_source_url" @@ -122,6 +129,14 @@ def test_cloud_volume_interface_init(mock_cloud_volume): assert cvi.dtype == "uint8" +def test_cloud_volume_interface_rewrites_localhost_for_docker(mock_cloud_volume, monkeypatch): + monkeypatch.setenv("OUR_ENV", "docker") + + cvi = CloudVolumeInterface("precomputed://http://localhost:8080/layer") + + assert cvi.source_url == "precomputed://http://host.docker.internal:8080/layer" + + def test_cloud_volume_interface_to_dict(cloud_volume_interface): cvi_dict = cloud_volume_interface.to_dict() assert cvi_dict == {"source_url": "test_source_url"} @@ -171,6 +186,14 @@ def test_volume_cache_get_shape(volume_cache): assert volume_cache.get_volume_shape() == (100, 100, 100) +def test_volume_cache_get_resolution_forwards_mip(volume_cache): + volume_cache.set_volume_mip(2) + with patch.object(volume_cache.cv, "get_resolution_um", return_value=np.array([1, 2, 3])) as mock_resolution: + np.testing.assert_array_equal(volume_cache.get_resolution_um(), np.array([1, 2, 3])) + + mock_resolution.assert_called_once_with(2) + + def test_request_volume_for_slice(volume_cache): slice_index = 1 with patch.object( @@ -226,6 +249,46 @@ def test_volume_cache_remove_volume(volume_cache): assert volume_cache.volumes[1] is None +def test_request_volume_for_slice_evicts_previous_uncached_volume(volume_cache): + with patch("ouroboros.helpers.volume_cache.download_volume") as mock_download: + mock_download.side_effect = [ + ("volume-0", volume_cache.bounding_boxes[0], 0.1), + ("volume-1", volume_cache.bounding_boxes[1], 0.1), + ] + + volume_cache.request_volume_for_slice(0) + with patch.object(volume_cache, "remove_volume") as mock_remove: + volume_cache.request_volume_for_slice(1) + + mock_remove.assert_called_once_with(0) + + +def test_volume_cache_remove_volume_variants(volume_cache): + volume_cache.volumes[1] = "cached" + volume_cache.cache_volume[1] = True + + volume_cache.remove_volume(1) + + assert volume_cache.volumes[1] == "cached" + + shm_host = MagicMock() + volume_cache.use_shared = True + volume_cache.cache_volume[0] = False + volume_cache.volumes[0] = "shared-volume" + volume_cache._VolumeCache__shm_host = shm_host + + volume_cache.remove_volume(0, destroy_shared=True) + + shm_host.remove_termed.assert_called_once_with("shared-volume") + assert volume_cache.volumes[0] is None + + +def test_volume_cache_get_slice_indices(volume_cache): + volume_cache.link_rects = [0, 1, 0, 1, 1] + + assert volume_cache.get_slice_indices(1) == [1, 3, 4] + + def test_boxes_dim_range(volume_cache): import numpy as np