Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion pylabrobot/thermocycling/opentrons_backend.py
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,10 @@ async def get_block_status(self) -> BlockStatus:
return BlockStatus.IDLE

async def get_hold_time(self) -> float:
return cast(float, self._find_module().get("holdTime", 0.0))
hold_time = self._find_module().get("holdTime")
if hold_time is None:
raise RuntimeError("Hold time is not available. Is a profile running?")
return cast(float, hold_time)

async def get_current_cycle_index(self) -> int:
"""Get the zero-based index of the current cycle from the Opentrons API."""
Expand Down
8 changes: 8 additions & 0 deletions pylabrobot/thermocycling/opentrons_backend_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -109,3 +109,11 @@ async def test_getters_return_correct_data(self, mock_list_connected_modules):
# assert await self.thermocycler_backend.get_total_cycle_count() == 10
assert await self.thermocycler_backend.get_current_step_index() == 0 # 1 - 1 = 0 (zero-based)
assert await self.thermocycler_backend.get_total_step_count() == 3

@patch("pylabrobot.thermocycling.opentrons_backend.list_connected_modules")
async def test_get_hold_time_raises_if_not_running(self, mock_list_connected_modules):
mock_list_connected_modules.return_value = [{"id": "test_id", "data": {}}]

with self.assertRaises(RuntimeError) as e:
await self.thermocycler_backend.get_hold_time()
self.assertEqual(str(e.exception), "Hold time is not available. Is a profile running?")
14 changes: 14 additions & 0 deletions pylabrobot/thermocycling/thermocycler_tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,3 +137,17 @@ async def test_is_profile_running_logic(self):
self.tc.backend.get_total_step_count.return_value = total_steps # type: ignore
print(f"Testing with hold={hold}, cycle={cycle}, total_cycles={total_cycles}, ")
assert await self.tc.is_profile_running() is expected

async def test_is_profile_running_raises_if_no_profile_running(self):
"""RuntimeError from get_hold_time should propagate."""
self.tc.backend.get_hold_time.side_effect = RuntimeError("No profile is running") # type: ignore

with self.assertRaises(RuntimeError):
await self.tc.is_profile_running()

async def test_is_profile_running_preserves_not_implemented_error(self):
"""Unsupported backends should still surface NotImplementedError."""
self.tc.backend.get_hold_time.side_effect = NotImplementedError # type: ignore

with self.assertRaises(NotImplementedError):
await self.tc.is_profile_running()
Loading