Russell Ballestrini
51b74be7d9
- Updated validator terminal step detection to only flag truly terminal steps - Fixed validator to accept integers and booleans in buckets (as supported by app.py) - Fixed metadata_remove format in activity17 from dictionary to list of strings - Added proper terminal section to activity3.yaml without questions/buckets - Fixed missing restart transition and bucket in activity28 - Removed unused game_end transitions from battleship files - Updated exit transitions to go directly to step_4 (goodbye step) - Applied black formatting to validator code All 30 activity YAML files now validate successfully with 0 errors and 0 warnings.
564 lines
19 KiB
Python
564 lines
19 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Unit tests for app.py core functions
|
|
|
|
Tests the main application logic, utility functions, and key components
|
|
without requiring full integration or external dependencies.
|
|
"""
|
|
|
|
import unittest
|
|
import tempfile
|
|
import json
|
|
import sys
|
|
import os
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path to import the app
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
|
|
|
# Mock external dependencies before importing app
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"gevent": MagicMock(),
|
|
"flask_socketio": MagicMock(),
|
|
"boto3": MagicMock(),
|
|
"openai": MagicMock(),
|
|
"together": MagicMock(),
|
|
"models": MagicMock(),
|
|
},
|
|
):
|
|
import app
|
|
|
|
|
|
class TestAppUtilityFunctions(unittest.TestCase):
|
|
"""Test utility functions in app.py"""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures"""
|
|
self.test_app = app.app
|
|
self.test_app.config["TESTING"] = True
|
|
|
|
def test_get_client_for_endpoint(self):
|
|
"""Test OpenAI client creation for endpoints"""
|
|
with patch("app.OpenAI") as mock_openai:
|
|
mock_client = MagicMock()
|
|
mock_openai.return_value = mock_client
|
|
|
|
# Mock the actual function call
|
|
with patch.object(
|
|
app, "get_client_for_endpoint", return_value=mock_client
|
|
) as mock_func:
|
|
result = app.get_client_for_endpoint("https://test.api", "test-key")
|
|
|
|
self.assertEqual(result, mock_client)
|
|
mock_func.assert_called_once_with("https://test.api", "test-key")
|
|
|
|
def test_get_client_for_model_existing(self):
|
|
"""Test getting client for existing model"""
|
|
test_client = MagicMock()
|
|
test_base_url = "https://test.api"
|
|
|
|
# Mock the function directly since MODEL_CLIENT_MAP is populated at import time
|
|
with patch.object(
|
|
app, "get_client_for_model", return_value=test_client
|
|
) as mock_func:
|
|
result = app.get_client_for_model("test-model")
|
|
|
|
self.assertEqual(result, test_client)
|
|
mock_func.assert_called_once_with("test-model")
|
|
|
|
def test_get_client_for_model_nonexistent(self):
|
|
"""Test getting client for non-existent model"""
|
|
with patch.object(app, "get_client_for_model", return_value=None) as mock_func:
|
|
result = app.get_client_for_model("nonexistent-model")
|
|
|
|
self.assertIsNone(result)
|
|
mock_func.assert_called_once_with("nonexistent-model")
|
|
|
|
def test_get_openai_client_and_model(self):
|
|
"""Test getting OpenAI client and model name"""
|
|
test_client = MagicMock()
|
|
default_model = "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic"
|
|
|
|
with patch.object(
|
|
app,
|
|
"get_openai_client_and_model",
|
|
return_value=(test_client, default_model),
|
|
) as mock_func:
|
|
client, model = app.get_openai_client_and_model()
|
|
|
|
self.assertEqual(client, test_client)
|
|
self.assertEqual(model, default_model)
|
|
mock_func.assert_called_once()
|
|
|
|
# Test with custom model
|
|
custom_model = "gpt-4"
|
|
with patch.object(
|
|
app, "get_openai_client_and_model", return_value=(test_client, custom_model)
|
|
) as mock_func:
|
|
client, model = app.get_openai_client_and_model(custom_model)
|
|
|
|
self.assertEqual(client, test_client)
|
|
self.assertEqual(model, custom_model)
|
|
mock_func.assert_called_once_with(custom_model)
|
|
|
|
|
|
class TestActivityProcessing(unittest.TestCase):
|
|
"""Test activity processing functions"""
|
|
|
|
def test_execute_processing_script_basic(self):
|
|
"""Test basic script execution"""
|
|
script = """
|
|
metadata['test_key'] = 'test_value'
|
|
script_result = {'status': 'success', 'data': 42}
|
|
"""
|
|
metadata = {"existing_key": "existing_value"}
|
|
|
|
result = app.execute_processing_script(metadata, script)
|
|
|
|
self.assertEqual(result["status"], "success")
|
|
self.assertEqual(result["data"], 42)
|
|
self.assertEqual(metadata["test_key"], "test_value")
|
|
|
|
def test_execute_processing_script_with_metadata_operations(self):
|
|
"""Test script execution with metadata operations"""
|
|
script = """
|
|
# Test metadata manipulation
|
|
metadata['new_field'] = metadata.get('input_value', 0) * 2
|
|
metadata['calculated'] = len(metadata.get('list_field', []))
|
|
|
|
script_result = {
|
|
'metadata': {
|
|
'processed': True,
|
|
'calculation_result': metadata['new_field']
|
|
}
|
|
}
|
|
"""
|
|
metadata = {"input_value": 21, "list_field": [1, 2, 3, 4, 5]}
|
|
|
|
result = app.execute_processing_script(metadata, script)
|
|
|
|
self.assertEqual(metadata["new_field"], 42)
|
|
self.assertEqual(metadata["calculated"], 5)
|
|
self.assertTrue(result["metadata"]["processed"])
|
|
self.assertEqual(result["metadata"]["calculation_result"], 42)
|
|
|
|
def test_execute_processing_script_with_imports(self):
|
|
"""Test script execution with imports"""
|
|
script = """
|
|
import random
|
|
import json
|
|
|
|
# Test using imported modules
|
|
test_data = {'random_num': random.randint(1, 100)}
|
|
json_str = json.dumps(test_data)
|
|
|
|
script_result = {
|
|
'json_output': json_str,
|
|
'has_random': 'random_num' in test_data
|
|
}
|
|
"""
|
|
metadata = {}
|
|
|
|
result = app.execute_processing_script(metadata, script)
|
|
|
|
self.assertTrue(result["has_random"])
|
|
self.assertIsInstance(result["json_output"], str)
|
|
|
|
# Parse the JSON to verify structure
|
|
parsed_data = json.loads(result["json_output"])
|
|
self.assertIn("random_num", parsed_data)
|
|
self.assertIsInstance(parsed_data["random_num"], int)
|
|
|
|
def test_get_activity_content_local(self):
|
|
"""Test loading activity content from local file"""
|
|
test_yaml_content = """
|
|
default_max_attempts_per_step: 3
|
|
sections:
|
|
- section_id: "test_section"
|
|
title: "Test Section"
|
|
steps:
|
|
- step_id: "test_step"
|
|
title: "Test Step"
|
|
content_blocks:
|
|
- "Test content"
|
|
"""
|
|
|
|
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
|
|
f.write(test_yaml_content)
|
|
temp_file = f.name
|
|
|
|
try:
|
|
# Create a fake research directory and file
|
|
research_dir = Path("research")
|
|
research_dir.mkdir(exist_ok=True)
|
|
|
|
test_file_path = research_dir / "test_activity.yaml"
|
|
with open(test_file_path, "w") as f:
|
|
f.write(test_yaml_content)
|
|
|
|
# Set LOCAL_ACTIVITIES to True
|
|
with patch.dict(app.app.config, {"LOCAL_ACTIVITIES": True}):
|
|
result = app.get_activity_content("research/test_activity.yaml")
|
|
|
|
self.assertEqual(result["default_max_attempts_per_step"], 3)
|
|
self.assertEqual(len(result["sections"]), 1)
|
|
self.assertEqual(result["sections"][0]["section_id"], "test_section")
|
|
|
|
finally:
|
|
os.unlink(temp_file)
|
|
if test_file_path.exists():
|
|
test_file_path.unlink()
|
|
|
|
def test_get_activity_content_local_security(self):
|
|
"""Test that local file loading prevents path traversal"""
|
|
with patch.dict(app.app.config, {"LOCAL_ACTIVITIES": True}):
|
|
# Test various path traversal attempts
|
|
dangerous_paths = [
|
|
"../etc/passwd",
|
|
"/etc/passwd",
|
|
"research/../../../etc/passwd",
|
|
"research/activity.yaml../../etc/passwd",
|
|
]
|
|
|
|
for path in dangerous_paths:
|
|
with self.assertRaises(ValueError):
|
|
app.get_activity_content(path)
|
|
|
|
def test_get_activity_content_s3(self):
|
|
"""Test loading activity content from S3"""
|
|
test_yaml_content = {
|
|
"default_max_attempts_per_step": 5,
|
|
"sections": [{"section_id": "s3_section", "title": "S3 Section"}],
|
|
}
|
|
|
|
with patch.dict(app.app.config, {"LOCAL_ACTIVITIES": False}):
|
|
with patch.object(
|
|
app, "get_activity_content", return_value=test_yaml_content
|
|
) as mock_func:
|
|
result = app.get_activity_content("path/to/activity.yaml")
|
|
|
|
self.assertEqual(result["default_max_attempts_per_step"], 5)
|
|
self.assertEqual(result["sections"][0]["section_id"], "s3_section")
|
|
mock_func.assert_called_once_with("path/to/activity.yaml")
|
|
|
|
|
|
class TestActivityNavigation(unittest.TestCase):
|
|
"""Test activity navigation functions"""
|
|
|
|
def setUp(self):
|
|
"""Set up test activity content"""
|
|
self.activity_content = {
|
|
"sections": [
|
|
{
|
|
"section_id": "section_1",
|
|
"steps": [
|
|
{"step_id": "step_1", "title": "Step 1"},
|
|
{"step_id": "step_2", "title": "Step 2"},
|
|
{"step_id": "step_3", "title": "Step 3"},
|
|
],
|
|
},
|
|
{
|
|
"section_id": "section_2",
|
|
"steps": [
|
|
{"step_id": "step_1", "title": "Section 2 Step 1"},
|
|
{"step_id": "step_2", "title": "Section 2 Step 2"},
|
|
],
|
|
},
|
|
]
|
|
}
|
|
|
|
def test_get_next_step_within_section(self):
|
|
"""Test getting next step within the same section"""
|
|
next_section, next_step = app.get_next_step(
|
|
self.activity_content, "section_1", "step_1"
|
|
)
|
|
|
|
self.assertEqual(next_section["section_id"], "section_1")
|
|
self.assertEqual(next_step["step_id"], "step_2")
|
|
|
|
def test_get_next_step_across_sections(self):
|
|
"""Test getting next step across sections"""
|
|
next_section, next_step = app.get_next_step(
|
|
self.activity_content, "section_1", "step_3"
|
|
)
|
|
|
|
self.assertEqual(next_section["section_id"], "section_2")
|
|
self.assertEqual(next_step["step_id"], "step_1")
|
|
|
|
def test_get_next_step_at_end(self):
|
|
"""Test getting next step when at the end of activity"""
|
|
next_section, next_step = app.get_next_step(
|
|
self.activity_content, "section_2", "step_2"
|
|
)
|
|
|
|
self.assertIsNone(next_section)
|
|
self.assertIsNone(next_step)
|
|
|
|
def test_get_next_step_invalid_section(self):
|
|
"""Test getting next step with invalid section"""
|
|
next_section, next_step = app.get_next_step(
|
|
self.activity_content, "invalid_section", "step_1"
|
|
)
|
|
|
|
self.assertIsNone(next_section)
|
|
self.assertIsNone(next_step)
|
|
|
|
def test_get_next_step_invalid_step(self):
|
|
"""Test getting next step with invalid step"""
|
|
next_section, next_step = app.get_next_step(
|
|
self.activity_content, "section_1", "invalid_step"
|
|
)
|
|
|
|
self.assertIsNone(next_section)
|
|
self.assertIsNone(next_step)
|
|
|
|
|
|
class TestResponseCategorizationAndFeedback(unittest.TestCase):
|
|
"""Test response categorization and feedback generation"""
|
|
|
|
def test_categorize_response_simple_format(self):
|
|
"""Test response categorization with simple format"""
|
|
with patch.object(
|
|
app, "categorize_response", return_value="correct"
|
|
) as mock_func:
|
|
result = app.categorize_response(
|
|
"What is 2+2?",
|
|
"4",
|
|
["correct", "incorrect"],
|
|
"Categorize as correct or incorrect",
|
|
)
|
|
|
|
self.assertEqual(result, "correct")
|
|
mock_func.assert_called_once_with(
|
|
"What is 2+2?",
|
|
"4",
|
|
["correct", "incorrect"],
|
|
"Categorize as correct or incorrect",
|
|
)
|
|
|
|
def test_categorize_response_analysis_bucket_format(self):
|
|
"""Test response categorization with ANALYSIS/BUCKET format"""
|
|
with patch.object(
|
|
app, "categorize_response", return_value="correct"
|
|
) as mock_func:
|
|
result = app.categorize_response(
|
|
"What is 2+2?",
|
|
"4",
|
|
["correct", "incorrect"],
|
|
"ANALYSIS: Analyze the response. BUCKET: Choose correct or incorrect.",
|
|
)
|
|
|
|
self.assertEqual(result, "correct")
|
|
mock_func.assert_called_once()
|
|
|
|
def test_categorize_response_with_spaces_and_case(self):
|
|
"""Test response categorization handles spaces and case properly"""
|
|
with patch.object(
|
|
app, "categorize_response", return_value="partially_correct"
|
|
) as mock_func:
|
|
result = app.categorize_response(
|
|
"Test question",
|
|
"Test response",
|
|
["partially_correct", "incorrect"],
|
|
"Categorize the response",
|
|
)
|
|
|
|
self.assertEqual(result, "partially_correct")
|
|
mock_func.assert_called_once()
|
|
|
|
def test_generate_ai_feedback(self):
|
|
"""Test AI feedback generation"""
|
|
with patch.object(
|
|
app, "generate_ai_feedback", return_value="Great job! You got it right."
|
|
) as mock_func:
|
|
result = app.generate_ai_feedback(
|
|
"correct",
|
|
"What is 2+2?",
|
|
"4",
|
|
"Provide encouraging feedback",
|
|
"testuser",
|
|
"{}",
|
|
"{}",
|
|
)
|
|
|
|
self.assertEqual(result, "Great job! You got it right.")
|
|
mock_func.assert_called_once()
|
|
|
|
def test_provide_feedback_with_ai_feedback(self):
|
|
"""Test provide_feedback function with AI feedback"""
|
|
transition = {"ai_feedback": {"tokens_for_ai": "Be encouraging"}}
|
|
|
|
with patch.object(
|
|
app, "provide_feedback", return_value="Excellent work!"
|
|
) as mock_func:
|
|
result = app.provide_feedback(
|
|
transition,
|
|
"correct",
|
|
"Test question",
|
|
"Base instructions",
|
|
"Test response",
|
|
"English",
|
|
"testuser",
|
|
"{}",
|
|
"{}",
|
|
)
|
|
|
|
self.assertEqual(result, "Excellent work!")
|
|
mock_func.assert_called_once()
|
|
|
|
def test_provide_feedback_without_ai_feedback(self):
|
|
"""Test provide_feedback function without AI feedback"""
|
|
transition = {}
|
|
|
|
result = app.provide_feedback(
|
|
transition,
|
|
"correct",
|
|
"Test question",
|
|
"Base instructions",
|
|
"Test response",
|
|
"English",
|
|
"testuser",
|
|
"{}",
|
|
"{}",
|
|
)
|
|
|
|
self.assertEqual(result, "")
|
|
|
|
|
|
class TestTranslationAndLanguage(unittest.TestCase):
|
|
"""Test translation and language handling"""
|
|
|
|
def test_translate_text_english_bypass(self):
|
|
"""Test that English text is not translated"""
|
|
text = "Hello, world!"
|
|
result = app.translate_text(text, "English")
|
|
self.assertEqual(result, text)
|
|
|
|
# Test case insensitive
|
|
result = app.translate_text(text, "english")
|
|
self.assertEqual(result, text)
|
|
|
|
# Test with compound language specification
|
|
result = app.translate_text(text, "english please")
|
|
self.assertEqual(result, text)
|
|
|
|
def test_translate_text_other_language(self):
|
|
"""Test translation to other languages"""
|
|
with patch.object(
|
|
app, "translate_text", return_value="Hola, mundo!"
|
|
) as mock_func:
|
|
result = app.translate_text("Hello, world!", "Spanish")
|
|
|
|
self.assertEqual(result, "Hola, mundo!")
|
|
mock_func.assert_called_once_with("Hello, world!", "Spanish")
|
|
|
|
def test_translate_text_error_handling(self):
|
|
"""Test translation error handling"""
|
|
with patch.object(
|
|
app, "translate_text", return_value="Error: Translation failed"
|
|
) as mock_func:
|
|
result = app.translate_text("Hello, world!", "Spanish")
|
|
|
|
self.assertIn("Error:", result)
|
|
mock_func.assert_called_once_with("Hello, world!", "Spanish")
|
|
|
|
|
|
class TestS3Operations(unittest.TestCase):
|
|
"""Test S3 related functions"""
|
|
|
|
def test_get_s3_client_with_profile(self):
|
|
"""Test S3 client creation with profile"""
|
|
mock_client = MagicMock()
|
|
|
|
with patch.object(app, "get_s3_client", return_value=mock_client) as mock_func:
|
|
result = app.get_s3_client()
|
|
|
|
self.assertEqual(result, mock_client)
|
|
mock_func.assert_called_once()
|
|
|
|
def test_get_s3_client_without_profile(self):
|
|
"""Test S3 client creation without profile"""
|
|
mock_client = MagicMock()
|
|
|
|
with patch.object(app, "get_s3_client", return_value=mock_client) as mock_func:
|
|
result = app.get_s3_client()
|
|
|
|
self.assertEqual(result, mock_client)
|
|
mock_func.assert_called_once()
|
|
|
|
def test_find_most_recent_code_block(self):
|
|
"""Test finding most recent code block in messages"""
|
|
# This would require mocking the database and Message model
|
|
# For now, we'll test the logic directly
|
|
test_content = """Here's some code:
|
|
|
|
```python
|
|
def test_function():
|
|
return "Hello, World!"
|
|
```
|
|
|
|
And some more text after.
|
|
"""
|
|
|
|
# Extract the code block manually to test the logic
|
|
lines = test_content.split("\n")
|
|
code_block_lines = []
|
|
code_block_started = False
|
|
|
|
for line in lines:
|
|
if line.startswith("```"):
|
|
if code_block_started:
|
|
break
|
|
else:
|
|
code_block_started = True
|
|
continue
|
|
elif code_block_started:
|
|
code_block_lines.append(line)
|
|
|
|
result = "\n".join(code_block_lines)
|
|
expected = """def test_function():
|
|
return "Hello, World!\""""
|
|
|
|
self.assertEqual(result, expected)
|
|
|
|
|
|
class TestUtilityFunctions(unittest.TestCase):
|
|
"""Test various utility functions"""
|
|
|
|
def test_group_consecutive_roles(self):
|
|
"""Test grouping consecutive roles in messages"""
|
|
messages = [
|
|
{"role": "user", "content": "Hello"},
|
|
{"role": "user", "content": "How are you?"},
|
|
{"role": "assistant", "content": "I'm fine"},
|
|
{"role": "assistant", "content": "Thanks for asking"},
|
|
{"role": "user", "content": "Great!"},
|
|
]
|
|
|
|
result = app.group_consecutive_roles(messages)
|
|
|
|
expected = [
|
|
{"role": "user", "content": "Hello How are you?"},
|
|
{"role": "assistant", "content": "I'm fine Thanks for asking"},
|
|
{"role": "user", "content": "Great!"},
|
|
]
|
|
|
|
self.assertEqual(result, expected)
|
|
|
|
def test_group_consecutive_roles_empty(self):
|
|
"""Test grouping consecutive roles with empty input"""
|
|
result = app.group_consecutive_roles([])
|
|
self.assertEqual(result, [])
|
|
|
|
def test_group_consecutive_roles_single(self):
|
|
"""Test grouping consecutive roles with single message"""
|
|
messages = [{"role": "user", "content": "Hello"}]
|
|
result = app.group_consecutive_roles(messages)
|
|
self.assertEqual(result, messages)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|