mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-06-29 23:41:12 +08:00
b9445c67e2
## Summary - Coerce `None` canvas values to `""` before string comparison operators in `Switch.process_operator`. - Prevents `AttributeError` when upstream components yield `None` and the Switch uses contains/start with/end with. ## Test plan - [x] `.v/bin/python -m ruff check agent/component/switch.py test/unit_test/agent/component/test_switch.py` - [x] `.v/bin/python -m pytest test/unit_test/agent/component/test_switch.py -q` (3 passed) Fixes #16315 --------- Co-authored-by: Harsh Kashyap <harshkashyap@Harshs-MacBook-Pro.local>
100 lines
2.5 KiB
Python
100 lines
2.5 KiB
Python
import pytest
|
|
|
|
from agent.component.switch import Switch, SwitchParam
|
|
|
|
|
|
class _Canvas:
|
|
def __init__(self, variables=None):
|
|
self.variables = variables or {}
|
|
|
|
def is_canceled(self):
|
|
return False
|
|
|
|
def get_variable_value(self, cpn_id):
|
|
return self.variables[cpn_id]
|
|
|
|
def get_component_name(self, cpn_id):
|
|
return cpn_id
|
|
|
|
|
|
def _switch(param, variables=None):
|
|
cpn = Switch.__new__(Switch)
|
|
cpn._canvas = _Canvas(variables)
|
|
cpn._id = "switch"
|
|
cpn._param = param
|
|
return cpn
|
|
|
|
|
|
def test_switch_empty_condition_falls_through_to_else():
|
|
param = SwitchParam()
|
|
param.conditions = [
|
|
{
|
|
"logical_operator": "and",
|
|
"items": [{"cpn_id": "", "operator": "=", "value": "yes"}],
|
|
"to": ["case_target"],
|
|
}
|
|
]
|
|
param.end_cpn_ids = ["else_target"]
|
|
|
|
cpn = _switch(param)
|
|
cpn._invoke()
|
|
|
|
assert cpn.output("_next") == ["else_target"]
|
|
assert cpn.output("next") == ["else_target"]
|
|
|
|
|
|
def test_switch_non_empty_and_condition_still_matches():
|
|
param = SwitchParam()
|
|
param.conditions = [
|
|
{
|
|
"logical_operator": "and",
|
|
"items": [{"cpn_id": "answer", "operator": "=", "value": "yes"}],
|
|
"to": ["case_target"],
|
|
}
|
|
]
|
|
param.end_cpn_ids = ["else_target"]
|
|
|
|
cpn = _switch(param, {"answer": "yes"})
|
|
cpn._invoke()
|
|
|
|
assert cpn.output("_next") == ["case_target"]
|
|
assert cpn.output("next") == ["case_target"]
|
|
|
|
|
|
@pytest.mark.p1
|
|
def test_switch_none_input_contains_falls_through_to_else():
|
|
param = SwitchParam()
|
|
param.conditions = [
|
|
{
|
|
"logical_operator": "and",
|
|
"items": [{"cpn_id": "answer", "operator": "contains", "value": "foo"}],
|
|
"to": ["case_target"],
|
|
}
|
|
]
|
|
param.end_cpn_ids = ["else_target"]
|
|
|
|
cpn = _switch(param, {"answer": None})
|
|
cpn._invoke()
|
|
|
|
assert cpn.output("_next") == ["else_target"]
|
|
assert cpn.output("next") == ["else_target"]
|
|
|
|
|
|
@pytest.mark.p1
|
|
def test_switch_none_value_contains_does_not_raise():
|
|
param = SwitchParam()
|
|
param.conditions = [
|
|
{
|
|
"logical_operator": "and",
|
|
"items": [{"cpn_id": "answer", "operator": "contains", "value": None}],
|
|
"to": ["case_target"],
|
|
}
|
|
]
|
|
param.end_cpn_ids = ["else_target"]
|
|
|
|
cpn = _switch(param, {"answer": "foobar"})
|
|
cpn._invoke()
|
|
|
|
assert cpn.output("_next") == ["case_target"]
|
|
assert cpn.output("next") == ["case_target"]
|