Skip to content

Commit

Permalink
added raising tests to all files. read WIP
Browse files Browse the repository at this point in the history
  • Loading branch information
TamarZanzouri committed Dec 23, 2024
1 parent 6d37f1f commit 192e39d
Show file tree
Hide file tree
Showing 4 changed files with 252 additions and 8 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,7 @@ async def test_close_lid_raises_no_gripper_offset(
subject: CloseLidImpl,
absorbance_def: LabwareDefinition,
) -> None:
"""Should raise an error that the hardware module not found."""
"""Should raise an error that gripper offset not found."""
params = CloseLidParams(
moduleId="unverified-module-id",
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from opentrons.drivers.types import ABSMeasurementMode
from opentrons.hardware_control.modules import AbsorbanceReader
from opentrons.protocol_engine.errors import InvalidWavelengthError

from opentrons.protocol_engine.execution import EquipmentHandler
from opentrons.protocol_engine.state import update_types
Expand All @@ -23,19 +24,27 @@
)


@pytest.fixture
def subject(
state_view: StateView,
equipment: EquipmentHandler,
) -> InitializeImpl:
"""Subject command implementation to test."""
return InitializeImpl(state_view=state_view, equipment=equipment)


@pytest.mark.parametrize(
"input_sample_wave_length, input_measure_mode", [([1, 2], "multi"), ([1], "single")]
)
async def test_absorbance_reader_implementation(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
input_sample_wave_length: List[int],
input_measure_mode: str,
subject: InitializeImpl,
state_view: StateView,
equipment: EquipmentHandler,
) -> None:
"""It should validate, find hardware module if not virtualized, and disengage."""
subject = InitializeImpl(state_view=state_view, equipment=equipment)

params = InitializeParams(
moduleId="unverified-module-id",
measureMode=input_measure_mode, # type: ignore[arg-type]
Expand Down Expand Up @@ -82,3 +91,149 @@ async def test_absorbance_reader_implementation(
)
),
)


@pytest.mark.parametrize(
"input_sample_wave_length, input_measure_mode",
[([1, 2, 3], "multi"), ([3], "single")],
)
async def test_initialize_raises_invalid_wave_length(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
subject: InitializeImpl,
input_sample_wave_length: List[int],
input_measure_mode: str,
) -> None:
"""Should raise an InvalidWavelengthError error."""
params = InitializeParams(
moduleId="unverified-module-id",
measureMode=input_measure_mode, # type: ignore[arg-type]
sampleWavelengths=input_sample_wave_length,
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
verified_module_id = AbsorbanceReaderId("module-id")
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)

decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2])

with pytest.raises(InvalidWavelengthError):
await subject.execute(params=params)


@pytest.mark.parametrize(
"input_sample_wave_length, input_measure_mode",
[([], "multi"), ([], "single")],
)
async def test_initialize_raises_measure_mode_not_matching(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
subject: InitializeImpl,
input_sample_wave_length: List[int],
input_measure_mode: str,
) -> None:
"""Should raise an error that the measure mode does not match sample wave."""
params = InitializeParams(
moduleId="unverified-module-id",
measureMode=input_measure_mode, # type: ignore[arg-type]
sampleWavelengths=input_sample_wave_length,
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
verified_module_id = AbsorbanceReaderId("module-id")
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)

decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2])

with pytest.raises(ValueError):
await subject.execute(params=params)


async def test_initialize_single_raises_reference_wave_length_not_matching(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
subject: InitializeImpl,
) -> None:
"""Should raise an error that the measure mode does not match sample wave."""
params = InitializeParams(
moduleId="unverified-module-id",
measureMode="single",
sampleWavelengths=[1],
referenceWavelength=3,
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
verified_module_id = AbsorbanceReaderId("module-id")
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)

decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2])

with pytest.raises(InvalidWavelengthError):
await subject.execute(params=params)


async def test_initialize_multi_raises_no_reference_wave_length(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
subject: InitializeImpl,
) -> None:
"""Should raise an error that the measure mode does not match sample wave."""
params = InitializeParams(
moduleId="unverified-module-id",
measureMode="multi",
sampleWavelengths=[1, 2],
referenceWavelength=3,
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
verified_module_id = AbsorbanceReaderId("module-id")
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)

decoy.when((absorbance_module_hw.supported_wavelengths)).then_return([1, 2])

with pytest.raises(ValueError):
await subject.execute(params=params)
Original file line number Diff line number Diff line change
Expand Up @@ -141,7 +141,7 @@ async def test_absorbance_reader_implementation(
)


async def test_close_lid_raises_no_module(
async def test_open_lid_raises_no_module(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
Expand All @@ -166,14 +166,14 @@ async def test_close_lid_raises_no_module(
await subject.execute(params=params)


async def test_close_lid_raises_no_gripper_offset(
async def test_open_lid_raises_no_gripper_offset(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
subject: OpenLidImpl,
absorbance_def: LabwareDefinition,
) -> None:
"""Should raise an error that the hardware module not found."""
"""Should raise an error that gripper offset not found."""
params = OpenLidParams(
moduleId="unverified-module-id",
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from opentrons.drivers.types import ABSMeasurementMode, ABSMeasurementConfig
from opentrons.hardware_control.modules import AbsorbanceReader
from opentrons.protocol_engine.errors import CannotPerformModuleAction, StorageLimitReachedError

from opentrons.protocol_engine.execution import EquipmentHandler
from opentrons.protocol_engine.resources import FileProvider
Expand Down Expand Up @@ -84,3 +85,91 @@ async def test_absorbance_reader_implementation(
),
),
)


async def test_read_raises_cannot_preform_action(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
file_provider: FileProvider,
) -> None:
"""It should raise CannotPerformModuleAction when not configured/lid is not on."""
subject = ReadAbsorbanceImpl(
state_view=state_view, equipment=equipment, file_provider=file_provider
)

params = ReadAbsorbanceParams(
moduleId="unverified-module-id",
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

verified_module_id = AbsorbanceReaderId("module-id")

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)

decoy.when(mabsorbance_module_substate.configured).then_return(False)

with pytest.raises(CannotPerformModuleAction):
await subject.execute(params=params)

decoy.when(mabsorbance_module_substate.configured).then_return(True)

decoy.when(mabsorbance_module_substate.is_lid_on).then_return(False)

with pytest.raises(CannotPerformModuleAction):
await subject.execute(params=params)


async def test_read_raises_storage_limit(
decoy: Decoy,
state_view: StateView,
equipment: EquipmentHandler,
file_provider: FileProvider,
) -> None:
"""It should raise StorageLimitReachedError when not configured/lid is not on."""
subject = ReadAbsorbanceImpl(
state_view=state_view, equipment=equipment, file_provider=file_provider
)

params = ReadAbsorbanceParams(
moduleId="unverified-module-id",
)

mabsorbance_module_substate = decoy.mock(cls=AbsorbanceReaderSubState)
absorbance_module_hw = decoy.mock(cls=AbsorbanceReader)

verified_module_id = AbsorbanceReaderId("module-id")

decoy.when(
state_view.modules.get_absorbance_reader_substate("unverified-module-id")
).then_return(mabsorbance_module_substate)

decoy.when(mabsorbance_module_substate.module_id).then_return(verified_module_id)

decoy.when(equipment.get_module_hardware_api(verified_module_id)).then_return(
absorbance_module_hw
)
decoy.when(await absorbance_module_hw.start_measure()).then_return([[1.2, 1.3]])

decoy.when(absorbance_module_hw._measurement_config).then_return(
ABSMeasurementConfig(
measure_mode=ABSMeasurementMode.SINGLE,
sample_wavelengths=[1, 2],
reference_wavelength=None,
)
)
decoy.when(mabsorbance_module_substate.configured_wavelengths).then_return(11)

decoy.when(state_view.files.get_filecount()).then_return(390)
with pytest.raises(StorageLimitReachedError):
await subject.execute(params=params)

0 comments on commit 192e39d

Please sign in to comment.