diff --git a/test/unit_test/agent/component/test_switch_empty_condition.py b/test/unit_test/agent/component/test_switch_empty_condition.py new file mode 100644 index 0000000000..b4d6de5a8e --- /dev/null +++ b/test/unit_test/agent/component/test_switch_empty_condition.py @@ -0,0 +1,94 @@ +# +# Copyright 2025 The InfiniFlow Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Unit tests for ``Switch`` condition evaluation. + +Regression: an "and" condition whose ``items`` are all skipped (empty +``cpn_id``) leaves the per-condition result list empty, and ``all([]) is True``, +so the Switch matched that condition unconditionally and routed to it before +ever reaching the else/end branch. +""" + +from types import SimpleNamespace + +import pytest + +from agent.component.switch import Switch + + +pytestmark = [pytest.mark.p1] + + +class _FakeCanvas: + def __init__(self, refs=None): + self._refs = refs or {} + + def get_variable_value(self, token): + return self._refs.get(token) + + def get_component_name(self, cpn_id): + return cpn_id + + +def _make_switch(conditions, end_cpn_ids, refs=None): + sw = Switch.__new__(Switch) + sw._param = SimpleNamespace(conditions=conditions, end_cpn_ids=end_cpn_ids) + sw._canvas = _FakeCanvas(refs) + sw.check_if_canceled = lambda _msg: False + sw.set_input_value = lambda *_args, **_kwargs: None + outputs = {} + sw.set_output = lambda key, value: outputs.__setitem__(key, value) + sw._invoke() + return outputs + + +def test_empty_condition_falls_through_to_end(): + # The only item is skipped (blank cpn_id) -> result list is empty. + conditions = [ + { + "logical_operator": "and", + "items": [{"cpn_id": "", "operator": "contains", "value": "x"}], + "to": ["TARGET"], + } + ] + outputs = _make_switch(conditions, end_cpn_ids=["END"]) + assert outputs["_next"] == ["END"] + + +def test_empty_and_items_fall_through_to_end(): + conditions = [ + { + "logical_operator": "and", + "items": [], + "to": ["TARGET"], + } + ] + outputs = _make_switch(conditions, end_cpn_ids=["END"]) + assert outputs["_next"] == ["END"] + + +def test_satisfied_and_condition_still_routes(): + # A genuinely satisfied "and" condition must still route to its target + # (the fix must not break the real all(res) path). + conditions = [ + { + "logical_operator": "and", + "items": [{"cpn_id": "c@out", "operator": "contains", "value": "hello"}], + "to": ["TARGET"], + } + ] + outputs = _make_switch(conditions, end_cpn_ids=["END"], refs={"c@out": "hello world"}) + assert outputs["_next"] == ["TARGET"]