From 3a8953f2f99eb9208ec0655676642775ecbe1a2b Mon Sep 17 00:00:00 2001 From: carloea2 Date: Sun, 30 Aug 2026 21:08:58 -0600 Subject: [PATCH] fix(pyamber): validate executor definitions --- .../core/architecture/managers/executor_manager.py | 3 ++- .../architecture/managers/test_executor_manager.py | 10 ++++------ 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/amber/src/main/python/core/architecture/managers/executor_manager.py b/amber/src/main/python/core/architecture/managers/executor_manager.py index 8daa0b9e644..b2f544fe55f 100644 --- a/amber/src/main/python/core/architecture/managers/executor_manager.py +++ b/amber/src/main/python/core/architecture/managers/executor_manager.py @@ -125,7 +125,8 @@ def load_executor_definition(self, code: str) -> type(Operator): executors = list( filter(self.is_concrete_operator, executor_module.__dict__.values()) ) - assert len(executors) == 1, "There should be one and only one Operator defined" + if len(executors) != 1: + raise ValueError("There should be one and only one Operator defined") return executors[0] def close(self) -> None: diff --git a/amber/src/test/python/core/architecture/managers/test_executor_manager.py b/amber/src/test/python/core/architecture/managers/test_executor_manager.py index 07ac054c062..4928d22c5dc 100644 --- a/amber/src/test/python/core/architecture/managers/test_executor_manager.py +++ b/amber/src/test/python/core/architecture/managers/test_executor_manager.py @@ -407,17 +407,15 @@ def test_update_with_source_mismatch_raises_assertion(self, initialized_manager) ) assert "SourceOperator API" in str(exc_info.value) - def test_update_with_no_operator_class_raises_assertion(self, initialized_manager): - # load_executor_definition asserts exactly one Operator subclass exists - # in the module — an empty module trips that assertion. - with pytest.raises(AssertionError) as exc_info: + def test_update_with_no_operator_class_is_rejected(self, initialized_manager): + with pytest.raises(ValueError) as exc_info: initialized_manager.update_executor(code=NO_OPERATOR_CODE, is_source=False) assert "one and only one Operator" in str(exc_info.value) - def test_update_with_multiple_operator_classes_raises_assertion( + def test_update_with_multiple_operator_classes_is_rejected( self, initialized_manager ): - with pytest.raises(AssertionError) as exc_info: + with pytest.raises(ValueError) as exc_info: initialized_manager.update_executor( code=TWO_OPERATORS_CODE, is_source=False )