Files
CoreRasurae 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
test: add coverage gap tests improving coverage from 81.4% to 96.90%
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.
2026-06-02 20:02:01 +01:00

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!")