Spaces:
Sleeping
Sleeping
| import re | |
| import unittest | |
| from dataclasses import dataclass, field | |
| from typing import Any, Dict, Optional | |
| # --- Mock Context --- | |
| class WorkflowContext: | |
| variables: Dict[str, Any] = field(default_factory=dict) | |
| results: Dict[str, Any] = field(default_factory=dict) | |
| # --- The Code Under Test (extracted from AdvancedWorkflowOrchestrator) --- | |
| class VariableResolver: | |
| def _resolve_variables(self, value: Any, context: WorkflowContext) -> Any: | |
| """Resolve variables in a value (string, dict, or list)""" | |
| if isinstance(value, str): | |
| # Replace {{variable}} with value from context.variables | |
| matches = re.findall(r'\{\{([^}]+)\}\}', value) | |
| for match in matches: | |
| # Support nested access like {{step_id.key}} | |
| if '.' in match: | |
| parts = match.split('.') | |
| step_id = parts[0] | |
| key = parts[1] | |
| if step_id in context.results: | |
| val = context.results[step_id].get(key, "") | |
| value = value.replace(f"{{{{{match}}}}}", str(val)) | |
| elif match in context.variables: | |
| value = value.replace(f"{{{{{match}}}}}", str(context.variables[match])) | |
| return value | |
| elif isinstance(value, dict): | |
| return {k: self._resolve_variables(v, context) for k, v in value.items()} | |
| elif isinstance(value, list): | |
| return [self._resolve_variables(v, context) for v in value] | |
| return value | |
| class TestVariableResolution(unittest.TestCase): | |
| def setUp(self): | |
| self.resolver = VariableResolver() | |
| self.context = WorkflowContext() | |
| self.context.variables = { | |
| "user": "Alice", | |
| "count": 10, | |
| "nested_ptr": "user" | |
| } | |
| self.context.results = { | |
| "step_1": {"output": "Success", "id": 123} | |
| } | |
| def test_basic_resolution(self): | |
| """Test simple variable replacement""" | |
| result = self.resolver._resolve_variables("Hello {{user}}", self.context) | |
| print(f"\n[Test Basic] 'Hello {{{{user}}}}' -> '{result}'") | |
| self.assertEqual(result, "Hello Alice") | |
| def test_step_output_resolution(self): | |
| """Test accessing step results""" | |
| result = self.resolver._resolve_variables("ID: {{step_1.id}}", self.context) | |
| print(f"[Test Step] 'ID: {{{{step_1.id}}}}' -> '{result}'") | |
| self.assertEqual(result, "ID: 123") | |
| def test_undefined_variable(self): | |
| """Test variable that doesn't exist""" | |
| input_str = "Value: {{missing_var}}" | |
| result = self.resolver._resolve_variables(input_str, self.context) | |
| print(f"[Test Undefined] '{input_str}' -> '{result}'") | |
| # CURRENT BEHAVIOR: It likely leaves it as-is because the `if match in context.variables` check fails. | |
| # This confirms "Silent Failure" or "Leakage" | |
| self.assertEqual(result, input_str) | |
| def test_nested_curly_braces(self): | |
| """Test nested braces which Regex struggles with""" | |
| # Intent: Resolve {{nested_ptr}} to "user", then resolve {{user}} to "Alice" | |
| # Actual Regex: likely matches "nested_ptr}} to output {{user" or similar weirdness | |
| input_str = "Double: {{ {{nested_ptr}} }}" | |
| result = self.resolver._resolve_variables(input_str, self.context) | |
| print(f"[Test Nested] '{input_str}' -> '{result}'") | |
| # This will almost certainly fail or produce garbage | |
| self.assertNotEqual(result, "Double: Alice") | |
| def test_partial_match_ambiguity(self): | |
| """Test when regex greedy match might fail""" | |
| input_str = "{{user}} and {{count}}" | |
| result = self.resolver._resolve_variables(input_str, self.context) | |
| print(f"[Test Multi] '{input_str}' -> '{result}'") | |
| self.assertEqual(result, "Alice and 10") | |
| if __name__ == '__main__': | |
| unittest.main() | |