From d99c6dfe07cd79633d43a8f503d61947b477fe08 Mon Sep 17 00:00:00 2001 From: Callum Forrester Date: Tue, 15 Aug 2023 12:15:48 +0100 Subject: [PATCH] Write block of tests that check behavoir of read paths, write paths and root paths in file store mixin class --- ophyd/tests/test_areadetector.py | 212 ++++++++++++++++++++++++++++++- 1 file changed, 211 insertions(+), 1 deletion(-) diff --git a/ophyd/tests/test_areadetector.py b/ophyd/tests/test_areadetector.py index 6816c46e6..3903eaa68 100644 --- a/ophyd/tests/test_areadetector.py +++ b/ophyd/tests/test_areadetector.py @@ -4,9 +4,11 @@ import shutil import time import uuid +from dataclasses import dataclass from io import StringIO from pathlib import Path, PurePath -from unittest.mock import Mock +from typing import Optional +from unittest.mock import Mock, patch import numpy as np import pytest @@ -24,6 +26,7 @@ from ophyd.areadetector.filestore_mixins import ( FileStoreHDF5, FileStoreIterativeWrite, + FileStorePluginBase, FileStoreTIFF, ) @@ -582,6 +585,213 @@ class MyDetector(SingleTrigger, SimDetector): assert Path(fn).exists() +@dataclass +class FileStorePathTestCase: + root: Optional[str] + write_path_template: str + read_path_template: Optional[str] + expected_write_path: str + expected_resource_root: str + expected_resource_path_directory: str + + +class DummyFileStorePlugin(Device, FileStorePluginBase): + file_write_mode: Signal = Component(Signal, value="test") + file_path: Signal = Component(Signal, value="") + file_name: Signal = Component(Signal, value="") + file_number: Signal = Component(Signal, value=0) + file_path_exists: Signal = Component(Signal, value=1) + file_template: Signal = Component(Signal, value="") + auto_increment: Signal = Component(Signal, value=0) + auto_save: Signal = Component(Signal, value=0) + num_capture: Signal = Component(Signal, value=0) + array_counter: Signal = Component(Signal, value=0) + capture: Signal = Component(Signal, value=0) + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.filestore_spec = "FS_TEST" # spec name stored in resource doc + self.stage_sigs.update( + [ + ("file_template", "%s%s_%6.6d.h5"), + ("file_write_mode", "Stream"), + ("capture", 1), + ] + ) + + def stage(self): + super().stage() + self._generate_resource({}) + + +@pytest.fixture +def mock_now() -> Mock: + with patch("ophyd.areadetector.filestore_mixins.datetime") as mock: + mock.now.return_value = datetime.datetime(year=2017, month=1, day=2) + yield mock + + +FILE_STORE_TEST_CASES = [ + FileStorePathTestCase( + root=None, + write_path_template="/tmp/ophyd_file_store_test/data1/%Y/%m/%d", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/2017/01/02", + expected_resource_root="/", + expected_resource_path_directory="tmp/ophyd_file_store_test/data1/2017/01/02", + ), + FileStorePathTestCase( + root="/", + write_path_template="/tmp/ophyd_file_store_test/data1/%Y/%m/%d", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/2017/01/02", + expected_resource_root="/", + expected_resource_path_directory="tmp/ophyd_file_store_test/data1/2017/01/02", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test/", + write_path_template="data1/%Y/%m/%d", + read_path_template="/tmp/ophyd_file_store_test/data1/%Y/%m/%d", + expected_write_path="data1/2017/01/02", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory="data1/2017/01/02", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test/", + write_path_template="data1/%Y/%m/%d", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/2017/01/02", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory="data1/2017/01/02", + ), + FileStorePathTestCase( + root=None, + write_path_template="/tmp/ophyd_file_store_test/data1/test_dataset", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/test_dataset", + expected_resource_root="/", + expected_resource_path_directory="tmp/ophyd_file_store_test/data1/test_dataset", + ), + FileStorePathTestCase( + root="/", + write_path_template="/tmp/ophyd_file_store_test/data1/test_dataset", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/test_dataset", + expected_resource_root="/", + expected_resource_path_directory="tmp/ophyd_file_store_test/data1/test_dataset", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test/", + write_path_template="data1/test_dataset", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/test_dataset", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory="data1/test_dataset", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test", + write_path_template=".", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory=".", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test", + write_path_template=".", + read_path_template=".", + expected_write_path=".", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory=".", + ), + FileStorePathTestCase( + root="/tmp/ophyd_file_store_test", + write_path_template="data1/test_dataset", + read_path_template=None, + expected_write_path="/tmp/ophyd_file_store_test/data1/test_dataset", + expected_resource_root="/tmp/ophyd_file_store_test", + expected_resource_path_directory="data1/test_dataset", + ), +] + + +@pytest.mark.parametrize("test_case", FILE_STORE_TEST_CASES) +def test_file_store_paths_with_constructor( + mock_now: Mock, test_case: FileStorePathTestCase +) -> None: + fs = DummyFS() + + file_store = DummyFileStorePlugin( + name="test_file_store", + write_path_template=test_case.write_path_template, + read_path_template=test_case.read_path_template, + root=test_case.root, + reg=fs, + ) + check_file_store_paths(file_store, test_case) + + +@pytest.mark.parametrize("test_case", FILE_STORE_TEST_CASES) +def test_file_store_paths_with_setters( + mock_now: Mock, test_case: FileStorePathTestCase +) -> None: + fs = DummyFS() + + file_store = DummyFileStorePlugin( + name="test_file_store", + write_path_template="", + read_path_template="", + root=None, + reg=fs, + ) + file_store.reg_root = test_case.root + file_store.write_path_template = test_case.write_path_template + file_store.read_path_template = test_case.read_path_template + check_file_store_paths(file_store, test_case) + + +def test_file_store_paths_can_change(mock_now: Mock) -> None: + fs = DummyFS() + + file_store = DummyFileStorePlugin( + name="test_file_store", + write_path_template="", + read_path_template="", + root=None, + reg=fs, + ) + + for test_case in FILE_STORE_TEST_CASES[0], FILE_STORE_TEST_CASES[1]: + file_store.reg_root = test_case.root + file_store.write_path_template = test_case.write_path_template + file_store.read_path_template = test_case.read_path_template + check_file_store_paths(file_store, test_case) + + +def check_file_store_paths( + file_store: DummyFileStorePlugin, + test_case: FileStorePathTestCase, +) -> None: + file_store.stage() + file_store.generate_datum("test", 0, {}) + + file_path = file_store.file_path.get() + assert file_path == test_case.expected_write_path + + docs = list(file_store.collect_asset_docs()) + + assert docs[0][0] == "resource" + resource = docs[0][1] + + assert resource["spec"] == "FS_TEST" + assert resource["root"] == test_case.expected_resource_root + + resource_path_directory = str(Path(resource["resource_path"]).parent) + assert resource_path_directory == test_case.expected_resource_path_directory + + file_store.unstage() + + @pytest.mark.adsim @pytest.mark.xfail def test_many_connect(ad_prefix, cleanup):