9753b31c7e
CI / quality (push) Successful in 36s
CI / lint (push) Successful in 38s
CI / typecheck (push) Successful in 49s
CI / security (push) Successful in 51s
CI / integration_tests (push) Successful in 55s
CI / unit_tests (push) Successful in 3m35s
CI / build (push) Successful in 33s
CI / coverage (push) Successful in 3m36s
CI / status-check (push) Successful in 3s
Add 13 BDD scenarios covering previously uncovered code paths: - SAFE_BUILTINS validation (sandbox.py: 0% → 100%) - ConfigurationError re-raise path (config.py: 99.3% → 100%) - CLI hello/main functions (cli.py: 72.7% → 90.9%) - GraphState message truncation (state.py: 98.7% → 100%) - ProgressBarManager update/context rendering (progress.py: 0% → 87.7%) - MessageRouter regex/exact/contains routing (message_router.py: 0% → 59.4%) - RoutingAdapter parse_routing_command (routing_adapter.py) - DynamicRouterNode pattern-based routing (dynamic_router.py) - EnhancedTemplateRegistry unknown template type (enhanced_registry.py: 99.2%) - CompositeAgent null-graph error path (composite.py: 97.8% → 98.6%) Cover routing_adapter.py (17% → 100%): all GOTO/ROUTE patterns, create_routing_node, create_conditional_router, dynamic config conversion. Cover dynamic_router.py (21% → 87%): execute with empty/dict/string messages, extract_message with colon parsing, config creation, graph extension with edge generation. Cover message_router.py (59% → 78%): regex, exact, contains, prefix, suffix match types, invalid regex handling, non-string message, set_state. Exercise Node._prepare_conversation_history with invalid configs, _runtime error paths, _execute_message_router with rules, _execute_agent with current_message and metadata propagation, _execute_function with dynamic_router, and _execute_conditional with content_contains/content_not_contains/content_starts_with and custom condition types. Improves nodes.py from 71.3% to 72.5%. Exercise PureGraphConfig, PureLangGraph init with dict/config, RxPyLangGraphBridge registration/connection/lookup, ReactiveStreamRouter operator creation and condition functions, ReactiveConfigParser config/route/graph parsing, ToolAgent tool execution with JSON, space-separated, single, file_read, progress_bar invocations, and ReactiveCleverAgentsApp init/dispose/visualization.
371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""Simple step definitions for template loaders."""
|
|
|
|
import os
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveractors.core.exceptions import TemplateError
|
|
from cleveractors.templates.loaders import (
|
|
ConfigTemplateLoader,
|
|
DirectoryTemplateLoader,
|
|
FileTemplateLoader,
|
|
TemplateLoader,
|
|
load_from_file,
|
|
load_from_string,
|
|
)
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
|
|
@given("I initialize the test environment")
|
|
def step_initialize_environment(context):
|
|
"""Initialize test environment."""
|
|
context.results = []
|
|
|
|
|
|
@when("I test all template loader classes")
|
|
def step_test_all_loaders(context):
|
|
"""Test all template loader functionality."""
|
|
results = []
|
|
|
|
# Test base TemplateLoader
|
|
try:
|
|
renderer = Mock(spec=TemplateRenderer)
|
|
loader = TemplateLoader(renderer)
|
|
assert loader.renderer == renderer
|
|
|
|
try:
|
|
loader.load()
|
|
results.append("ERROR: Should have raised NotImplementedError")
|
|
except NotImplementedError as e:
|
|
assert "Method 'load' is not implemented yet" in str(e)
|
|
results.append("TemplateLoader base class: PASS")
|
|
except Exception as e:
|
|
results.append(f"TemplateLoader base class: FAIL - {e}")
|
|
|
|
# Test FileTemplateLoader
|
|
temp_files = []
|
|
try:
|
|
# Create temp files
|
|
temp1 = tempfile.NamedTemporaryFile(mode="w", suffix=".j2", delete=False)
|
|
temp1.write("Template 1: {{ var1 }}")
|
|
temp1.close()
|
|
temp_files.append(temp1.name)
|
|
|
|
temp2 = tempfile.NamedTemporaryFile(mode="w", suffix=".j2", delete=False)
|
|
temp2.write("Template 2: {{ var2 }}")
|
|
temp2.close()
|
|
temp_files.append(temp2.name)
|
|
|
|
# Test with valid files
|
|
file_paths = [Path(f) for f in temp_files]
|
|
file_loader = FileTemplateLoader(renderer, file_paths)
|
|
assert file_loader.file_paths == file_paths
|
|
|
|
renderer.register_template.reset_mock()
|
|
file_loader.load()
|
|
assert renderer.register_template.call_count == 2
|
|
|
|
# Test with non-existent file
|
|
bad_path = Path("/non/existent/file.j2")
|
|
bad_loader = FileTemplateLoader(renderer, [bad_path])
|
|
try:
|
|
bad_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for non-existent file"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Template file not found" in str(e)
|
|
|
|
# Test with unreadable file
|
|
unreadable_temp = tempfile.NamedTemporaryFile(mode="w", delete=False)
|
|
unreadable_temp.write("content")
|
|
unreadable_temp.close()
|
|
temp_files.append(unreadable_temp.name)
|
|
|
|
unreadable_loader = FileTemplateLoader(renderer, [Path(unreadable_temp.name)])
|
|
with patch("builtins.open", side_effect=PermissionError("Permission denied")):
|
|
try:
|
|
unreadable_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for unreadable file"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Failed to load template from file" in str(e)
|
|
|
|
results.append("FileTemplateLoader: PASS")
|
|
|
|
except Exception as e:
|
|
results.append(f"FileTemplateLoader: FAIL - {e}")
|
|
finally:
|
|
# Clean up temp files
|
|
for temp_file in temp_files:
|
|
try:
|
|
os.chmod(temp_file, 0o644)
|
|
os.unlink(temp_file)
|
|
except (OSError, PermissionError):
|
|
pass
|
|
|
|
# Test DirectoryTemplateLoader
|
|
temp_dirs = []
|
|
try:
|
|
# Create temp directory with files
|
|
temp_dir = Path(tempfile.mkdtemp())
|
|
temp_dirs.append(temp_dir)
|
|
|
|
# Create template files
|
|
(temp_dir / "template1.j2").write_text("Template 1: {{ var1 }}")
|
|
(temp_dir / "template2.j2").write_text("Template 2: {{ var2 }}")
|
|
|
|
# Create nested directory
|
|
nested_dir = temp_dir / "nested"
|
|
nested_dir.mkdir()
|
|
(nested_dir / "nested.j2").write_text("Nested: {{ var3 }}")
|
|
|
|
# Test basic directory loading
|
|
dir_loader = DirectoryTemplateLoader(renderer, temp_dir)
|
|
assert dir_loader.directory_path == temp_dir
|
|
assert dir_loader.recursive is True
|
|
assert dir_loader.pattern == "*.j2"
|
|
|
|
renderer.register_template.reset_mock()
|
|
dir_loader.load()
|
|
# Should load 3 templates (2 root + 1 nested)
|
|
assert renderer.register_template.call_count == 3
|
|
|
|
# Test non-recursive
|
|
non_recursive_loader = DirectoryTemplateLoader(
|
|
renderer, temp_dir, recursive=False
|
|
)
|
|
renderer.register_template.reset_mock()
|
|
non_recursive_loader.load()
|
|
# Should load only 2 templates (root only)
|
|
assert renderer.register_template.call_count == 2
|
|
|
|
# Test custom pattern
|
|
(temp_dir / "template.txt").write_text("TXT template")
|
|
pattern_loader = DirectoryTemplateLoader(renderer, temp_dir, pattern="*.txt")
|
|
renderer.register_template.reset_mock()
|
|
pattern_loader.load()
|
|
# Should load only 1 template (txt file)
|
|
assert renderer.register_template.call_count == 1
|
|
|
|
# Test non-existent directory
|
|
bad_dir_loader = DirectoryTemplateLoader(renderer, Path("/non/existent/dir"))
|
|
try:
|
|
bad_dir_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for non-existent directory"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Template directory not found" in str(e)
|
|
|
|
# Test file instead of directory
|
|
temp_file = tempfile.NamedTemporaryFile(delete=False)
|
|
temp_file.close()
|
|
temp_files.append(temp_file.name)
|
|
|
|
file_as_dir_loader = DirectoryTemplateLoader(renderer, Path(temp_file.name))
|
|
try:
|
|
file_as_dir_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for file instead of directory"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Not a directory" in str(e)
|
|
|
|
# Test directory with unreadable file
|
|
unreadable_file = temp_dir / "unreadable.j2"
|
|
unreadable_file.write_text("content")
|
|
|
|
problem_loader = DirectoryTemplateLoader(renderer, temp_dir)
|
|
with patch("builtins.open", side_effect=PermissionError("Permission denied")):
|
|
try:
|
|
problem_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for unreadable file in directory"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Failed to load template from file" in str(e)
|
|
|
|
results.append("DirectoryTemplateLoader: PASS")
|
|
|
|
except Exception as e:
|
|
results.append(f"DirectoryTemplateLoader: FAIL - {e}")
|
|
finally:
|
|
# Clean up temp directories
|
|
for temp_dir in temp_dirs:
|
|
try:
|
|
import shutil
|
|
|
|
# Restore permissions
|
|
for root, dirs, files in os.walk(temp_dir):
|
|
for d in dirs:
|
|
os.chmod(os.path.join(root, d), 0o755)
|
|
for f in files:
|
|
os.chmod(os.path.join(root, f), 0o644)
|
|
shutil.rmtree(temp_dir)
|
|
except (OSError, PermissionError):
|
|
pass
|
|
|
|
# Test ConfigTemplateLoader
|
|
try:
|
|
# Test with string templates
|
|
config = {
|
|
"templates": {
|
|
"template1": "String template 1: {{ var1 }}",
|
|
"template2": "String template 2: {{ var2 }}",
|
|
}
|
|
}
|
|
config_loader = ConfigTemplateLoader(renderer, config)
|
|
assert config_loader.config == config
|
|
|
|
renderer.register_template.reset_mock()
|
|
config_loader.load()
|
|
assert renderer.register_template.call_count == 2
|
|
|
|
# Test with dict templates
|
|
dict_config = {
|
|
"templates": {
|
|
"template1": {
|
|
"content": "Dict template 1: {{ var1 }}",
|
|
"metadata": {"type": "dict"},
|
|
},
|
|
"template2": {
|
|
"content": "Dict template 2: {{ var2 }}",
|
|
"metadata": {"type": "dict"},
|
|
},
|
|
}
|
|
}
|
|
dict_loader = ConfigTemplateLoader(renderer, dict_config)
|
|
renderer.register_template.reset_mock()
|
|
dict_loader.load()
|
|
assert renderer.register_template.call_count == 2
|
|
|
|
# Test with empty config
|
|
empty_loader = ConfigTemplateLoader(renderer, {})
|
|
renderer.register_template.reset_mock()
|
|
empty_loader.load()
|
|
assert renderer.register_template.call_count == 0
|
|
|
|
# Test without templates section
|
|
no_templates_loader = ConfigTemplateLoader(renderer, {"other": "section"})
|
|
renderer.register_template.reset_mock()
|
|
no_templates_loader.load()
|
|
assert renderer.register_template.call_count == 0
|
|
|
|
# Test with invalid template definition
|
|
invalid_config = {
|
|
"templates": {"valid": "Valid content", "invalid": ["invalid", "list"]}
|
|
}
|
|
invalid_loader = ConfigTemplateLoader(renderer, invalid_config)
|
|
try:
|
|
invalid_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for invalid template definition"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Invalid template definition" in str(e)
|
|
|
|
# Test with exception during processing
|
|
exception_renderer = Mock(spec=TemplateRenderer)
|
|
exception_renderer.register_template.side_effect = Exception("Mock exception")
|
|
exception_loader = ConfigTemplateLoader(
|
|
exception_renderer, {"templates": {"t": "content"}}
|
|
)
|
|
try:
|
|
exception_loader.load()
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for processing exception"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Failed to load templates from configuration" in str(e)
|
|
|
|
results.append("ConfigTemplateLoader: PASS")
|
|
|
|
except Exception as e:
|
|
results.append(f"ConfigTemplateLoader: FAIL - {e}")
|
|
|
|
# Test load_from_file function
|
|
temp_file = None
|
|
try:
|
|
temp_file = tempfile.NamedTemporaryFile(mode="w", delete=False)
|
|
test_content = "Test file content: {{ variable }}"
|
|
temp_file.write(test_content)
|
|
temp_file.close()
|
|
|
|
# Test with existing file
|
|
result = load_from_file(temp_file.name)
|
|
assert result == test_content
|
|
|
|
# Test with non-existent file
|
|
result = load_from_file("/non/existent/file.txt")
|
|
assert result is None
|
|
|
|
# Test with unreadable file
|
|
with patch("builtins.open", side_effect=PermissionError("Permission denied")):
|
|
try:
|
|
load_from_file(temp_file.name)
|
|
results.append(
|
|
"ERROR: Should have raised TemplateError for unreadable file"
|
|
)
|
|
except TemplateError as e:
|
|
assert "Failed to load template from file" in str(e)
|
|
|
|
results.append("load_from_file: PASS")
|
|
|
|
except Exception as e:
|
|
results.append(f"load_from_file: FAIL - {e}")
|
|
finally:
|
|
if temp_file:
|
|
try:
|
|
os.chmod(temp_file.name, 0o644)
|
|
os.unlink(temp_file.name)
|
|
except (OSError, PermissionError):
|
|
pass
|
|
|
|
# Test load_from_string function
|
|
try:
|
|
test_string = "Template string content: {{ variable }}"
|
|
result = load_from_string(test_string)
|
|
assert result == test_string
|
|
assert result is test_string # Should be same object
|
|
|
|
results.append("load_from_string: PASS")
|
|
|
|
except Exception as e:
|
|
results.append(f"load_from_string: FAIL - {e}")
|
|
|
|
context.results = results
|
|
|
|
|
|
@then("the loaders should work correctly")
|
|
def step_verify_loaders_work(context):
|
|
"""Verify all loaders work correctly."""
|
|
for result in context.results:
|
|
print(f" {result}")
|
|
|
|
# Check that we have passing results for all major components
|
|
passing_tests = [r for r in context.results if "PASS" in r]
|
|
failing_tests = [r for r in context.results if "FAIL" in r or "ERROR" in r]
|
|
|
|
if failing_tests:
|
|
print(f"FAILURES: {failing_tests}")
|
|
raise AssertionError(f"Some tests failed: {failing_tests}")
|
|
|
|
expected_passes = [
|
|
"TemplateLoader base class: PASS",
|
|
"FileTemplateLoader: PASS",
|
|
"DirectoryTemplateLoader: PASS",
|
|
"ConfigTemplateLoader: PASS",
|
|
"load_from_file: PASS",
|
|
"load_from_string: PASS",
|
|
]
|
|
|
|
for expected in expected_passes:
|
|
assert expected in passing_tests, f"Missing expected pass: {expected}"
|
|
|
|
print(f"All {len(passing_tests)} tests passed!")
|