test_analysis.py

#!/usr/bin/env python3
"""
Test Script for Big Data Text Analysis Pipeline
Validates the functionality of all components
"""

import sys
import os
import tempfile
import subprocess
import logging
from typing import List, Dict, Any
from utils import setup_logging, preprocess_text
from config import Config

# Setup logging
setup_logging()
logger = logging.getLogger(__name__)

class AnalysisTester:
    """Test class for the analysis pipeline"""
    
    def __init__(self):
        self.test_text = """
        This is a test document for the EKU student handbook analysis.
        The university provides various academic services to students.
        Students must follow the academic policies and conduct guidelines.
        The campus offers many resources for student success.
        """
        self.temp_dir = tempfile.mkdtemp()
    
    def test_preprocessing(self) -> bool:
        """Test text preprocessing functionality"""
        try:
            logger.info("Testing text preprocessing...")
            
            tokens = preprocess_text(self.test_text, remove_stops=True)
            
            if not tokens:
                logger.error("Preprocessing returned no tokens")
                return False
            
            # Check that stopwords are removed
            stopwords = Config.get_stopwords()
            for token in tokens:
                if token in stopwords:
                    logger.error(f"Stopword '{token}' not removed")
                    return False
            
            logger.info(f"Preprocessing successful: {len(tokens)} tokens")
            return True
            
        except Exception as e:
            logger.error(f"Preprocessing test failed: {e}")
            return False
    
    def test_word_count_pipeline(self) -> bool:
        """Test word count MapReduce pipeline"""
        try:
            logger.info("Testing word count pipeline...")
            
            # Create temporary input file
            input_file = os.path.join(self.temp_dir, "test_input.txt")
            with open(input_file, 'w') as f:
                f.write(self.test_text)
            
            # Run preprocessing - use appropriate command for OS
            if os.name == 'nt':  # Windows
                preprocess_cmd = f"type {input_file} | python preprocess.py"
            else:  # Unix/Linux/Mac
                preprocess_cmd = f"cat {input_file} | python preprocess.py"
                
            preprocess_result = subprocess.run(
                preprocess_cmd, shell=True, capture_output=True, text=True
            )
            
            if preprocess_result.returncode != 0:
                logger.error(f"Preprocessing failed: {preprocess_result.stderr}")
                return False
            
            # Run mapper
            mapper_cmd = f"echo '{preprocess_result.stdout}' | python mapper.py"
            mapper_result = subprocess.run(
                mapper_cmd, shell=True, capture_output=True, text=True
            )
            
            if mapper_result.returncode != 0:
                logger.error(f"Mapper failed: {mapper_result.stderr}")
                return False
            
            # Run reducer
            reducer_cmd = f"echo '{mapper_result.stdout}' | python reducer.py"
            reducer_result = subprocess.run(
                reducer_cmd, shell=True, capture_output=True, text=True
            )
            
            if reducer_result.returncode != 0:
                logger.error(f"Reducer failed: {reducer_result.stderr}")
                return False
            
            # Check output
            output_lines = reducer_result.stdout.strip().split('\n')
            if len(output_lines) == 0:
                logger.error("No word count output")
                return False
            
            logger.info("Word count pipeline test successful")
            return True
            
        except Exception as e:
            logger.error(f"Word count pipeline test failed: {e}")
            return False
    
    def test_tfidf_pipeline(self) -> bool:
        """Test TF-IDF pipeline"""
        try:
            logger.info("Testing TF-IDF pipeline...")
            
            # Create temporary input file
            input_file = os.path.join(self.temp_dir, "test_input.txt")
            with open(input_file, 'w') as f:
                f.write(self.test_text)
            
            # Run TF-IDF preprocessing - use appropriate command for OS
            if os.name == 'nt':  # Windows
                preprocess_cmd = f"type {input_file} | python preprocess_tfidf.py"
            else:  # Unix/Linux/Mac
                preprocess_cmd = f"cat {input_file} | python preprocess_tfidf.py"
                
            preprocess_result = subprocess.run(
                preprocess_cmd, shell=True, capture_output=True, text=True
            )
            
            if preprocess_result.returncode != 0:
                logger.error(f"TF-IDF preprocessing failed: {preprocess_result.stderr}")
                return False
            
            # Run TF-IDF mapper
            mapper_cmd = f"echo '{preprocess_result.stdout}' | python mapper_tfidf.py test_doc"
            mapper_result = subprocess.run(
                mapper_cmd, shell=True, capture_output=True, text=True
            )
            
            if mapper_result.returncode != 0:
                logger.error(f"TF-IDF mapper failed: {mapper_result.stderr}")
                return False
            
            # Run TF-IDF reducer
            reducer_cmd = f"echo '{mapper_result.stdout}' | python reducer_tfidf.py"
            reducer_result = subprocess.run(
                reducer_cmd, shell=True, capture_output=True, text=True
            )
            
            if reducer_result.returncode != 0:
                logger.error(f"TF-IDF reducer failed: {reducer_result.stderr}")
                return False
            
            logger.info("TF-IDF pipeline test successful")
            return True
            
        except Exception as e:
            logger.error(f"TF-IDF pipeline test failed: {e}")
            return False
    
    def test_lda_pipeline(self) -> bool:
        """Test LDA pipeline"""
        try:
            logger.info("Testing LDA pipeline...")
            
            # Create temporary input file
            input_file = os.path.join(self.temp_dir, "test_input.txt")
            with open(input_file, 'w') as f:
                f.write(self.test_text)
            
            # Run LDA mapper - use appropriate command for OS
            if os.name == 'nt':  # Windows
                mapper_cmd = f"type {input_file} | python mapper_lda.py"
            else:  # Unix/Linux/Mac
                mapper_cmd = f"cat {input_file} | python mapper_lda.py"
                
            mapper_result = subprocess.run(
                mapper_cmd, shell=True, capture_output=True, text=True
            )
            
            if mapper_result.returncode != 0:
                logger.error(f"LDA mapper failed: {mapper_result.stderr}")
                return False
            
            # Run LDA reducer
            reducer_cmd = f"echo '{mapper_result.stdout}' | python reducer_lda.py"
            reducer_result = subprocess.run(
                reducer_cmd, shell=True, capture_output=True, text=True
            )
            
            if reducer_result.returncode != 0:
                logger.error(f"LDA reducer failed: {reducer_result.stderr}")
                return False
            
            logger.info("LDA pipeline test successful")
            return True
            
        except Exception as e:
            logger.error(f"LDA pipeline test failed: {e}")
            return False
    
    def run_all_tests(self) -> Dict[str, bool]:
        """Run all tests and return results"""
        logger.info("Starting analysis pipeline tests...")
        
        results = {
            'preprocessing': self.test_preprocessing(),
            'word_count': self.test_word_count_pipeline(),
            'tfidf': self.test_tfidf_pipeline(),
            'lda': self.test_lda_pipeline()
        }
        
        # Print summary
        logger.info("\n" + "="*50)
        logger.info("TEST RESULTS SUMMARY")
        logger.info("="*50)
        
        all_passed = True
        for test_name, passed in results.items():
            status = "PASSED" if passed else "FAILED"
            logger.info(f"{test_name:15} : {status}")
            if not passed:
                all_passed = False
        
        logger.info("="*50)
        if all_passed:
            logger.info("ALL TESTS PASSED! ✅")
        else:
            logger.error("SOME TESTS FAILED! ❌")
        
        return results

def main():
    """Main test function"""
    try:
        tester = AnalysisTester()
        results = tester.run_all_tests()
        
        # Exit with appropriate code
        if all(results.values()):
            sys.exit(0)
        else:
            sys.exit(1)
            
    except Exception as e:
        logger.error(f"Test execution failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()