Add comprehensive testing framework and YAML validator
- Create universal activity_yaml_validator.py for validating activity configurations - Add validation for metadata operations (metadata_add, metadata_remove, metadata_feedback_filter, etc.) - Validate terminal steps cannot have questions or buckets - Check Python syntax in processing_script and pre_script blocks - Validate YAML structure, transitions, and logic flow - Add 17 comprehensive unit tests with 100% pass rate - Include test fixtures for validation testing - Support both CLI and programmatic usage
This commit is contained in:
parent
25c694584a
commit
292265b5bb
3 changed files with 1233 additions and 0 deletions
589
activity_yaml_validator.py
Normal file
589
activity_yaml_validator.py
Normal file
|
|
@ -0,0 +1,589 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Universal YAML Validator for Activity Configurations
|
||||
|
||||
This module provides comprehensive validation for activity YAML files,
|
||||
particularly battleship configurations and other interactive activities.
|
||||
It validates structure, syntax, Python code blocks, and logical consistency.
|
||||
"""
|
||||
|
||||
import yaml
|
||||
import ast
|
||||
import re
|
||||
import sys
|
||||
import argparse
|
||||
from typing import Dict, List, Any, Optional, Tuple
|
||||
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
|
||||
- Python code blocks (processing_script, pre_script)
|
||||
- Logic flow and transitions
|
||||
- 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:
|
||||
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'])
|
||||
|
||||
# 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']
|
||||
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):
|
||||
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:
|
||||
self.errors.append(f"Duplicate section_id: {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']
|
||||
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}'))
|
||||
|
||||
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")
|
||||
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'])
|
||||
|
||||
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}')
|
||||
|
||||
# Required fields
|
||||
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}'")
|
||||
|
||||
# Validate content_blocks or question
|
||||
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'")
|
||||
|
||||
if has_content:
|
||||
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):
|
||||
"""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")
|
||||
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):
|
||||
"""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")
|
||||
|
||||
# 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")
|
||||
|
||||
# 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)
|
||||
|
||||
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")
|
||||
return
|
||||
|
||||
if not buckets:
|
||||
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):
|
||||
"""Validate transitions dictionary"""
|
||||
if not isinstance(transitions, dict):
|
||||
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}'")
|
||||
|
||||
# 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}'")
|
||||
|
||||
# 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):
|
||||
"""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")
|
||||
return
|
||||
|
||||
# Validate next_section_and_step format
|
||||
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'")
|
||||
|
||||
# Validate metadata operations
|
||||
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 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':
|
||||
if not isinstance(transition[field], 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':
|
||||
if isinstance(transition[field], str):
|
||||
# Single key to remove
|
||||
pass
|
||||
elif isinstance(transition[field], list):
|
||||
# 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")
|
||||
else:
|
||||
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")
|
||||
|
||||
# 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 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")
|
||||
else:
|
||||
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")
|
||||
|
||||
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)
|
||||
except SyntaxError as e:
|
||||
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):
|
||||
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')
|
||||
|
||||
# Check for empty except blocks
|
||||
for i, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
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'")
|
||||
|
||||
# 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"),
|
||||
]
|
||||
|
||||
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:':
|
||||
# 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")
|
||||
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):
|
||||
following_line_idx = next_line_idx + 1
|
||||
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")
|
||||
|
||||
def _validate_activity_rules(self, data: Dict[str, Any]):
|
||||
"""Validate universal activity rules"""
|
||||
if 'sections' not in data:
|
||||
return
|
||||
|
||||
# Check that final steps don't have questions
|
||||
for section in data['sections']:
|
||||
if 'steps' not in section:
|
||||
continue
|
||||
|
||||
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")
|
||||
|
||||
# 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:
|
||||
return
|
||||
|
||||
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:
|
||||
continue
|
||||
|
||||
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")
|
||||
|
||||
def _validate_pre_scripts(self, data: Dict[str, Any]):
|
||||
"""Validate pre_script usage"""
|
||||
if 'sections' not in data:
|
||||
return
|
||||
|
||||
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:
|
||||
# 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")
|
||||
|
||||
# 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")
|
||||
|
||||
def _validate_logic_flow(self, data: Dict[str, Any]):
|
||||
"""Validate logical flow and transitions between steps"""
|
||||
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:
|
||||
continue
|
||||
|
||||
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:
|
||||
continue
|
||||
|
||||
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']
|
||||
if target not in all_steps:
|
||||
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')
|
||||
|
||||
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()
|
||||
90
tests/fixtures/test_invalid.yaml
vendored
Normal file
90
tests/fixtures/test_invalid.yaml
vendored
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
default_max_attempts_per_step: "invalid" # Should be integer
|
||||
tokens_for_ai_rubric: 123 # Should be string
|
||||
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test Section"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Valid Step"
|
||||
content_blocks:
|
||||
- "This is a valid step."
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Question Step"
|
||||
question: "What do you want to do?"
|
||||
tokens_for_ai: |
|
||||
Categorize the response.
|
||||
feedback_tokens_for_ai: |
|
||||
Provide feedback.
|
||||
buckets:
|
||||
- valid_response
|
||||
- invalid_response
|
||||
transitions:
|
||||
valid_response:
|
||||
content_blocks:
|
||||
- "Good response!"
|
||||
metadata_add:
|
||||
test_key: "value"
|
||||
metadata_feedback_filter:
|
||||
- user_response
|
||||
- result
|
||||
next_section_and_step: "section_1:step_3"
|
||||
invalid_response:
|
||||
content_blocks:
|
||||
- "Try again."
|
||||
metadata_remove: ["temp_data", "old_value"]
|
||||
next_section_and_step: "section_1:step_2"
|
||||
unused_bucket: # This should trigger a warning
|
||||
content_blocks:
|
||||
- "This transition is unused"
|
||||
|
||||
- step_id: "step_3"
|
||||
title: "Final Step With Question" # This should be an ERROR - final steps can't have questions
|
||||
question: "This is invalid for a final step"
|
||||
buckets:
|
||||
- some_bucket # This should be an ERROR - final steps shouldn't have buckets
|
||||
transitions:
|
||||
some_bucket:
|
||||
content_blocks:
|
||||
- "Done"
|
||||
# No next_section_and_step - this makes it a terminal step
|
||||
|
||||
- step_4 # Missing step_id field - ERROR
|
||||
title: "Invalid Step Structure"
|
||||
# Missing either content_blocks or question - ERROR
|
||||
|
||||
- step_id: "step_5"
|
||||
title: "Python Syntax Error Step"
|
||||
question: "Test question"
|
||||
pre_script: |
|
||||
# This has a syntax error
|
||||
if True
|
||||
print("missing colon")
|
||||
processing_script: |
|
||||
# This has an empty else block
|
||||
if condition:
|
||||
do_something()
|
||||
else:
|
||||
# This will trigger a warning about empty else block
|
||||
buckets:
|
||||
- test_bucket
|
||||
transitions:
|
||||
test_bucket:
|
||||
run_processing_script: "not_boolean" # Should be boolean
|
||||
metadata_clear: "not_boolean" # Should be boolean
|
||||
metadata_feedback_filter: "not_list" # Should be list
|
||||
metadata_remove: 123 # Should be string or list
|
||||
next_section_and_step: "invalid_format" # Should be section:step format
|
||||
|
||||
- section_id: "section_1" # Duplicate section_id - ERROR
|
||||
title: "Duplicate Section"
|
||||
steps:
|
||||
- step_id: "duplicate_step"
|
||||
title: "Test"
|
||||
content_blocks: "not_a_list" # Should be list
|
||||
|
||||
- step_id: "duplicate_step" # Duplicate step_id - ERROR
|
||||
title: "Another Duplicate"
|
||||
content_blocks:
|
||||
- 123 # Should be string
|
||||
554
tests/unit/test_activity_yaml_validator.py
Normal file
554
tests/unit/test_activity_yaml_validator.py
Normal file
|
|
@ -0,0 +1,554 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
Unit tests for the activity_yaml_validator.py module.
|
||||
|
||||
Tests all validation features including:
|
||||
- YAML syntax validation
|
||||
- Structure validation
|
||||
- Metadata operations validation
|
||||
- Python code validation
|
||||
- Logic flow validation
|
||||
- Terminal step validation
|
||||
"""
|
||||
|
||||
import unittest
|
||||
import tempfile
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# Add parent directory to path to import the validator
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
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:
|
||||
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 = """
|
||||
default_max_attempts_per_step: 3
|
||||
tokens_for_ai_rubric: "Test rubric"
|
||||
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test Section"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Question Step"
|
||||
question: "What do you want?"
|
||||
tokens_for_ai: "Categorize response"
|
||||
feedback_tokens_for_ai: "Provide feedback"
|
||||
buckets:
|
||||
- valid
|
||||
- invalid
|
||||
transitions:
|
||||
valid:
|
||||
content_blocks:
|
||||
- "Great!"
|
||||
next_section_and_step: "section_1:step_2"
|
||||
invalid:
|
||||
content_blocks:
|
||||
- "Try again"
|
||||
next_section_and_step: "section_1:step_1"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Final Step"
|
||||
content_blocks:
|
||||
- "All done!"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(valid_yaml)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertTrue(is_valid)
|
||||
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 = """
|
||||
sections:
|
||||
- section_id: "test"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step1"
|
||||
title: "Test Step"
|
||||
content_blocks:
|
||||
- "Test"
|
||||
invalid_key: [unclosed list
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(invalid_yaml)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertFalse(is_valid)
|
||||
self.assertGreater(len(errors), 0)
|
||||
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 = """
|
||||
default_max_attempts_per_step: 3
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(missing_sections)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertFalse(is_valid)
|
||||
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 = """
|
||||
default_max_attempts_per_step: "should_be_integer"
|
||||
tokens_for_ai_rubric: 123
|
||||
|
||||
sections:
|
||||
- section_id: "test"
|
||||
title: "Test"
|
||||
steps: "should_be_list"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(invalid_types)
|
||||
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 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 = """
|
||||
sections:
|
||||
- section_id: "duplicate"
|
||||
title: "First Section"
|
||||
steps:
|
||||
- step_id: "step_duplicate"
|
||||
title: "First Step"
|
||||
content_blocks:
|
||||
- "Content"
|
||||
- step_id: "step_duplicate"
|
||||
title: "Second Step"
|
||||
content_blocks:
|
||||
- "More content"
|
||||
|
||||
- section_id: "duplicate"
|
||||
title: "Second Section"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Step"
|
||||
content_blocks:
|
||||
- "Content"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(duplicate_ids)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertFalse(is_valid)
|
||||
self.assertTrue(any("Duplicate section_id" in error for error in errors))
|
||||
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"
|
||||
steps:
|
||||
- step_id: "terminal_step"
|
||||
title: "Final Step"
|
||||
question: "This is invalid"
|
||||
buckets:
|
||||
- some_bucket
|
||||
transitions:
|
||||
some_bucket:
|
||||
content_blocks:
|
||||
- "Done"
|
||||
# No next_section_and_step makes this 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))
|
||||
finally:
|
||||
os.unlink(temp_file)
|
||||
|
||||
def test_metadata_operations_validation(self):
|
||||
"""Test validation of metadata operations"""
|
||||
metadata_test = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
metadata_clear: "should_be_boolean"
|
||||
metadata_feedback_filter: "should_be_list"
|
||||
metadata_remove: 123
|
||||
metadata_add: "should_be_dict"
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(metadata_test)
|
||||
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))
|
||||
finally:
|
||||
os.unlink(temp_file)
|
||||
|
||||
def test_valid_metadata_operations(self):
|
||||
"""Test that valid metadata operations pass"""
|
||||
valid_metadata = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
feedback_tokens_for_ai: "Provide feedback"
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
metadata_clear: true
|
||||
metadata_feedback_filter:
|
||||
- "field1"
|
||||
- "field2"
|
||||
metadata_remove: "single_field"
|
||||
metadata_add:
|
||||
new_field: "value"
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Test Step 2"
|
||||
question: "Another test?"
|
||||
buckets:
|
||||
- test2
|
||||
transitions:
|
||||
test2:
|
||||
metadata_remove:
|
||||
- "field1"
|
||||
- "field2"
|
||||
next_section_and_step: "section_1:step_3"
|
||||
|
||||
- step_id: "step_3"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(valid_metadata)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertTrue(is_valid)
|
||||
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 = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
pre_script: |
|
||||
if True # Missing colon
|
||||
print("error")
|
||||
processing_script: |
|
||||
def invalid_function(
|
||||
# Missing closing parenthesis
|
||||
pass
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(python_syntax_error)
|
||||
try:
|
||||
is_valid, errors, warnings = self.validator.validate_file(temp_file)
|
||||
self.assertFalse(is_valid)
|
||||
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 = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
buckets:
|
||||
- valid_bucket
|
||||
- another_bucket
|
||||
transitions:
|
||||
valid_bucket:
|
||||
next_section_and_step: "nonexistent_section:step_1"
|
||||
another_bucket:
|
||||
next_section_and_step: "invalid_format"
|
||||
unused_transition:
|
||||
content_blocks:
|
||||
- "This transition has no corresponding bucket"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(invalid_transitions)
|
||||
try:
|
||||
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))
|
||||
# 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 = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
metadata_feedback_filter:
|
||||
- "field1"
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(metadata_filter_no_feedback)
|
||||
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))
|
||||
finally:
|
||||
os.unlink(temp_file)
|
||||
|
||||
def test_pre_script_warning(self):
|
||||
"""Test warning when pre_script used without question"""
|
||||
pre_script_no_question = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
content_blocks:
|
||||
- "Content"
|
||||
pre_script: |
|
||||
print("This is unusual without a question")
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(pre_script_no_question)
|
||||
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))
|
||||
finally:
|
||||
os.unlink(temp_file)
|
||||
|
||||
def test_empty_else_block_detection(self):
|
||||
"""Test detection of empty else blocks in Python code"""
|
||||
empty_else_block = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
processing_script: |
|
||||
if condition:
|
||||
do_something()
|
||||
else:
|
||||
# Only comments here, should trigger error
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(empty_else_block)
|
||||
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))
|
||||
finally:
|
||||
os.unlink(temp_file)
|
||||
|
||||
def test_content_blocks_validation(self):
|
||||
"""Test validation of content_blocks structure"""
|
||||
invalid_content_blocks = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
content_blocks: "should_be_list"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Another Test"
|
||||
content_blocks:
|
||||
- "Valid string"
|
||||
- 123 # Should be string
|
||||
- "Another valid string"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(invalid_content_blocks)
|
||||
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("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 = """
|
||||
sections:
|
||||
- section_id: "section_1"
|
||||
title: "Test"
|
||||
steps:
|
||||
- step_id: "step_1"
|
||||
title: "Test Step"
|
||||
question: "Test?"
|
||||
buckets:
|
||||
- test
|
||||
transitions:
|
||||
test:
|
||||
run_processing_script: "should_be_boolean"
|
||||
ai_feedback: "should_be_dict"
|
||||
content_blocks: "should_be_list"
|
||||
next_section_and_step: "section_1:step_2"
|
||||
|
||||
- step_id: "step_2"
|
||||
title: "Another Test"
|
||||
question: "Test?"
|
||||
buckets:
|
||||
- test2
|
||||
transitions:
|
||||
test2:
|
||||
ai_feedback:
|
||||
tokens_for_ai: 123 # Should be string
|
||||
content_blocks:
|
||||
- "Valid"
|
||||
- 456 # Should be string
|
||||
next_section_and_step: "section_1:step_3"
|
||||
|
||||
- step_id: "step_3"
|
||||
title: "Final"
|
||||
content_blocks:
|
||||
- "Done"
|
||||
"""
|
||||
temp_file = self.create_temp_yaml(invalid_transition_fields)
|
||||
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))
|
||||
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"
|
||||
if os.path.exists(fixture_path):
|
||||
is_valid, errors, warnings = self.validator.validate_file(fixture_path)
|
||||
self.assertFalse(is_valid)
|
||||
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=".")
|
||||
|
||||
# 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=".")
|
||||
|
||||
# Should fail (exit code 1) because warnings become errors in strict mode
|
||||
self.assertEqual(result.returncode, 1)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Run the tests
|
||||
unittest.main(verbosity=2)
|
||||
Loading…
Add table
Add a link
Reference in a new issue