Source code for ewokstomo.tests.test_utils
from pathlib import Path
from importlib.resources import files
from unittest.mock import Mock
from blissdata.redis_engine.scan import ScanState
from ewokstomo.tasks.utils import (
wait_for_scan_state,
_get_output_format_from_nabu_config,
)
DATA_ROOT = Path(__file__).resolve().parent / "data"
COLLECTION = "TestEwoksTomo"
ROTATION_AXIS_POSITION = 8.0
[docs]
def get_json_file(file_name: str) -> Path:
"""Return the path to a workflow JSON file shipped with ewokstomo."""
return Path(str(files("ewokstomo.workflows").joinpath(file_name)))
[docs]
def get_data_file(file_name: str) -> str:
"""Return the path to a raw Bliss HDF5 file for the given dataset."""
return str(DATA_ROOT / "RAW_DATA" / COLLECTION / file_name / f"{file_name}.h5")
[docs]
def get_data_dir(scan_name: str) -> Path:
"""Return the PROCESSED_DATA directory for the given dataset."""
return DATA_ROOT / "PROCESSED_DATA" / COLLECTION / scan_name
[docs]
def get_raw_data_dir(sample_dataset: str) -> Path:
"""Return the RAW_DATA directory for the given dataset."""
return DATA_ROOT / "RAW_DATA" / COLLECTION / sample_dataset
[docs]
def test_wait_for_scan_state_advances_until_reached():
class FakeScan:
def __init__(self):
self.state = ScanState.CREATED
self.updates = 0
def update(self):
self.updates += 1
self.state = ScanState.STARTED
scan = FakeScan()
wait_for_scan_state(scan, ScanState.STARTED)
assert scan.state == ScanState.STARTED
assert scan.updates == 1
[docs]
def test_wait_for_scan_state_returns_when_already_reached():
class FakeScan:
def __init__(self):
self.state = ScanState.CLOSED
self.updates = 0
def update(self):
self.updates += 1
scan = FakeScan()
wait_for_scan_state(scan, ScanState.STARTED)
assert scan.updates == 0
[docs]
def test_wait_for_scan_state_reaches_desired_state():
scan = Mock()
states = [
ScanState.CREATED.value,
ScanState.PREPARED.value,
ScanState.STARTED.value,
ScanState.STOPPED.value,
]
scan.state = states[0]
def update_state():
scan.state = states[scan.update.call_count]
scan.update.side_effect = update_state
wait_for_scan_state(scan, ScanState.STOPPED)
assert scan.update.call_count == 3
[docs]
def test_wait_for_scan_state_higher_state():
scan = Mock()
scan.state = ScanState.CREATED.value
def update_state():
scan.state = ScanState.STARTED.value
scan.update.side_effect = update_state
wait_for_scan_state(scan, ScanState.PREPARED)
assert scan.update.call_count == 1
