From 51b74be7d9db58729217abbd1cb968457b6ac6d3 Mon Sep 17 00:00:00 2001 From: Russell Ballestrini Date: Sun, 10 Aug 2025 17:47:55 -0400 Subject: [PATCH] Fix YAML validator and activity file validation errors - 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. --- activity_yaml_validator.py | 678 +++++++++++------- requirements-test.txt | 2 +- research/activity17-choose-adventure.yaml | 24 +- research/activity28-killer-squares.yaml | 6 + research/activity29-battleship.yaml | 4 +- research/activity29-testship.yaml | 4 +- research/activity3.yaml | 16 + tests/conftest.py | 19 +- tests/functional/test_battleship_game_flow.py | 244 ++++--- tests/integration/test_activity_processing.py | 255 +++---- tests/unit/test_activity_yaml_validator.py | 207 ++++-- tests/unit/test_app.py | 394 +++++----- 12 files changed, 1073 insertions(+), 780 deletions(-) diff --git a/activity_yaml_validator.py b/activity_yaml_validator.py index a2f0ed8..ea3b53c 100644 --- a/activity_yaml_validator.py +++ b/activity_yaml_validator.py @@ -18,13 +18,14 @@ from pathlib import Path class ValidationError(Exception): """Custom exception for validation errors""" + pass class ActivityYAMLValidator: """ Comprehensive validator for activity YAML configurations - + Validates: - YAML syntax and structure - Required fields and schema compliance @@ -33,253 +34,324 @@ class ActivityYAMLValidator: - Battleship-specific rules - Token limits and AI prompt structures """ - + def __init__(self): self.errors = [] self.warnings = [] self.current_file = None - + def validate_file(self, file_path: str) -> Tuple[bool, List[str], List[str]]: """ Validate a YAML file and return results - + Returns: Tuple of (is_valid, errors, warnings) """ self.errors = [] self.warnings = [] self.current_file = file_path - + try: - with open(file_path, 'r', encoding='utf-8') as f: + with open(file_path, "r", encoding="utf-8") as f: content = f.read() - + # Parse YAML try: data = yaml.safe_load(content) except yaml.YAMLError as e: self.errors.append(f"YAML syntax error: {e}") return False, self.errors, self.warnings - + # Validate structure self._validate_structure(data) - + # Validate sections - if 'sections' in data: - self._validate_sections(data['sections']) - + if "sections" in data: + self._validate_sections(data["sections"]) + # Validate universal activity rules self._validate_activity_rules(data) - + # Validate Python code blocks self._validate_python_code(data) - + # Validate logic flow self._validate_logic_flow(data) - + return len(self.errors) == 0, self.errors, self.warnings - + except Exception as e: self.errors.append(f"Unexpected error: {e}") return False, self.errors, self.warnings - + def _validate_structure(self, data: Dict[str, Any]): """Validate basic YAML structure""" if not isinstance(data, dict): self.errors.append("Root level must be a dictionary") return - + # Check required top-level fields - required_fields = ['sections'] + required_fields = ["sections"] for field in required_fields: if field not in data: self.errors.append(f"Missing required field: {field}") - + # Validate optional fields - if 'default_max_attempts_per_step' in data: - if not isinstance(data['default_max_attempts_per_step'], int) or data['default_max_attempts_per_step'] < 1: - self.errors.append("default_max_attempts_per_step must be a positive integer") - - if 'tokens_for_ai_rubric' in data: - if not isinstance(data['tokens_for_ai_rubric'], str): + if "default_max_attempts_per_step" in data: + if ( + not isinstance(data["default_max_attempts_per_step"], int) + or data["default_max_attempts_per_step"] < 1 + ): + self.errors.append( + "default_max_attempts_per_step must be a positive integer" + ) + + if "tokens_for_ai_rubric" in data: + if not isinstance(data["tokens_for_ai_rubric"], str): self.errors.append("tokens_for_ai_rubric must be a string") - + def _validate_sections(self, sections: List[Dict[str, Any]]): """Validate sections structure""" if not isinstance(sections, list): self.errors.append("sections must be a list") return - + if not sections: self.errors.append("At least one section is required") return - + section_ids = set() for i, section in enumerate(sections): if not isinstance(section, dict): self.errors.append(f"Section {i} must be a dictionary") continue - + # Validate section structure self._validate_section(section, i) - + # Check for duplicate section IDs - if 'section_id' in section: - if section['section_id'] in section_ids: + if "section_id" in section: + if section["section_id"] in section_ids: self.errors.append(f"Duplicate section_id: {section['section_id']}") - section_ids.add(section['section_id']) - + section_ids.add(section["section_id"]) + def _validate_section(self, section: Dict[str, Any], section_index: int): """Validate individual section""" - required_fields = ['section_id', 'title', 'steps'] + required_fields = ["section_id", "title", "steps"] for field in required_fields: if field not in section: - self.errors.append(f"Section {section_index}: Missing required field '{field}'") - - if 'steps' in section: - self._validate_steps(section['steps'], section.get('section_id', f'section_{section_index}')) - + self.errors.append( + f"Section {section_index}: Missing required field '{field}'" + ) + + if "steps" in section: + self._validate_steps( + section["steps"], section.get("section_id", f"section_{section_index}") + ) + def _validate_steps(self, steps: List[Dict[str, Any]], section_id: str): """Validate steps within a section""" if not isinstance(steps, list): self.errors.append(f"Section {section_id}: steps must be a list") return - + if not steps: self.errors.append(f"Section {section_id}: At least one step is required") return - + step_ids = set() for i, step in enumerate(steps): if not isinstance(step, dict): - self.errors.append(f"Section {section_id}, step {i}: Must be a dictionary") + self.errors.append( + f"Section {section_id}, step {i}: Must be a dictionary" + ) continue - + self._validate_step(step, section_id, i) - + # Check for duplicate step IDs - if 'step_id' in step: - if step['step_id'] in step_ids: - self.errors.append(f"Section {section_id}: Duplicate step_id '{step['step_id']}'") - step_ids.add(step['step_id']) - + if "step_id" in step: + if step["step_id"] in step_ids: + self.errors.append( + f"Section {section_id}: Duplicate step_id '{step['step_id']}'" + ) + step_ids.add(step["step_id"]) + def _validate_step(self, step: Dict[str, Any], section_id: str, step_index: int): """Validate individual step""" - step_id = step.get('step_id', f'step_{step_index}') - + step_id = step.get("step_id", f"step_{step_index}") + # Required fields - required_fields = ['step_id', 'title'] + required_fields = ["step_id", "title"] for field in required_fields: if field not in step: - self.errors.append(f"Section {section_id}, step {step_id}: Missing required field '{field}'") - + self.errors.append( + f"Section {section_id}, step {step_id}: Missing required field '{field}'" + ) + # Validate content_blocks or question - has_content = 'content_blocks' in step - has_question = 'question' in step - + has_content = "content_blocks" in step + has_question = "question" in step + if not has_content and not has_question: - self.errors.append(f"Section {section_id}, step {step_id}: Must have either 'content_blocks' or 'question'") - + self.errors.append( + f"Section {section_id}, step {step_id}: Must have either 'content_blocks' or 'question'" + ) + if has_content: - self._validate_content_blocks(step['content_blocks'], section_id, step_id) - + self._validate_content_blocks(step["content_blocks"], section_id, step_id) + if has_question: self._validate_question_step(step, section_id, step_id) - - def _validate_content_blocks(self, content_blocks: List[str], section_id: str, step_id: str): + + def _validate_content_blocks( + self, content_blocks: List[str], section_id: str, step_id: str + ): """Validate content blocks""" if not isinstance(content_blocks, list): - self.errors.append(f"Section {section_id}, step {step_id}: content_blocks must be a list") + self.errors.append( + f"Section {section_id}, step {step_id}: content_blocks must be a list" + ) return - + for i, block in enumerate(content_blocks): if not isinstance(block, str): - self.errors.append(f"Section {section_id}, step {step_id}: content_blocks[{i}] must be a string") - - def _validate_question_step(self, step: Dict[str, Any], section_id: str, step_id: str): + self.errors.append( + f"Section {section_id}, step {step_id}: content_blocks[{i}] must be a string" + ) + + def _validate_question_step( + self, step: Dict[str, Any], section_id: str, step_id: str + ): """Validate question-type step""" - if 'question' in step and not isinstance(step['question'], str): - self.errors.append(f"Section {section_id}, step {step_id}: 'question' must be a string") - + if "question" in step and not isinstance(step["question"], str): + self.errors.append( + f"Section {section_id}, step {step_id}: 'question' must be a string" + ) + # Validate AI tokens - if 'tokens_for_ai' in step: - if not isinstance(step['tokens_for_ai'], str): - self.errors.append(f"Section {section_id}, step {step_id}: 'tokens_for_ai' must be a string") - - if 'feedback_tokens_for_ai' in step: - if not isinstance(step['feedback_tokens_for_ai'], str): - self.errors.append(f"Section {section_id}, step {step_id}: 'feedback_tokens_for_ai' must be a string") - + if "tokens_for_ai" in step: + if not isinstance(step["tokens_for_ai"], str): + self.errors.append( + f"Section {section_id}, step {step_id}: 'tokens_for_ai' must be a string" + ) + + if "feedback_tokens_for_ai" in step: + if not isinstance(step["feedback_tokens_for_ai"], str): + self.errors.append( + f"Section {section_id}, step {step_id}: 'feedback_tokens_for_ai' must be a string" + ) + # Validate buckets and transitions - if 'buckets' in step: - self._validate_buckets(step['buckets'], section_id, step_id) - - if 'transitions' in step: - self._validate_transitions(step['transitions'], step.get('buckets', []), section_id, step_id) - + if "buckets" in step: + self._validate_buckets(step["buckets"], section_id, step_id) + + if "transitions" in step: + self._validate_transitions( + step["transitions"], step.get("buckets", []), section_id, step_id + ) + def _validate_buckets(self, buckets: List[str], section_id: str, step_id: str): """Validate buckets list""" if not isinstance(buckets, list): - self.errors.append(f"Section {section_id}, step {step_id}: 'buckets' must be a list") + self.errors.append( + f"Section {section_id}, step {step_id}: 'buckets' must be a list" + ) return - + if not buckets: - self.warnings.append(f"Section {section_id}, step {step_id}: Empty buckets list") + self.warnings.append( + f"Section {section_id}, step {step_id}: Empty buckets list" + ) return - + for i, bucket in enumerate(buckets): - if not isinstance(bucket, str): - self.errors.append(f"Section {section_id}, step {step_id}: buckets[{i}] must be a string") - - def _validate_transitions(self, transitions: Dict[str, Any], buckets: List[str], section_id: str, step_id: str): + if not isinstance(bucket, (str, int, bool)): + self.errors.append( + f"Section {section_id}, step {step_id}: buckets[{i}] must be a string, integer, or boolean" + ) + + def _validate_transitions( + self, + transitions: Dict[str, Any], + buckets: List[Any], + section_id: str, + step_id: str, + ): """Validate transitions dictionary""" if not isinstance(transitions, dict): - self.errors.append(f"Section {section_id}, step {step_id}: 'transitions' must be a dictionary") + self.errors.append( + f"Section {section_id}, step {step_id}: 'transitions' must be a dictionary" + ) return - + # Check that all buckets have corresponding transitions for bucket in buckets: if bucket not in transitions: - self.errors.append(f"Section {section_id}, step {step_id}: Missing transition for bucket '{bucket}'") - + self.errors.append( + f"Section {section_id}, step {step_id}: Missing transition for bucket '{bucket}'" + ) + # Check for unused transitions for transition_key in transitions: if transition_key not in buckets: - self.warnings.append(f"Section {section_id}, step {step_id}: Unused transition '{transition_key}'") - + self.warnings.append( + f"Section {section_id}, step {step_id}: Unused transition '{transition_key}'" + ) + # Validate each transition for bucket, transition in transitions.items(): self._validate_transition(transition, bucket, section_id, step_id) - - def _validate_transition(self, transition: Dict[str, Any], bucket: str, section_id: str, step_id: str): + + def _validate_transition( + self, transition: Dict[str, Any], bucket: str, section_id: str, step_id: str + ): """Validate individual transition""" if not isinstance(transition, dict): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: Transition must be a dictionary") + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: Transition must be a dictionary" + ) return - + # Validate next_section_and_step format - if 'next_section_and_step' in transition: - next_step = transition['next_section_and_step'] + if "next_section_and_step" in transition: + next_step = transition["next_section_and_step"] if not isinstance(next_step, str): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be a string") - elif ':' not in next_step: - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be in format 'section_id:step_id'") - + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be a string" + ) + elif ":" not in next_step: + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: 'next_section_and_step' must be in format 'section_id:step_id'" + ) + # Validate metadata operations - metadata_fields = ['metadata_add', 'metadata_tmp_add', 'metadata_remove', 'metadata_clear', 'metadata_feedback_filter'] + metadata_fields = [ + "metadata_add", + "metadata_tmp_add", + "metadata_remove", + "metadata_clear", + "metadata_feedback_filter", + ] for field in metadata_fields: if field in transition: - if field == 'metadata_clear': + if field == "metadata_clear": if not isinstance(transition[field], bool): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be boolean") - elif field == 'metadata_feedback_filter': + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be boolean" + ) + elif field == "metadata_feedback_filter": if not isinstance(transition[field], list): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a list") + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a list" + ) else: for item in transition[field]: if not isinstance(item, str): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' items must be strings") - elif field == 'metadata_remove': + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' items must be strings" + ) + elif field == "metadata_remove": if isinstance(transition[field], str): # Single key to remove pass @@ -287,39 +359,58 @@ class ActivityYAMLValidator: # List of keys to remove for item in transition[field]: if not isinstance(item, str): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' list items must be strings") + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' list items must be strings" + ) else: - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a string or list of strings") + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a string or list of strings" + ) else: if not isinstance(transition[field], dict): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a dictionary") - + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: '{field}' must be a dictionary" + ) + # Validate other transition fields - if 'run_processing_script' in transition: - if not isinstance(transition['run_processing_script'], bool): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'run_processing_script' must be boolean") - - if 'ai_feedback' in transition: - ai_feedback = transition['ai_feedback'] + if "run_processing_script" in transition: + if not isinstance(transition["run_processing_script"], bool): + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: 'run_processing_script' must be boolean" + ) + + if "ai_feedback" in transition: + ai_feedback = transition["ai_feedback"] if not isinstance(ai_feedback, dict): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'ai_feedback' must be a dictionary") - elif 'tokens_for_ai' in ai_feedback and not isinstance(ai_feedback['tokens_for_ai'], str): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: ai_feedback.tokens_for_ai must be a string") - - if 'content_blocks' in transition: - if not isinstance(transition['content_blocks'], list): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: 'content_blocks' must be a list") + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: 'ai_feedback' must be a dictionary" + ) + elif "tokens_for_ai" in ai_feedback and not isinstance( + ai_feedback["tokens_for_ai"], str + ): + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: ai_feedback.tokens_for_ai must be a string" + ) + + if "content_blocks" in transition: + if not isinstance(transition["content_blocks"], list): + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: 'content_blocks' must be a list" + ) else: - for i, block in enumerate(transition['content_blocks']): + for i, block in enumerate(transition["content_blocks"]): if not isinstance(block, str): - self.errors.append(f"Section {section_id}, step {step_id}, bucket {bucket}: content_blocks[{i}] must be a string") - + self.errors.append( + f"Section {section_id}, step {step_id}, bucket {bucket}: content_blocks[{i}] must be a string" + ) + def _validate_python_code(self, data: Dict[str, Any]): """Validate Python code blocks in scripts""" + def validate_code_block(code: str, location: str): if not code or not isinstance(code, str): return - + try: # Parse the code to check for syntax errors ast.parse(code) @@ -327,263 +418,310 @@ class ActivityYAMLValidator: self.errors.append(f"{location}: Python syntax error - {e}") except Exception as e: self.errors.append(f"{location}: Python parsing error - {e}") - + # Check for common issues self._check_python_code_quality(code, location) - + # Recursively find and validate all Python code blocks self._find_and_validate_scripts(data, validate_code_block) - + def _find_and_validate_scripts(self, obj: Any, validator, path: str = "root"): """Recursively find and validate Python scripts""" if isinstance(obj, dict): for key, value in obj.items(): current_path = f"{path}.{key}" - if key in ['processing_script', 'pre_script'] and isinstance(value, str): + if key in ["processing_script", "pre_script"] and isinstance( + value, str + ): validator(value, current_path) else: self._find_and_validate_scripts(value, validator, current_path) elif isinstance(obj, list): for i, item in enumerate(obj): self._find_and_validate_scripts(item, validator, f"{path}[{i}]") - + def _check_python_code_quality(self, code: str, location: str): """Check Python code for common issues and best practices""" - lines = code.split('\n') - + lines = code.split("\n") + # Check for empty except blocks for i, line in enumerate(lines): stripped = line.strip() - if stripped.startswith('except'): + if stripped.startswith("except"): # Look for the next non-empty line next_line_idx = i + 1 while next_line_idx < len(lines) and not lines[next_line_idx].strip(): next_line_idx += 1 - + if next_line_idx < len(lines): next_line = lines[next_line_idx].strip() - if next_line == 'pass': - self.warnings.append(f"{location} line {i+1}: Empty except block with only 'pass'") - + if next_line == "pass": + self.warnings.append( + f"{location} line {i+1}: Empty except block with only 'pass'" + ) + # Check for potential security issues dangerous_patterns = [ - ('exec(', "Use of exec() can be dangerous"), - ('eval(', "Use of eval() can be dangerous"), - ('__import__(', "Dynamic imports should be used carefully"), + ("exec(", "Use of exec() can be dangerous"), + ("eval(", "Use of eval() can be dangerous"), + ("__import__(", "Dynamic imports should be used carefully"), ] - + for pattern, message in dangerous_patterns: if pattern in code: self.warnings.append(f"{location}: {message}") - + # Check for proper indentation in else blocks for i, line in enumerate(lines): stripped = line.strip() - if stripped == 'else:': + if stripped == "else:": # Check if the next non-empty line exists and is properly indented next_line_idx = i + 1 while next_line_idx < len(lines) and not lines[next_line_idx].strip(): next_line_idx += 1 - + if next_line_idx >= len(lines): - self.errors.append(f"{location} line {i+1}: 'else:' block has no content") + self.errors.append( + f"{location} line {i+1}: 'else:' block has no content" + ) elif next_line_idx < len(lines): next_line = lines[next_line_idx] if not next_line.strip(): continue # Skip empty lines # Check if it's just a comment - if next_line.strip().startswith('#') and next_line_idx + 1 < len(lines): + if next_line.strip().startswith("#") and next_line_idx + 1 < len( + lines + ): following_line_idx = next_line_idx + 1 - while following_line_idx < len(lines) and not lines[following_line_idx].strip(): + while ( + following_line_idx < len(lines) + and not lines[following_line_idx].strip() + ): following_line_idx += 1 - if following_line_idx >= len(lines) or lines[following_line_idx].strip().startswith('#'): - self.errors.append(f"{location} line {i+1}: 'else:' block contains only comments - add 'pass' statement") - + if following_line_idx >= len(lines) or lines[ + following_line_idx + ].strip().startswith("#"): + self.errors.append( + f"{location} line {i+1}: 'else:' block contains only comments - add 'pass' statement" + ) + def _validate_activity_rules(self, data: Dict[str, Any]): """Validate universal activity rules""" - if 'sections' not in data: + if "sections" not in data: return - - # Check that final steps don't have questions - for section in data['sections']: - if 'steps' not in section: + + sections = data["sections"] + + # Find truly terminal steps (last step of last section with no transitions) + for section_idx, section in enumerate(sections): + if "steps" not in section: continue - - steps = section['steps'] + + steps = section["steps"] if not steps: continue - - # Find steps that don't have next transitions (terminal steps) - terminal_steps = [] - for step in steps: - if 'transitions' not in step: - terminal_steps.append(step) - continue - - has_continuing_transition = False - for transition in step['transitions'].values(): - if 'next_section_and_step' in transition: - has_continuing_transition = True - break - - if not has_continuing_transition: - terminal_steps.append(step) - - # Validate terminal steps - for step in terminal_steps: - step_id = step.get('step_id', 'unknown') - section_id = section.get('section_id', 'unknown') - - if 'question' in step: - self.errors.append(f"Section {section_id}, step {step_id}: Final/terminal steps cannot have questions") - - if 'buckets' in step and step['buckets']: - self.errors.append(f"Section {section_id}, step {step_id}: Final/terminal steps should not have buckets") - + + # Check if this is the last section + is_last_section = section_idx == len(sections) - 1 + + for step_idx, step in enumerate(steps): + step_id = step.get("step_id", "unknown") + section_id = section.get("section_id", "unknown") + + # Check if this is the last step in the section + is_last_step_in_section = step_idx == len(steps) - 1 + + # A step is truly terminal only if: + # 1. It's the last step of the last section AND has no transitions with next_section_and_step + # OR + # 2. All its transitions explicitly end the activity (no next_section_and_step anywhere) + is_terminal = False + + if "transitions" in step: + # Check if any transition continues the flow + has_continuing_transition = False + for transition in step["transitions"].values(): + if "next_section_and_step" in transition: + has_continuing_transition = True + break + + # If this is the last step of the last section and has no continuing transitions + if ( + is_last_section + and is_last_step_in_section + and not has_continuing_transition + ): + is_terminal = True + elif is_last_section and is_last_step_in_section: + # No transitions at all and it's the last step of the last section + is_terminal = True + + # Only validate true terminal steps + if is_terminal: + if "question" in step: + self.errors.append( + f"Section {section_id}, step {step_id}: Final/terminal steps cannot have questions" + ) + + if "buckets" in step and step["buckets"]: + self.errors.append( + f"Section {section_id}, step {step_id}: Final/terminal steps should not have buckets" + ) + # Validate metadata_feedback_filter usage self._validate_metadata_filters(data) - + # Validate pre_script usage self._validate_pre_scripts(data) - + def _validate_metadata_filters(self, data: Dict[str, Any]): """Validate metadata_feedback_filter usage""" - if 'sections' not in data: + if "sections" not in data: return - - for section in data['sections']: - if 'steps' not in section: + + for section in data["sections"]: + if "steps" not in section: continue - - section_id = section.get('section_id', 'unknown') - for step in section['steps']: - step_id = step.get('step_id', 'unknown') - if 'transitions' not in step: + + section_id = section.get("section_id", "unknown") + for step in section["steps"]: + step_id = step.get("step_id", "unknown") + if "transitions" not in step: continue - - for bucket, transition in step['transitions'].items(): - if 'metadata_feedback_filter' in transition: + + for bucket, transition in step["transitions"].items(): + if "metadata_feedback_filter" in transition: # Check if step has feedback_tokens_for_ai - if 'feedback_tokens_for_ai' not in step: - self.warnings.append(f"Section {section_id}, step {step_id}: metadata_feedback_filter used but no feedback_tokens_for_ai defined") - + if "feedback_tokens_for_ai" not in step: + self.warnings.append( + f"Section {section_id}, step {step_id}: metadata_feedback_filter used but no feedback_tokens_for_ai defined" + ) + def _validate_pre_scripts(self, data: Dict[str, Any]): """Validate pre_script usage""" - if 'sections' not in data: + if "sections" not in data: return - - for section in data['sections']: - if 'steps' not in section: + + for section in data["sections"]: + if "steps" not in section: continue - - section_id = section.get('section_id', 'unknown') - for step in section['steps']: - step_id = step.get('step_id', 'unknown') - - if 'pre_script' in step: + + section_id = section.get("section_id", "unknown") + for step in section["steps"]: + step_id = step.get("step_id", "unknown") + + if "pre_script" in step: # Check if step has a question (pre_script should be used with questions) - if 'question' not in step: - self.warnings.append(f"Section {section_id}, step {step_id}: pre_script typically used with question steps") - + if "question" not in step: + self.warnings.append( + f"Section {section_id}, step {step_id}: pre_script typically used with question steps" + ) + # Validate pre_script is a string - if not isinstance(step['pre_script'], str): - self.errors.append(f"Section {section_id}, step {step_id}: pre_script must be a string") - + if not isinstance(step["pre_script"], str): + self.errors.append( + f"Section {section_id}, step {step_id}: pre_script must be a string" + ) + def _validate_logic_flow(self, data: Dict[str, Any]): """Validate logical flow and transitions between steps""" - if 'sections' not in data: + if "sections" not in data: return - + # Build a map of all available steps all_steps = {} - for section in data['sections']: - section_id = section.get('section_id') - if not section_id or 'steps' not in section: + for section in data["sections"]: + section_id = section.get("section_id") + if not section_id or "steps" not in section: continue - - for step in section['steps']: - step_id = step.get('step_id') + + for step in section["steps"]: + step_id = step.get("step_id") if step_id: all_steps[f"{section_id}:{step_id}"] = step - + # Validate all transition targets - for section in data['sections']: - section_id = section.get('section_id') - if not section_id or 'steps' not in section: + for section in data["sections"]: + section_id = section.get("section_id") + if not section_id or "steps" not in section: continue - - for step in section['steps']: - step_id = step.get('step_id') - if not step_id or 'transitions' not in step: + + for step in section["steps"]: + step_id = step.get("step_id") + if not step_id or "transitions" not in step: continue - - for bucket, transition in step['transitions'].items(): - if 'next_section_and_step' in transition: - target = transition['next_section_and_step'] + + for bucket, transition in step["transitions"].items(): + if "next_section_and_step" in transition: + target = transition["next_section_and_step"] if target not in all_steps: - self.errors.append(f"Section {section_id}, step {step_id}: Invalid transition target '{target}'") + self.errors.append( + f"Section {section_id}, step {step_id}: Invalid transition target '{target}'" + ) def main(): """Command line interface for the validator""" - parser = argparse.ArgumentParser(description='Validate activity YAML files') - parser.add_argument('files', nargs='+', help='YAML files to validate') - parser.add_argument('--strict', action='store_true', help='Treat warnings as errors') - parser.add_argument('--quiet', action='store_true', help='Only show errors') - + parser = argparse.ArgumentParser(description="Validate activity YAML files") + parser.add_argument("files", nargs="+", help="YAML files to validate") + parser.add_argument( + "--strict", action="store_true", help="Treat warnings as errors" + ) + parser.add_argument("--quiet", action="store_true", help="Only show errors") + args = parser.parse_args() - + validator = ActivityYAMLValidator() total_errors = 0 total_warnings = 0 - + for file_path in args.files: if not Path(file_path).exists(): print(f"āŒ File not found: {file_path}") total_errors += 1 continue - + if not args.quiet: print(f"\nšŸ“„ Validating: {file_path}") print("=" * 50) - + is_valid, errors, warnings = validator.validate_file(file_path) - + if errors: print(f"āŒ {len(errors)} error(s):") for error in errors: print(f" ā€¢ {error}") total_errors += len(errors) - + if warnings and not args.quiet: print(f"āš ļø {len(warnings)} warning(s):") for warning in warnings: print(f" ā€¢ {warning}") total_warnings += len(warnings) - + if is_valid and not warnings: print(f"āœ… {file_path} is valid!") elif is_valid: print(f"āœ… {file_path} is valid (with warnings)") else: print(f"āŒ {file_path} has errors") - + # Summary if not args.quiet: print(f"\nšŸ“Š Summary:") print(f" Files checked: {len(args.files)}") print(f" Errors: {total_errors}") print(f" Warnings: {total_warnings}") - + # Exit code exit_code = 0 if total_errors > 0: exit_code = 1 elif args.strict and total_warnings > 0: exit_code = 1 - + sys.exit(exit_code) -if __name__ == '__main__': - main() \ No newline at end of file +if __name__ == "__main__": + main() diff --git a/requirements-test.txt b/requirements-test.txt index 16b5181..ca2c3b8 100644 --- a/requirements-test.txt +++ b/requirements-test.txt @@ -3,4 +3,4 @@ pytest-cov pytest-mock pytest-flask pytest-asyncio -together \ No newline at end of file +black \ No newline at end of file diff --git a/research/activity17-choose-adventure.yaml b/research/activity17-choose-adventure.yaml index 078b410..cc7342d 100644 --- a/research/activity17-choose-adventure.yaml +++ b/research/activity17-choose-adventure.yaml @@ -86,7 +86,7 @@ sections: metadata_add: mystical_amulet: true metadata_remove: - golden_keychain: true + - golden_keychain offer_mysterious_amulet: metadata_conditions: mysterious_amulet: true @@ -97,7 +97,7 @@ sections: metadata_add: rare_gemstone: true metadata_remove: - mysterious_amulet: true + - mysterious_amulet offer_rare_gemstone: metadata_conditions: rare_gemstone: true @@ -108,7 +108,7 @@ sections: metadata_add: ancient_scroll: true metadata_remove: - rare_gemstone: true + - rare_gemstone offer_ancient_scroll: metadata_conditions: ancient_scroll: true @@ -119,7 +119,7 @@ sections: metadata_add: magical_wand: true metadata_remove: - ancient_scroll: true + - ancient_scroll offer_magical_wand: metadata_conditions: magical_wand: true @@ -130,7 +130,7 @@ sections: metadata_add: treasure_map: true metadata_remove: - magical_wand: true + - magical_wand offer_treasure_map: metadata_conditions: treasure_map: true @@ -141,7 +141,7 @@ sections: metadata_add: silver_coin: true metadata_remove: - treasure_map: true + - treasure_map offer_silver_coin: metadata_conditions: silver_coin: true @@ -152,7 +152,7 @@ sections: metadata_add: mystical_ring: true metadata_remove: - silver_coin: true + - silver_coin offer_mystical_ring: metadata_conditions: mystical_ring: true @@ -163,7 +163,7 @@ sections: metadata_add: rare_book: true metadata_remove: - mystical_ring: true + - mystical_ring offer_rare_book: metadata_conditions: rare_book: true @@ -174,7 +174,7 @@ sections: metadata_add: magical_potion: true metadata_remove: - rare_book: true + - rare_book offer_magical_potion: metadata_conditions: magical_potion: true @@ -185,12 +185,12 @@ sections: metadata_add: golden_keychain: true metadata_remove: - magical_potion: true + - magical_potion offer_shadow_charm: metadata_conditions: shadow_charm: true metadata_remove: - shadow_charm: true + - shadow_charm content_blocks: - "You offered the Shadow Charm to the god. šŸ–¤" - "The god summons the Shadow Beast! Prepare for battle!" @@ -199,7 +199,7 @@ sections: metadata_conditions: flame_charm: true metadata_remove: - flame_charm: true + - flame_charm content_blocks: - "You offered the Flame Charm to the god. šŸ”„" - "The god summons the Fire Drake! Prepare for battle!" diff --git a/research/activity28-killer-squares.yaml b/research/activity28-killer-squares.yaml index 69c4fae..df50233 100644 --- a/research/activity28-killer-squares.yaml +++ b/research/activity28-killer-squares.yaml @@ -84,6 +84,11 @@ sections: next_section_and_step: "section_1:step_1" exit: next_section_and_step: "section_1:step_3" + restart: + ai_feedback: + tokens_for_ai: "Restarting the game. Let's start fresh!" + metadata_clear: True + next_section_and_step: "section_1:step_0" - step_id: "step_2" title: "Kill a Square" @@ -242,6 +247,7 @@ sections: - valid_move - invalid_move - exit + - restart transitions: valid_move: run_processing_script: True diff --git a/research/activity29-battleship.yaml b/research/activity29-battleship.yaml index ac1c628..997275d 100644 --- a/research/activity29-battleship.yaml +++ b/research/activity29-battleship.yaml @@ -864,14 +864,12 @@ sections: user_shot: "the-users-response" next_section_and_step: "section_1:step_2" exit: - next_section_and_step: "section_1:step_3" + next_section_and_step: "section_1:step_4" restart: content_blocks: - "Restarting the game. Let's start fresh!" metadata_clear: True next_section_and_step: "section_1:step_0" - game_end: - next_section_and_step: "section_1:step_3" - step_id: "step_3" title: "Game Over" diff --git a/research/activity29-testship.yaml b/research/activity29-testship.yaml index aacb207..ec6d0a0 100644 --- a/research/activity29-testship.yaml +++ b/research/activity29-testship.yaml @@ -832,14 +832,12 @@ sections: user_shot: "the-users-response" next_section_and_step: "section_1:step_2" exit: - next_section_and_step: "section_1:step_3" + next_section_and_step: "section_1:step_4" restart: content_blocks: - "Restarting the game. Let's start fresh!" metadata_clear: True next_section_and_step: "section_1:step_0" - game_end: - next_section_and_step: "section_1:step_3" - step_id: "step_3" title: "Game Over" diff --git a/research/activity3.yaml b/research/activity3.yaml index 6b51c47..2fc6206 100644 --- a/research/activity3.yaml +++ b/research/activity3.yaml @@ -284,3 +284,19 @@ sections: - "I see you have some questions. Let's answer them." ai_feedback: tokens_for_ai: "Answer the child's clarifying questions and provide additional information in a friendly and engaging manner." + + - section_id: "section_5" + title: "Congratulations!" + steps: + - step_id: "step_1" + title: "You're an Elephant Expert!" + content_blocks: + - "šŸŽ‰ Congratulations! You've learned so much about elephants today!" + - "You now know:" + - "āœ… What elephants look like and how big they are" + - "āœ… What elephants eat with their trunks" + - "āœ… How elephants communicate with each other" + - "āœ… Why elephants need our help" + - "āœ… Ways we can help protect elephants" + - "You're now an elephant expert! Keep learning and caring about animals! šŸ˜šŸŒŸ" + - "Thank you for taking this journey with us!" diff --git a/tests/conftest.py b/tests/conftest.py index aee24a8..f904fe2 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,20 +11,22 @@ from unittest.mock import patch, MagicMock # Set up test environment variables immediately at import time TEST_ENV_VARS = { - 'MODEL_ENDPOINT_1': 'https://test.api', - 'MODEL_NAME_1': 'test-model', - 'MODEL_KEY_1': 'test-key' + "MODEL_ENDPOINT_1": "https://test.api", + "MODEL_NAME_1": "test-model", + "MODEL_KEY_1": "test-key", } # Apply environment variables immediately for import os.environ.update(TEST_ENV_VARS) -@pytest.fixture(scope='session', autouse=True) + +@pytest.fixture(scope="session", autouse=True) def setup_test_environment(): """Set up test environment variables for all tests""" with patch.dict(os.environ, TEST_ENV_VARS): yield + @pytest.fixture def mock_openai_client(): """Mock OpenAI client for testing""" @@ -34,13 +36,12 @@ def mock_openai_client(): mock_client.chat.completions.create.return_value = mock_response return mock_client + @pytest.fixture def mock_s3_client(): """Mock S3 client for testing""" mock_client = MagicMock() - mock_response = { - 'Body': MagicMock() - } - mock_response['Body'].read.return_value.decode.return_value = "test: content" + mock_response = {"Body": MagicMock()} + mock_response["Body"].read.return_value.decode.return_value = "test: content" mock_client.get_object.return_value = mock_response - return mock_client \ No newline at end of file + return mock_client diff --git a/tests/functional/test_battleship_game_flow.py b/tests/functional/test_battleship_game_flow.py index 913f875..1c4153e 100644 --- a/tests/functional/test_battleship_game_flow.py +++ b/tests/functional/test_battleship_game_flow.py @@ -17,22 +17,25 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) # Mock external dependencies -with patch.dict('sys.modules', { - 'gevent': MagicMock(), - 'flask_socketio': MagicMock(), - 'boto3': MagicMock(), - 'openai': MagicMock(), - 'together': MagicMock(), - 'models': MagicMock(), - 'matplotlib': MagicMock(), - 'matplotlib.pyplot': MagicMock(), -}): +with patch.dict( + "sys.modules", + { + "gevent": MagicMock(), + "flask_socketio": MagicMock(), + "boto3": MagicMock(), + "openai": MagicMock(), + "together": MagicMock(), + "models": MagicMock(), + "matplotlib": MagicMock(), + "matplotlib.pyplot": MagicMock(), + }, +): import app class MockBattleshipState: """Mock battleship activity state for testing""" - + def __init__(self): self.section_id = "section_1" self.step_id = "step_2" # Game step @@ -41,26 +44,28 @@ class MockBattleshipState: self.dict_metadata = {} self.json_metadata = "{}" self.s3_file_path = "activity29-battleship.yaml" - + # Initialize with typical battleship metadata - self.dict_metadata.update({ - "ai_mode": "random", - "user_shots": [], - "ai_shots": [], - "user_hits": [], - "ai_hits": [], - "game_over": False, - "user_wins": False, - "ai_wins": False, - "user_sunk_ships": [], - "ai_sunk_ships": [] - }) + self.dict_metadata.update( + { + "ai_mode": "random", + "user_shots": [], + "ai_shots": [], + "user_hits": [], + "ai_hits": [], + "game_over": False, + "user_wins": False, + "ai_wins": False, + "user_sunk_ships": [], + "ai_sunk_ships": [], + } + ) self.json_metadata = json.dumps(self.dict_metadata) - + def add_metadata(self, key, value): self.dict_metadata[key] = value self.json_metadata = json.dumps(self.dict_metadata) - + def remove_metadata(self, key): if key in self.dict_metadata: del self.dict_metadata[key] @@ -69,26 +74,26 @@ class MockBattleshipState: class TestBattleshipGameFlow(unittest.TestCase): """Test complete battleship game scenarios""" - + def setUp(self): """Set up battleship test fixtures""" # Sample board with ships placed self.user_board = [-1] * 100 # Empty board - self.ai_board = [-1] * 100 # Empty board - + self.ai_board = [-1] * 100 # Empty board + # Place a destroyer (size 2) at positions 0, 1 self.ai_board[0] = "Destroyer" self.ai_board[1] = "Destroyer" - + # Place a cruiser (size 3) at positions 10, 20, 30 (vertical) self.user_board[10] = "Cruiser" - self.user_board[20] = "Cruiser" + self.user_board[20] = "Cruiser" self.user_board[30] = "Cruiser" - + self.battleship_state = MockBattleshipState() self.battleship_state.add_metadata("user_board", self.user_board) self.battleship_state.add_metadata("ai_board", self.ai_board) - + def test_battleship_setup_and_board_generation(self): """Test battleship game setup and board generation""" setup_script = """ @@ -139,51 +144,50 @@ script_result = { } } """ - + # Mock the script execution since it involves complex ship placement - mock_metadata = { - "user_board": [-1] * 100, - "ai_board": [-1] * 100 - } - + mock_metadata = {"user_board": [-1] * 100, "ai_board": [-1] * 100} + # Place some ships for testing mock_metadata["user_board"][0:5] = ["Carrier"] * 5 # Carrier mock_metadata["user_board"][10:14] = ["Battleship"] * 4 # Battleship mock_metadata["user_board"][20:23] = ["Cruiser"] * 3 # Cruiser mock_metadata["user_board"][30:33] = ["Submarine"] * 3 # Submarine mock_metadata["user_board"][40:42] = ["Destroyer"] * 2 # Destroyer - + mock_metadata["ai_board"][50:55] = ["Carrier"] * 5 # Carrier mock_metadata["ai_board"][60:64] = ["Battleship"] * 4 # Battleship mock_metadata["ai_board"][70:73] = ["Cruiser"] * 3 # Cruiser mock_metadata["ai_board"][80:83] = ["Submarine"] * 3 # Submarine mock_metadata["ai_board"][90:92] = ["Destroyer"] * 2 # Destroyer - - with patch.object(app, 'execute_processing_script', return_value={"metadata": mock_metadata}) as mock_exec: + + with patch.object( + app, "execute_processing_script", return_value={"metadata": mock_metadata} + ) as mock_exec: metadata = {} result = app.execute_processing_script(metadata, setup_script) - + # Verify boards were created self.assertIn("user_board", result["metadata"]) self.assertIn("ai_board", result["metadata"]) - + user_board = result["metadata"]["user_board"] ai_board = result["metadata"]["ai_board"] - + # Verify boards are correct size self.assertEqual(len(user_board), 100) self.assertEqual(len(ai_board), 100) - + # Count ship cells user_ship_cells = sum(1 for cell in user_board if cell != -1) ai_ship_cells = sum(1 for cell in ai_board if cell != -1) - + # Should have exactly 17 ship cells (5+4+3+3+2) self.assertEqual(user_ship_cells, 17) self.assertEqual(ai_ship_cells, 17) - + mock_exec.assert_called_once() - + def test_battleship_shot_processing(self): """Test processing a shot in battleship""" shot_script = """ @@ -226,7 +230,7 @@ if 0 <= user_shot < 100 and user_shot not in user_shots: } } """ - + # Set up metadata for the shot metadata = { "user_shot": "0", # Hit the destroyer @@ -235,24 +239,24 @@ if 0 <= user_shot < 100 and user_shot not in user_shots: "user_shots": [], "ai_shots": [], "user_hits": [], - "ai_hits": [] + "ai_hits": [], } - + result = app.execute_processing_script(metadata, shot_script) - + # Verify shot was processed self.assertIn("user_shots", result["metadata"]) self.assertIn("user_hit_result", result["metadata"]) self.assertIn("ai_shot", result["metadata"]) - + # Verify user hit the destroyer self.assertEqual(result["metadata"]["user_hit_result"], "hit") self.assertIn(0, result["metadata"]["user_hits"]) - + # Verify AI took a shot self.assertIsInstance(result["metadata"]["ai_shot"], int) self.assertIn(result["metadata"]["ai_shot"], result["metadata"]["ai_shots"]) - + def test_battleship_ship_sinking_logic(self): """Test ship sinking detection""" sinking_script = """ @@ -306,39 +310,43 @@ script_result = { } } """ - + # Set up metadata where destroyer is completely hit metadata = { "user_board": self.user_board, "ai_board": self.ai_board, "user_hits": [0, 1], # Both destroyer positions - "ai_hits": [10], # One cruiser position + "ai_hits": [10], # One cruiser position "user_sunk_ships": [], - "ai_sunk_ships": [] + "ai_sunk_ships": [], } - + mock_result = { "metadata": { "user_sunk_ships": ["Destroyer"], "ai_sunk_ships": [], "user_sunk_ship_this_round": "Destroyer", - "ai_sunk_ship_this_round": None + "ai_sunk_ship_this_round": None, } } - - with patch.object(app, 'execute_processing_script', return_value=mock_result) as mock_exec: + + with patch.object( + app, "execute_processing_script", return_value=mock_result + ) as mock_exec: result = app.execute_processing_script(metadata, sinking_script) - + # Verify destroyer was sunk self.assertIn("Destroyer", result["metadata"]["user_sunk_ships"]) - self.assertEqual(result["metadata"]["user_sunk_ship_this_round"], "Destroyer") - + self.assertEqual( + result["metadata"]["user_sunk_ship_this_round"], "Destroyer" + ) + # Verify cruiser was not sunk (only 1 of 3 positions hit) self.assertNotIn("Cruiser", result["metadata"]["ai_sunk_ships"]) self.assertIsNone(result["metadata"]["ai_sunk_ship_this_round"]) - + mock_exec.assert_called_once() - + def test_battleship_win_condition(self): """Test win condition detection""" win_script = """ @@ -380,35 +388,35 @@ script_result = { } } """ - + # Test user wins scenario metadata_user_wins = { "user_board": self.user_board, "ai_board": self.ai_board, "user_hits": [0, 1], # Hit all AI ships (only destroyer) - "ai_hits": [10] # Partial hit on user ships + "ai_hits": [10], # Partial hit on user ships } - + result = app.execute_processing_script(metadata_user_wins, win_script) - + self.assertTrue(result["metadata"]["game_over"]) self.assertTrue(result["metadata"]["user_wins"]) self.assertFalse(result["metadata"]["ai_wins"]) - + # Test AI wins scenario metadata_ai_wins = { "user_board": self.user_board, "ai_board": self.ai_board, - "user_hits": [0], # Partial hit on AI ships - "ai_hits": [10, 20, 30] # Hit all user ships (complete cruiser) + "user_hits": [0], # Partial hit on AI ships + "ai_hits": [10, 20, 30], # Hit all user ships (complete cruiser) } - + result = app.execute_processing_script(metadata_ai_wins, win_script) - + self.assertTrue(result["metadata"]["game_over"]) self.assertFalse(result["metadata"]["user_wins"]) self.assertTrue(result["metadata"]["ai_wins"]) - + def test_battleship_ai_modes(self): """Test different AI difficulty modes""" # Test random AI mode @@ -431,15 +439,15 @@ script_result = { } } """ - + metadata = {"ai_shots": [0, 1, 2, 3, 4]} - - with patch('random.choice', return_value=50): # Mock random choice + + with patch("random.choice", return_value=50): # Mock random choice result = app.execute_processing_script(metadata, random_ai_script) - + self.assertEqual(result["metadata"]["ai_shot"], 50) self.assertEqual(result["metadata"]["ai_mode"], "random") - + # Test hunter AI mode hunter_ai_script = """ ai_mode = "hunter" @@ -480,20 +488,24 @@ script_result = { } } """ - + # Test hunter mode with a hit metadata_with_hit = { "ai_shots": [45, 46], - "ai_hits": [45] # Hit at position 45 + "ai_hits": [45], # Hit at position 45 } - + result = app.execute_processing_script(metadata_with_hit, hunter_ai_script) - + # Should target adjacent to the hit (35, 55, 44, or 46, but 46 already shot) - expected_targets = [35, 55, 44] # Adjacent to 45, excluding already shot positions + expected_targets = [ + 35, + 55, + 44, + ] # Adjacent to 45, excluding already shot positions self.assertIn(result["metadata"]["ai_shot"], expected_targets) self.assertEqual(result["metadata"]["ai_mode"], "hunter") - + def test_battleship_game_state_validation(self): """Test battleship game state validation""" validation_script = """ @@ -533,41 +545,41 @@ script_result = { } } """ - + # Test valid state valid_metadata = { "user_shots": [0, 1, 2], "ai_shots": [10, 20, 30], "user_hits": [0, 1], - "ai_hits": [10] + "ai_hits": [10], } - + result = app.execute_processing_script(valid_metadata, validation_script) - + self.assertTrue(result["metadata"]["is_valid_state"]) self.assertEqual(len(result["metadata"]["validation_errors"]), 0) - + # Test invalid state invalid_metadata = { "user_shots": [0, 1], "ai_shots": [10, 20, 105], # Out of bounds shot - "user_hits": [0, 1, 2], # Hit not in shots - "ai_hits": [10] + "user_hits": [0, 1, 2], # Hit not in shots + "ai_hits": [10], } - + result = app.execute_processing_script(invalid_metadata, validation_script) - + self.assertFalse(result["metadata"]["is_valid_state"]) self.assertGreater(len(result["metadata"]["validation_errors"]), 0) class TestBattleshipEdgeCases(unittest.TestCase): """Test battleship edge cases and error handling""" - + def test_invalid_shot_handling(self): """Test handling of invalid shots""" invalid_shots = [-1, 100, 999, "invalid", None] - + for invalid_shot in invalid_shots: validation_script = f""" user_shot_input = {repr(invalid_shot)} @@ -586,10 +598,10 @@ script_result = {{ }} }} """ - + result = app.execute_processing_script({}, validation_script) self.assertFalse(result["metadata"]["is_valid_shot"]) - + def test_duplicate_shot_handling(self): """Test handling of duplicate shots""" duplicate_shot_script = """ @@ -607,20 +619,20 @@ script_result = { } } """ - + # First shot - should not be duplicate metadata = {"user_shots": [1, 2, 3]} result = app.execute_processing_script(metadata, duplicate_shot_script) - + self.assertFalse(result["metadata"]["is_duplicate"]) self.assertIn(42, result["metadata"]["user_shots"]) - + # Second shot - should be duplicate metadata = {"user_shots": [1, 2, 3, 42]} result = app.execute_processing_script(metadata, duplicate_shot_script) - + self.assertTrue(result["metadata"]["is_duplicate"]) - + def test_game_end_edge_cases(self): """Test edge cases in game ending""" # Test simultaneous win condition (both players hit all ships in same turn) @@ -654,26 +666,28 @@ script_result = { } } """ - + mock_result = { "metadata": { "game_over": True, "user_wins": True, "ai_wins": False, "all_ai_ships_hit": True, - "all_user_ships_hit": True + "all_user_ships_hit": True, } } - - with patch.object(app, 'execute_processing_script', return_value=mock_result) as mock_exec: + + with patch.object( + app, "execute_processing_script", return_value=mock_result + ) as mock_exec: result = app.execute_processing_script({}, simultaneous_win_script) - + self.assertTrue(result["metadata"]["game_over"]) self.assertTrue(result["metadata"]["user_wins"]) self.assertFalse(result["metadata"]["ai_wins"]) - + mock_exec.assert_called_once() -if __name__ == '__main__': - unittest.main(verbosity=2) \ No newline at end of file +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/integration/test_activity_processing.py b/tests/integration/test_activity_processing.py index 70b9ee6..e263742 100644 --- a/tests/integration/test_activity_processing.py +++ b/tests/integration/test_activity_processing.py @@ -18,20 +18,23 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent.parent)) # Mock external dependencies before importing -with patch.dict('sys.modules', { - 'gevent': MagicMock(), - 'flask_socketio': MagicMock(), - 'boto3': MagicMock(), - 'openai': MagicMock(), - 'together': MagicMock(), - 'models': MagicMock(), -}): +with patch.dict( + "sys.modules", + { + "gevent": MagicMock(), + "flask_socketio": MagicMock(), + "boto3": MagicMock(), + "openai": MagicMock(), + "together": MagicMock(), + "models": MagicMock(), + }, +): import app class MockActivityState: """Mock ActivityState for testing""" - + def __init__(self, section_id="test_section", step_id="test_step"): self.section_id = section_id self.step_id = step_id @@ -40,16 +43,16 @@ class MockActivityState: self.dict_metadata = {} self.json_metadata = "{}" self.s3_file_path = "test_activity.yaml" - + def add_metadata(self, key, value): self.dict_metadata[key] = value self.json_metadata = json.dumps(self.dict_metadata) - + def remove_metadata(self, key): if key in self.dict_metadata: del self.dict_metadata[key] self.json_metadata = json.dumps(self.dict_metadata) - + def clear_metadata(self): self.dict_metadata = {} self.json_metadata = "{}" @@ -57,7 +60,7 @@ class MockActivityState: class TestActivityProcessingIntegration(unittest.TestCase): """Integration tests for complete activity processing""" - + def setUp(self): """Set up test fixtures""" self.test_activity = { @@ -78,35 +81,35 @@ class TestActivityProcessingIntegration(unittest.TestCase): "correct": { "content_blocks": ["Great job!"], "metadata_add": {"score": "n+1"}, - "next_section_and_step": "section_1:step_2" + "next_section_and_step": "section_1:step_2", }, "incorrect": { "content_blocks": ["Try again!"], - "counts_as_attempt": True - } - } + "counts_as_attempt": True, + }, + }, }, { - "step_id": "step_2", + "step_id": "step_2", "title": "Final Step", - "content_blocks": ["Activity completed!"] - } - ] + "content_blocks": ["Activity completed!"], + }, + ], } - ] + ], } - + def test_complete_activity_flow_correct_answer(self): """Test complete activity flow with correct answer""" activity_state = MockActivityState("section_1", "step_1") activity_state.add_metadata("score", 0) - + # Mock the categorization to return "correct" # Simulate the core logic without external dependencies section = self.test_activity["sections"][0] step = section["steps"][0] transition = step["transitions"]["correct"] - + # Test metadata operations if "metadata_add" in transition: for key, value in transition["metadata_add"].items(): @@ -114,29 +117,29 @@ class TestActivityProcessingIntegration(unittest.TestCase): c = int(value[2:]) new_value = activity_state.dict_metadata.get(key, 0) + c activity_state.add_metadata(key, new_value) - + # Verify state after processing self.assertEqual(activity_state.dict_metadata["score"], 1) - + def test_complete_activity_flow_incorrect_answer(self): """Test complete activity flow with incorrect answer""" activity_state = MockActivityState("section_1", "step_1") - - section = self.test_activity["sections"][0] + + section = self.test_activity["sections"][0] step = section["steps"][0] transition = step["transitions"]["incorrect"] - + # Test that attempts increment for incorrect answers if transition.get("counts_as_attempt", True): activity_state.attempts += 1 - + self.assertEqual(activity_state.attempts, 1) - + def test_processing_script_execution_integration(self): """Test processing script execution with metadata updates""" script_step = { "step_id": "script_step", - "title": "Script Step", + "title": "Script Step", "question": "Test question", "processing_script": """ import random @@ -162,37 +165,36 @@ script_result = { "transitions": { "continue": { "run_processing_script": True, - "next_section_and_step": "section_1:step_2" + "next_section_and_step": "section_1:step_2", } - } + }, } - + activity_state = MockActivityState() activity_state.add_metadata("score", 25) - + transition = script_step["transitions"]["continue"] - + # Execute the processing script if transition.get("run_processing_script", False): result = app.execute_processing_script( - activity_state.dict_metadata, - script_step["processing_script"] + activity_state.dict_metadata, script_step["processing_script"] ) - + # Update metadata with results for key, value in result.get("metadata", {}).items(): activity_state.add_metadata(key, value) - + # Verify the script executed correctly - self.assertIn('generated_number', activity_state.dict_metadata) - self.assertIn('bonus', activity_state.dict_metadata) - self.assertTrue(activity_state.dict_metadata['processing_complete']) - self.assertIn('final_score', activity_state.dict_metadata) - + self.assertIn("generated_number", activity_state.dict_metadata) + self.assertIn("bonus", activity_state.dict_metadata) + self.assertTrue(activity_state.dict_metadata["processing_complete"]) + self.assertIn("final_score", activity_state.dict_metadata) + # Verify calculation - expected_score = 25 + activity_state.dict_metadata['bonus'] - self.assertEqual(activity_state.dict_metadata['final_score'], expected_score) - + expected_score = 25 + activity_state.dict_metadata["bonus"] + self.assertEqual(activity_state.dict_metadata["final_score"], expected_score) + def test_pre_script_execution_integration(self): """Test pre-script execution with user response""" pre_script_step = { @@ -221,81 +223,84 @@ script_result = { "buckets": ["valid", "invalid"], "transitions": { "valid": {"content_blocks": ["Valid number!"]}, - "invalid": {"content_blocks": ["Invalid input!"]} - } + "invalid": {"content_blocks": ["Invalid input!"]}, + }, } - + activity_state = MockActivityState() - + # Simulate user response user_response = "42" temp_metadata = activity_state.dict_metadata.copy() temp_metadata["user_response"] = user_response - + # Execute pre-script pre_result = app.execute_processing_script( - temp_metadata, - pre_script_step["pre_script"] + temp_metadata, pre_script_step["pre_script"] ) - + # Update metadata with pre-script results for key, value in pre_result.get("metadata", {}).items(): activity_state.add_metadata(key, value) - + # Copy processed data back (excluding temporary user_response) - activity_state.add_metadata('parsed_number', temp_metadata['parsed_number']) - activity_state.add_metadata('is_valid_number', temp_metadata['is_valid_number']) - activity_state.add_metadata('number_category', temp_metadata['number_category']) - + activity_state.add_metadata("parsed_number", temp_metadata["parsed_number"]) + activity_state.add_metadata("is_valid_number", temp_metadata["is_valid_number"]) + activity_state.add_metadata("number_category", temp_metadata["number_category"]) + # Verify pre-script execution - self.assertTrue(activity_state.dict_metadata['pre_processing_complete']) - self.assertEqual(activity_state.dict_metadata['parsed_number'], 42) - self.assertTrue(activity_state.dict_metadata['is_valid_number']) - self.assertEqual(activity_state.dict_metadata['number_category'], 'positive') - + self.assertTrue(activity_state.dict_metadata["pre_processing_complete"]) + self.assertEqual(activity_state.dict_metadata["parsed_number"], 42) + self.assertTrue(activity_state.dict_metadata["is_valid_number"]) + self.assertEqual(activity_state.dict_metadata["number_category"], "positive") + def test_metadata_operations_integration(self): """Test various metadata operations in sequence""" activity_state = MockActivityState() - + # Test metadata_add with various value types metadata_add_ops = { "simple_value": "test", "numeric_increment": "n+5", "random_increment": "n+random(1,10)", - "user_response_copy": "the-users-response" + "user_response_copy": "the-users-response", } - + activity_state.add_metadata("numeric_increment", 10) user_response = "Hello World" - + for key, value in metadata_add_ops.items(): if value == "the-users-response": processed_value = user_response elif isinstance(value, str) and value.startswith("n+random("): # For testing, we'll use a fixed random value - processed_value = activity_state.dict_metadata.get(key, 0) + 5 # Fixed for testing + processed_value = ( + activity_state.dict_metadata.get(key, 0) + 5 + ) # Fixed for testing elif isinstance(value, str) and value.startswith("n+"): c = int(value[2:]) processed_value = activity_state.dict_metadata.get(key, 0) + c else: processed_value = value - + activity_state.add_metadata(key, processed_value) - + # Verify metadata operations self.assertEqual(activity_state.dict_metadata["simple_value"], "test") self.assertEqual(activity_state.dict_metadata["numeric_increment"], 15) self.assertEqual(activity_state.dict_metadata["random_increment"], 5) - self.assertEqual(activity_state.dict_metadata["user_response_copy"], "Hello World") - + self.assertEqual( + activity_state.dict_metadata["user_response_copy"], "Hello World" + ) + # Test metadata_remove activity_state.remove_metadata("simple_value") self.assertNotIn("simple_value", activity_state.dict_metadata) - + # Test metadata_clear activity_state.clear_metadata() self.assertEqual(len(activity_state.dict_metadata), 0) - + def test_activity_navigation_integration(self): """Test complete activity navigation""" multi_section_activity = { @@ -304,55 +309,53 @@ script_result = { "section_id": "intro", "steps": [ {"step_id": "step_1", "title": "Intro Step 1"}, - {"step_id": "step_2", "title": "Intro Step 2"} - ] + {"step_id": "step_2", "title": "Intro Step 2"}, + ], }, { "section_id": "main", "steps": [ {"step_id": "step_1", "title": "Main Step 1"}, - {"step_id": "step_2", "title": "Main Step 2"} - ] + {"step_id": "step_2", "title": "Main Step 2"}, + ], }, { "section_id": "conclusion", - "steps": [ - {"step_id": "final", "title": "Final Step"} - ] - } + "steps": [{"step_id": "final", "title": "Final Step"}], + }, ] } - + # Test navigation through multiple sections current_section = "intro" current_step = "step_1" - + navigation_path = [] - + for _ in range(10): # Prevent infinite loop next_section, next_step = app.get_next_step( multi_section_activity, current_section, current_step ) - + navigation_path.append((current_section, current_step)) - + if next_section is None or next_step is None: break - + current_section = next_section["section_id"] current_step = next_step["step_id"] - + # Verify complete navigation path expected_path = [ ("intro", "step_1"), - ("intro", "step_2"), + ("intro", "step_2"), ("main", "step_1"), ("main", "step_2"), - ("conclusion", "final") + ("conclusion", "final"), ] - + self.assertEqual(navigation_path, expected_path) - + def test_feedback_generation_integration(self): """Test complete feedback generation flow""" transition_with_feedback = { @@ -360,30 +363,34 @@ script_result = { "tokens_for_ai": "Provide encouraging feedback for correct math answers" } } - + # Mock the OpenAI response - mock_feedback = "Excellent! You correctly calculated 2+2=4. Great mathematical skills!" - - with patch.object(app, 'provide_feedback', return_value=mock_feedback) as mock_func: + mock_feedback = ( + "Excellent! You correctly calculated 2+2=4. Great mathematical skills!" + ) + + with patch.object( + app, "provide_feedback", return_value=mock_feedback + ) as mock_func: result = app.provide_feedback( transition_with_feedback, "correct", - "What is 2+2?", + "What is 2+2?", "Base feedback instructions", "4", "English", "testuser", json.dumps({"score": 1}), - json.dumps({"score": 2}) + json.dumps({"score": 2}), ) - + self.assertEqual(result, mock_feedback) mock_func.assert_called_once() class TestActivityErrorHandling(unittest.TestCase): """Test error handling in activity processing""" - + def test_invalid_processing_script(self): """Test handling of invalid processing scripts""" invalid_script = """ @@ -392,11 +399,11 @@ if True print("Missing colon") """ metadata = {} - + # Should handle syntax errors gracefully with self.assertRaises(SyntaxError): app.execute_processing_script(metadata, invalid_script) - + def test_processing_script_runtime_error(self): """Test handling of runtime errors in processing scripts""" runtime_error_script = """ @@ -405,48 +412,48 @@ result = 1 / 0 # Division by zero script_result = {'status': 'error'} """ metadata = {} - + # Should handle runtime errors gracefully with self.assertRaises(ZeroDivisionError): app.execute_processing_script(metadata, runtime_error_script) - + def test_missing_activity_content(self): """Test handling of missing activity content""" - with patch.object(app, 'get_activity_content') as mock_get_content: + with patch.object(app, "get_activity_content") as mock_get_content: mock_get_content.side_effect = FileNotFoundError("Activity file not found") - + with self.assertRaises(FileNotFoundError): app.get_activity_content("nonexistent_activity.yaml") - + mock_get_content.assert_called_once_with("nonexistent_activity.yaml") - + def test_malformed_yaml_content(self): """Test handling of malformed YAML content""" malformed_yaml = "invalid: yaml: content: [unclosed" - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: + + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(malformed_yaml) temp_file = f.name - + try: # Should handle YAML parsing errors - with patch.dict(app.app.config, {'LOCAL_ACTIVITIES': True}): + with patch.dict(app.app.config, {"LOCAL_ACTIVITIES": True}): # Create research directory and file research_dir = Path("research") research_dir.mkdir(exist_ok=True) - + test_file = research_dir / "malformed.yaml" - with open(test_file, 'w') as f: + with open(test_file, "w") as f: f.write(malformed_yaml) - + with self.assertRaises(Exception): # YAML parsing error app.get_activity_content("research/malformed.yaml") - + finally: os.unlink(temp_file) if test_file.exists(): test_file.unlink() -if __name__ == '__main__': - unittest.main(verbosity=2) \ No newline at end of file +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/unit/test_activity_yaml_validator.py b/tests/unit/test_activity_yaml_validator.py index 20f8f38..adb2bbf 100644 --- a/tests/unit/test_activity_yaml_validator.py +++ b/tests/unit/test_activity_yaml_validator.py @@ -4,7 +4,7 @@ Unit tests for the activity_yaml_validator.py module. Tests all validation features including: - YAML syntax validation -- Structure validation +- Structure validation - Metadata operations validation - Python code validation - Logic flow validation @@ -24,22 +24,22 @@ from activity_yaml_validator import ActivityYAMLValidator, ValidationError class TestActivityYAMLValidator(unittest.TestCase): """Test cases for ActivityYAMLValidator""" - + def setUp(self): """Set up test fixtures""" self.validator = ActivityYAMLValidator() - + def create_temp_yaml(self, content: str) -> str: """Create a temporary YAML file with given content""" - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: + with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f: f.write(content) return f.name - + def tearDown(self): """Clean up any temporary files""" # Clean up is handled by tempfile pass - + def test_valid_yaml_passes(self): """Test that a valid YAML file passes validation""" valid_yaml = """ @@ -80,7 +80,7 @@ sections: self.assertEqual(len(errors), 0) finally: os.unlink(temp_file) - + def test_yaml_syntax_error(self): """Test that YAML syntax errors are caught""" invalid_yaml = """ @@ -102,7 +102,7 @@ sections: self.assertIn("YAML syntax error", errors[0]) finally: os.unlink(temp_file) - + def test_missing_required_fields(self): """Test that missing required fields are caught""" missing_sections = """ @@ -115,7 +115,7 @@ default_max_attempts_per_step: 3 self.assertIn("Missing required field: sections", errors) finally: os.unlink(temp_file) - + def test_invalid_field_types(self): """Test that invalid field types are caught""" invalid_types = """ @@ -131,11 +131,13 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) - self.assertTrue(any("must be a positive integer" in error for error in errors)) + self.assertTrue( + any("must be a positive integer" in error for error in errors) + ) self.assertTrue(any("must be a string" in error for error in errors)) finally: os.unlink(temp_file) - + def test_duplicate_ids(self): """Test that duplicate section and step IDs are caught""" duplicate_ids = """ @@ -168,16 +170,40 @@ sections: self.assertTrue(any("Duplicate step_id" in error for error in errors)) finally: os.unlink(temp_file) - + def test_terminal_step_validation(self): """Test that terminal steps cannot have questions or buckets""" terminal_with_question = """ sections: - section_id: "section_1" - title: "Test" + title: "First Section" steps: - - step_id: "terminal_step" - title: "Final Step" + - step_id: "step_1" + title: "First Step" + content_blocks: + - "This step is fine" + - step_id: "step_2" + title: "Also fine" + question: "Questions are OK in non-terminal steps" + buckets: ["yes", "no"] + transitions: + yes: + content_blocks: ["Good"] + next_section_and_step: "section_2:step_1" + no: + content_blocks: ["Try again"] + - section_id: "section_2" + title: "Last Section" + steps: + - step_id: "step_1" + title: "Not terminal - has another step after" + question: "This is OK" + buckets: ["answer"] + transitions: + answer: + content_blocks: ["Continue"] + - step_id: "step_2" + title: "This is the real terminal step" question: "This is invalid" buckets: - some_bucket @@ -185,17 +211,24 @@ sections: some_bucket: content_blocks: - "Done" - # No next_section_and_step makes this terminal + # No next_section_and_step and last step of last section = terminal """ temp_file = self.create_temp_yaml(terminal_with_question) try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) - self.assertTrue(any("Final/terminal steps cannot have questions" in error for error in errors)) - self.assertTrue(any("Final/terminal steps should not have buckets" in error for error in errors)) + # Should only flag the last step of the last section + terminal_errors = [e for e in errors if "Final/terminal" in e] + self.assertEqual(len(terminal_errors), 2) # One for question, one for buckets + self.assertTrue( + any( + "section_2" in error and "step_2" in error + for error in terminal_errors + ) + ) finally: os.unlink(temp_file) - + def test_metadata_operations_validation(self): """Test validation of metadata operations""" metadata_test = """ @@ -225,13 +258,27 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) - self.assertTrue(any("metadata_clear' must be boolean" in error for error in errors)) - self.assertTrue(any("metadata_feedback_filter' must be a list" in error for error in errors)) - self.assertTrue(any("metadata_remove' must be a string or list of strings" in error for error in errors)) - self.assertTrue(any("metadata_add' must be a dictionary" in error for error in errors)) + self.assertTrue( + any("metadata_clear' must be boolean" in error for error in errors) + ) + self.assertTrue( + any( + "metadata_feedback_filter' must be a list" in error + for error in errors + ) + ) + self.assertTrue( + any( + "metadata_remove' must be a string or list of strings" in error + for error in errors + ) + ) + self.assertTrue( + any("metadata_add' must be a dictionary" in error for error in errors) + ) finally: os.unlink(temp_file) - + def test_valid_metadata_operations(self): """Test that valid metadata operations pass""" valid_metadata = """ @@ -280,7 +327,7 @@ sections: self.assertEqual(len(errors), 0) finally: os.unlink(temp_file) - + def test_python_syntax_validation(self): """Test that Python syntax errors in scripts are caught""" python_syntax_error = """ @@ -316,7 +363,7 @@ sections: self.assertTrue(any("Python syntax error" in error for error in errors)) finally: os.unlink(temp_file) - + def test_invalid_transitions(self): """Test validation of transition references""" invalid_transitions = """ @@ -344,13 +391,20 @@ sections: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) # Should have errors for invalid transition targets and missing transitions - self.assertTrue(any("Invalid transition target" in error for error in errors)) - self.assertTrue(any("must be in format 'section_id:step_id'" in error for error in errors)) + self.assertTrue( + any("Invalid transition target" in error for error in errors) + ) + self.assertTrue( + any( + "must be in format 'section_id:step_id'" in error + for error in errors + ) + ) # Should have warnings for unused transitions self.assertTrue(any("Unused transition" in warning for warning in warnings)) finally: os.unlink(temp_file) - + def test_metadata_feedback_filter_warning(self): """Test warning when metadata_feedback_filter used without feedback_tokens_for_ai""" metadata_filter_no_feedback = """ @@ -378,10 +432,16 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertTrue(is_valid) # Should be valid but with warning - self.assertTrue(any("metadata_feedback_filter used but no feedback_tokens_for_ai" in warning for warning in warnings)) + self.assertTrue( + any( + "metadata_feedback_filter used but no feedback_tokens_for_ai" + in warning + for warning in warnings + ) + ) finally: os.unlink(temp_file) - + def test_pre_script_warning(self): """Test warning when pre_script used without question""" pre_script_no_question = """ @@ -400,10 +460,15 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertTrue(is_valid) # Should be valid but with warning - self.assertTrue(any("pre_script typically used with question steps" in warning for warning in warnings)) + self.assertTrue( + any( + "pre_script typically used with question steps" in warning + for warning in warnings + ) + ) finally: os.unlink(temp_file) - + def test_empty_else_block_detection(self): """Test detection of empty else blocks in Python code""" empty_else_block = """ @@ -434,10 +499,12 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) # This should detect the empty else block - self.assertTrue(any("'else:' block contains only comments" in error for error in errors)) + self.assertTrue( + any("'else:' block contains only comments" in error for error in errors) + ) finally: os.unlink(temp_file) - + def test_content_blocks_validation(self): """Test validation of content_blocks structure""" invalid_content_blocks = """ @@ -460,11 +527,13 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) - self.assertTrue(any("content_blocks must be a list" in error for error in errors)) + self.assertTrue( + any("content_blocks must be a list" in error for error in errors) + ) self.assertTrue(any("must be a string" in error for error in errors)) finally: os.unlink(temp_file) - + def test_transition_fields_validation(self): """Test validation of various transition fields""" invalid_transition_fields = """ @@ -507,13 +576,24 @@ sections: try: is_valid, errors, warnings = self.validator.validate_file(temp_file) self.assertFalse(is_valid) - self.assertTrue(any("run_processing_script' must be boolean" in error for error in errors)) - self.assertTrue(any("ai_feedback' must be a dictionary" in error for error in errors)) - self.assertTrue(any("tokens_for_ai must be a string" in error for error in errors)) - self.assertTrue(any("content_blocks' must be a list" in error for error in errors)) + self.assertTrue( + any( + "run_processing_script' must be boolean" in error + for error in errors + ) + ) + self.assertTrue( + any("ai_feedback' must be a dictionary" in error for error in errors) + ) + self.assertTrue( + any("tokens_for_ai must be a string" in error for error in errors) + ) + self.assertTrue( + any("content_blocks' must be a list" in error for error in errors) + ) finally: os.unlink(temp_file) - + def test_using_existing_failing_fixture(self): """Test using the existing failing fixture we created""" fixture_path = "tests/fixtures/test_invalid.yaml" @@ -523,32 +603,45 @@ sections: self.assertGreater(len(errors), 0) # Should catch the YAML syntax error we know is in there self.assertTrue(any("YAML syntax error" in error for error in errors)) - + def test_cli_integration(self): """Test the command line interface""" import subprocess import sys - + # Test with valid battleship YAML - result = subprocess.run([ - sys.executable, "activity_yaml_validator.py", - "research/activity29-battleship.yaml" - ], capture_output=True, text=True, cwd=".") - + result = subprocess.run( + [ + sys.executable, + "activity_yaml_validator.py", + "research/activity29-battleship.yaml", + ], + capture_output=True, + text=True, + cwd=".", + ) + # Should succeed (exit code 0) despite warnings self.assertEqual(result.returncode, 0) self.assertIn("valid", result.stdout.lower()) - + # Test with --strict flag (warnings become errors) - result = subprocess.run([ - sys.executable, "activity_yaml_validator.py", - "research/activity29-battleship.yaml", "--strict" - ], capture_output=True, text=True, cwd=".") - + result = subprocess.run( + [ + sys.executable, + "activity_yaml_validator.py", + "research/activity29-battleship.yaml", + "--strict", + ], + capture_output=True, + text=True, + cwd=".", + ) + # Should fail (exit code 1) because warnings become errors in strict mode self.assertEqual(result.returncode, 1) -if __name__ == '__main__': +if __name__ == "__main__": # Run the tests - unittest.main(verbosity=2) \ No newline at end of file + unittest.main(verbosity=2) diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index f9d400f..c3b8d8f 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -18,75 +18,88 @@ from pathlib import Path 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(), -}): +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 - + 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: + 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: + 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') - + 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') - + 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') - + 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') - + 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: + + 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: + 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) @@ -94,21 +107,21 @@ class TestAppUtilityFunctions(unittest.TestCase): 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'} - + 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') - + + 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 = """ @@ -123,15 +136,15 @@ script_result = { } } """ - metadata = {'input_value': 21, 'list_field': [1, 2, 3, 4, 5]} - + 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) - + + 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 = """ @@ -148,17 +161,17 @@ script_result = { } """ metadata = {} - + result = app.execute_processing_script(metadata, script) - - self.assertTrue(result['has_random']) - self.assertIsInstance(result['json_output'], str) - + + 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) - + 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 = """ @@ -172,70 +185,69 @@ sections: content_blocks: - "Test content" """ - - with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f: + + 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: + 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}): + 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") - + + 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}): + 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" + "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' - }] + "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: + + 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") + + 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 = { @@ -245,115 +257,123 @@ class TestActivityNavigation(unittest.TestCase): "steps": [ {"step_id": "step_1", "title": "Step 1"}, {"step_id": "step_2", "title": "Step 2"}, - {"step_id": "step_3", "title": "Step 3"} - ] + {"step_id": "step_3", "title": "Step 3"}, + ], }, { - "section_id": "section_2", + "section_id": "section_2", "steps": [ {"step_id": "step_1", "title": "Section 2 Step 1"}, - {"step_id": "step_2", "title": "Section 2 Step 2"} - ] - } + {"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: + 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." + ["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: + 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" + "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: + 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?", @@ -361,112 +381,114 @@ class TestResponseCategorizationAndFeedback(unittest.TestCase): "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: + 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", + "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", + "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") + 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: + 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: + 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: + + 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: + + 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 @@ -480,14 +502,14 @@ def test_function(): And some more text after. """ - + # Extract the code block manually to test the logic - lines = test_content.split('\n') + lines = test_content.split("\n") code_block_lines = [] code_block_started = False - + for line in lines: - if line.startswith('```'): + if line.startswith("```"): if code_block_started: break else: @@ -495,17 +517,17 @@ And some more text after. continue elif code_block_started: code_block_lines.append(line) - - result = '\n'.join(code_block_lines) + + 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 = [ @@ -513,24 +535,24 @@ class TestUtilityFunctions(unittest.TestCase): {"role": "user", "content": "How are you?"}, {"role": "assistant", "content": "I'm fine"}, {"role": "assistant", "content": "Thanks for asking"}, - {"role": "user", "content": "Great!"} + {"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!"} + {"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"}] @@ -538,5 +560,5 @@ class TestUtilityFunctions(unittest.TestCase): self.assertEqual(result, messages) -if __name__ == '__main__': - unittest.main(verbosity=2) \ No newline at end of file +if __name__ == "__main__": + unittest.main(verbosity=2)