scripts/test_email_matching.py

#!/usr/bin/env python3
"""
Test suite for verifying that kpi_correlation_app correctly matches emails
and retrieves corresponding values from KPI and scores CSV files.
"""

import os
import sys
import pandas as pd
import numpy as np
import tempfile
import unittest
from io import StringIO

# Add the current directory to Python path for imports
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from kpi_correlation_app import analyze_correlations, convert_percentage_to_numeric
from csv_utils import robust_csv_loader, find_email_column, find_ipm_columns


class TestEmailMatching(unittest.TestCase):
    """Test suite for email matching and value correspondence"""
    
    def setUp(self):
        """Set up test data files"""
        self.test_dir = tempfile.mkdtemp()
        
        # Create test scores data
        self.scores_data = pd.DataFrame({
            'email': [
                'john.doe@company.com',
                'jane.smith@company.com',
                'MIKE.JONES@company.com',  # Different case
                'sarah.williams@company.com',
                'tom.brown@company.com'
            ],
            'problem_score': [85.5, 92.0, 78.5, 88.0, 90.5],
            'ability_score': [4.2, 4.8, 3.9, 4.5, 4.6]
        })
        
        # Create test KPI data with known values
        self.kpi_data = pd.DataFrame({
            'Email': [
                'John.Doe@company.com',  # Different case than scores
                'jane.smith@company.com',  # Same case
                'mike.jones@company.com',  # Different case than scores
                'sarah.williams@company.com',
                'alice.wonder@company.com'  # Not in scores
            ],
            'FY23/24 全年IPM': ['85%', '90%', '75%', '88%', '92%'],
            'FY24/25  全年IPM': ['87%', '93%', '78%', '89%', '94%'],
            'Department': ['Sales', 'Marketing', 'IT', 'HR', 'Finance']
        })
        
        # Save test files
        self.scores_file = os.path.join(self.test_dir, 'lenovo-scores-0603.csv')
        self.kpi_file = os.path.join(self.test_dir, 'test_kpi.csv')
        
        self.scores_data.to_csv(self.scores_file, index=False)
        self.kpi_data.to_csv(self.kpi_file, index=False)
        
    def tearDown(self):
        """Clean up test files"""
        import shutil
        if os.path.exists(self.test_dir):
            shutil.rmtree(self.test_dir)
    
    def test_email_case_insensitive_matching(self):
        """Test that emails are matched case-insensitively"""
        # Temporarily change directory to test directory
        original_dir = os.getcwd()
        os.chdir(self.test_dir)
        
        try:
            # Analyze correlations
            results_text, _, _, _, _, correlation_results = analyze_correlations(self.kpi_file)
            
            # Check that case differences don't prevent matching
            self.assertIn("Matched emails: 4 records", results_text)
            
            # Verify that john.doe, jane.smith, mike.jones, and sarah.williams were matched
            # (alice.wonder and tom.brown should not match)
            
        finally:
            os.chdir(original_dir)
    
    def test_correct_value_correspondence(self):
        """Test that correct values are retrieved for matched emails"""
        # Create a simplified version for direct testing
        scores_df = self.scores_data.copy()
        kpi_df = self.kpi_data.copy()
        
        # Normalize emails
        scores_df['email_normalized'] = scores_df['email'].str.lower()
        kpi_df['email_normalized'] = kpi_df['Email'].str.lower()
        
        # Merge datasets
        merged_df = pd.merge(scores_df, kpi_df, on='email_normalized', how='inner')
        
        # Test specific email correspondences
        john_row = merged_df[merged_df['email_normalized'] == 'john.doe@company.com']
        self.assertEqual(len(john_row), 1)
        self.assertAlmostEqual(john_row['problem_score'].iloc[0], 85.5)
        self.assertAlmostEqual(john_row['ability_score'].iloc[0], 4.2)
        self.assertEqual(john_row['FY23/24 全年IPM'].iloc[0], '85%')
        self.assertEqual(john_row['FY24/25  全年IPM'].iloc[0], '87%')
        
        # Test another email
        jane_row = merged_df[merged_df['email_normalized'] == 'jane.smith@company.com']
        self.assertEqual(len(jane_row), 1)
        self.assertAlmostEqual(jane_row['problem_score'].iloc[0], 92.0)
        self.assertAlmostEqual(jane_row['ability_score'].iloc[0], 4.8)
        self.assertEqual(jane_row['FY23/24 全年IPM'].iloc[0], '90%')
        self.assertEqual(jane_row['FY24/25  全年IPM'].iloc[0], '93%')
        
    def test_percentage_conversion(self):
        """Test that percentage strings are correctly converted to numeric values"""
        # Test conversion function
        test_series = pd.Series(['85%', '90.5%', '100%', '0%', 'invalid'])
        converted = convert_percentage_to_numeric(test_series)
        
        self.assertAlmostEqual(converted.iloc[0], 0.85)
        self.assertAlmostEqual(converted.iloc[1], 0.905)
        self.assertAlmostEqual(converted.iloc[2], 1.0)
        self.assertAlmostEqual(converted.iloc[3], 0.0)
        self.assertTrue(pd.isna(converted.iloc[4]))  # 'invalid' should be NaN
        
    def test_missing_emails_handling(self):
        """Test that non-matching emails are handled correctly"""
        scores_df = self.scores_data.copy()
        kpi_df = self.kpi_data.copy()
        
        # Normalize emails
        scores_df['email_normalized'] = scores_df['email'].str.lower()
        kpi_df['email_normalized'] = kpi_df['Email'].str.lower()
        
        # Merge datasets
        merged_df = pd.merge(scores_df, kpi_df, on='email_normalized', how='inner')
        
        # tom.brown@company.com is in scores but not in KPI
        self.assertFalse('tom.brown@company.com' in merged_df['email_normalized'].values)
        
        # alice.wonder@company.com is in KPI but not in scores
        self.assertFalse('alice.wonder@company.com' in merged_df['email_normalized'].values)
        
        # Only 4 emails should match
        self.assertEqual(len(merged_df), 4)
        
    def test_excel_file_support(self):
        """Test that Excel files are correctly processed"""
        # Save KPI data as Excel
        excel_file = os.path.join(self.test_dir, 'test_kpi.xlsx')
        self.kpi_data.to_excel(excel_file, index=False)
        
        # Test loading Excel file
        loaded_df = robust_csv_loader(excel_file)
        
        # Verify data integrity
        self.assertEqual(len(loaded_df), len(self.kpi_data))
        self.assertListEqual(loaded_df.columns.tolist(), self.kpi_data.columns.tolist())
        
    def test_correlation_calculation_integrity(self):
        """Test that correlations are calculated on correctly matched data"""
        # Create specific test data with known correlations
        test_scores = pd.DataFrame({
            'email': ['user1@test.com', 'user2@test.com', 'user3@test.com', 'user4@test.com', 'user5@test.com'],
            'problem_score': [10, 20, 30, 40, 50],
            'ability_score': [1, 2, 3, 4, 5]
        })
        
        test_kpi = pd.DataFrame({
            'Email': ['user1@test.com', 'user2@test.com', 'user3@test.com', 'user4@test.com', 'user5@test.com'],
            'FY23/24 全年IPM': ['20%', '40%', '60%', '80%', '100%'],  # Perfect correlation
            'FY24/25  全年IPM': ['100%', '80%', '60%', '40%', '20%']   # Perfect negative correlation
        })
        
        # Save test files
        test_scores_file = os.path.join(self.test_dir, 'lenovo-scores-0603.csv')
        test_kpi_file = os.path.join(self.test_dir, 'perfect_correlation_kpi.csv')
        
        test_scores.to_csv(test_scores_file, index=False)
        test_kpi.to_csv(test_kpi_file, index=False)
        
        # Change to test directory and analyze
        original_dir = os.getcwd()
        os.chdir(self.test_dir)
        
        try:
            results_text, _, _, _, _, correlation_results = analyze_correlations(test_kpi_file)
            
            # Check AC correlation (problem_score vs FY23/24 IPM) - should be perfect positive
            if correlation_results and 'AC' in correlation_results:
                self.assertAlmostEqual(correlation_results['AC']['pearson'], 1.0, places=5)
                
            # Check AD correlation (problem_score vs FY24/25 IPM) - should be perfect negative
            if correlation_results and 'AD' in correlation_results:
                self.assertAlmostEqual(correlation_results['AD']['pearson'], -1.0, places=5)
                
        finally:
            os.chdir(original_dir)


class TestColumnFinding(unittest.TestCase):
    """Test suite for column finding utilities"""
    
    def test_find_email_column_variations(self):
        """Test finding email column with various naming conventions"""
        test_cases = [
            pd.DataFrame({'Email': [], 'Name': []}),
            pd.DataFrame({'email': [], 'Name': []}),
            pd.DataFrame({'EMAIL': [], 'Name': []}),
            pd.DataFrame({'Email Address': [], 'Name': []}),
            pd.DataFrame({'user_email': [], 'Name': []})
        ]
        
        for df in test_cases:
            email_col = find_email_column(df)
            self.assertIsNotNone(email_col, f"Failed to find email column in {df.columns.tolist()}")
            self.assertTrue('email' in email_col.lower())
    
    def test_find_ipm_columns(self):
        """Test finding IPM columns with exact naming"""
        df = pd.DataFrame({
            'Email': [],
            'FY23/24 全年IPM': [],
            'FY24/25  全年IPM': [],  # Note the extra space
            'Other Column': []
        })
        
        fy2425_col, fy2324_col = find_ipm_columns(df)
        
        self.assertEqual(fy2324_col, 'FY23/24 全年IPM')
        self.assertEqual(fy2425_col, 'FY24/25  全年IPM')


def run_detailed_tests():
    """Run tests with detailed output"""
    # Create a test suite
    loader = unittest.TestLoader()
    suite = unittest.TestSuite()
    
    # Add all test classes
    suite.addTests(loader.loadTestsFromTestCase(TestEmailMatching))
    suite.addTests(loader.loadTestsFromTestCase(TestColumnFinding))
    
    # Run with verbose output
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(suite)
    
    return result.wasSuccessful()


if __name__ == '__main__':
    print("=" * 70)
    print("Running Email Matching and Value Correspondence Tests")
    print("=" * 70)
    
    success = run_detailed_tests()
    
    if success:
        print("\n✅ All tests passed! The app correctly matches emails and retrieves corresponding values.")
    else:
        print("\n❌ Some tests failed. Please review the output above.")
    
    sys.exit(0 if success else 1)