test_reply_behavior.py

#!/usr/bin/env python3
"""
Test the new BOT_REPLY_BEHAVIOR configuration option.

This test validates:
1. Configuration accepts valid reply behavior values (ignore, mention, watch)
2. Configuration rejects invalid reply behavior values 
3. Default reply behavior is set correctly
4. Reply behavior is properly logged
"""

import os
import sys
import tempfile
from pathlib import Path
from unittest.mock import MagicMock

# Ensure the src directory is in Python path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))

def test_reply_behavior_config():
    """Test reply behavior configuration validation and defaults."""
    print("=== Testing Reply Behavior Configuration ===")
    
    # Test valid reply behaviors
    valid_behaviors = ["ignore", "mention", "watch"]
    
    for behavior in valid_behaviors:
        print(f"Testing valid behavior: {behavior}")
        
        # Set test environment
        os.environ["MATRIX_HOMESERVER_URL"] = "https://matrix.test.org"
        os.environ["MATRIX_USER_ID"] = "@testbot:matrix.test.org"
        os.environ["MATRIX_PASSWORD"] = "test_password"
        os.environ["LLM_API_KEY"] = "test_key"
        os.environ["BOT_REPLY_BEHAVIOR"] = behavior
        
        try:
            from config import Config
            
            # Mock logging to capture output
            import logging
            
            # Create a string buffer to capture log messages
            log_output = []
            original_info = logging.Logger.info
            
            def mock_info(self, msg, *args, **kwargs):
                log_output.append(msg % args if args else msg)
                original_info(self, msg, *args, **kwargs)
            
            logging.Logger.info = mock_info
            
            config = Config()
            
            # Restore original logging
            logging.Logger.info = original_info
            
            assert config.bot_reply_behavior == behavior, f"Expected {behavior}, got {config.bot_reply_behavior}"
            
            # Check that reply behavior is logged
            behavior_logged = any(f"Bot reply behavior: {behavior}" in log for log in log_output)
            assert behavior_logged, f"Reply behavior {behavior} was not logged"
            
            print(f"✓ Valid behavior '{behavior}' accepted and logged")
            
        except Exception as e:
            print(f"✗ Failed for valid behavior '{behavior}': {e}")
            return False
        finally:
            # Clean up environment
            if "BOT_REPLY_BEHAVIOR" in os.environ:
                del os.environ["BOT_REPLY_BEHAVIOR"]
    
    # Test invalid reply behavior
    print("Testing invalid behavior: invalid_mode")
    os.environ["BOT_REPLY_BEHAVIOR"] = "invalid_mode"
    
    try:
        from config import Config
        config = Config()
        print("✗ Invalid behavior should have been rejected")
        return False
    except ValueError as e:
        if "Invalid BOT_REPLY_BEHAVIOR" in str(e):
            print("✓ Invalid behavior correctly rejected")
        else:
            print(f"✗ Wrong error message: {e}")
            return False
    except Exception as e:
        print(f"✗ Unexpected error: {e}")
        return False
    finally:
        del os.environ["BOT_REPLY_BEHAVIOR"]
    
    # Test default behavior
    print("Testing default behavior")
    try:
        from config import Config
        
        # Reload the module to clear any cached imports
        if 'config' in sys.modules:
            del sys.modules['config']
        from config import Config
        
        config = Config()
        assert config.bot_reply_behavior == "mention", f"Expected default 'mention', got {config.bot_reply_behavior}"
        print("✓ Default behavior is 'mention'")
        
    except Exception as e:
        print(f"✗ Failed to test default behavior: {e}")
        return False
    
    print("🎉 All reply behavior configuration tests passed!")
    return True


def test_clean_reply_content():
    """Test the _clean_reply_content method."""
    print("\n=== Testing Reply Content Cleaning ===")
    
    # Mock setup for testing the method
    import importlib.util
    import sys
    from pathlib import Path
    
    # Load bot module with absolute import
    bot_path = Path(__file__).parent / 'src' / 'bot.py'
    spec = importlib.util.spec_from_file_location("bot", bot_path)
    bot_module = importlib.util.module_from_spec(spec)
    
    # Add the src directory to sys.path temporarily
    src_path = str(Path(__file__).parent / 'src')
    if src_path not in sys.path:
        sys.path.insert(0, src_path)
    
    try:
        spec.loader.exec_module(bot_module)
        
        # Create a minimal config for testing
        config_path = Path(__file__).parent / 'src' / 'config.py' 
        config_spec = importlib.util.spec_from_file_location("config", config_path)
        config_module = importlib.util.module_from_spec(config_spec)
        config_spec.loader.exec_module(config_module)
        
        config = config_module.Config()
        bot = bot_module.AskaosusBot(config)
        
    except Exception as e:
        print(f"✗ Failed to setup test environment: {e}")
        # Test the cleaning logic directly without the bot class
        return test_clean_reply_content_standalone()
    finally:
        # Remove the src path from sys.path
        if src_path in sys.path:
            sys.path.remove(src_path)
    
    # Test cases for content cleaning
    test_cases = [
        {
            "input": "@askaosus how do I install Ubuntu?",
            "expected": "how do I install Ubuntu?",
            "description": "Remove simple mention"
        },
        {
            "input": "> Original message here\n@askaosus what about this?",
            "expected": "what about this?",
            "description": "Remove quote lines and mention"
        },
        {
            "input": "askaosus please help with this issue",
            "expected": "please help with this issue",
            "description": "Remove mention without @ symbol"
        },
        {
            "input": "   @askaosus   help   me   please   ",
            "expected": "help me please",
            "description": "Clean up extra whitespace"
        },
        {
            "input": "> Quote line 1\n> Quote line 2\nActual question here @askaosus",
            "expected": "Actual question here",
            "description": "Remove multiple quote lines and mention"
        }
    ]
    
    bot_mentions = ["@askaosus", "askaosus"]
    
    for i, test_case in enumerate(test_cases):
        print(f"Test {i+1}: {test_case['description']}")
        
        try:
            result = bot._clean_reply_content(test_case["input"], bot_mentions)
            if result == test_case["expected"]:
                print(f"✓ Input: '{test_case['input']}' -> Output: '{result}'")
            else:
                print(f"✗ Expected: '{test_case['expected']}', Got: '{result}'")
                return False
        except Exception as e:
            print(f"✗ Error cleaning content: {e}")
            return False
    
    print("🎉 All content cleaning tests passed!")
    return True


def test_clean_reply_content_standalone():
    """Test content cleaning with standalone implementation."""
    import re
    
    def clean_reply_content(message_body: str, bot_mentions: list) -> str:
        """Standalone version of the cleaning function for testing."""
        cleaned = message_body
        
        # Remove bot mentions - need to handle @ symbols separately
        for mention in bot_mentions:
            if mention.startswith('@'):
                # For @mentions, remove the whole word
                cleaned = re.sub(rf"@{re.escape(mention[1:])}\b", "", cleaned, flags=re.IGNORECASE)
            # Also handle the mention without @ in case it's in the list
            cleaned = re.sub(rf"\b{re.escape(mention)}\b", "", cleaned, flags=re.IGNORECASE)
        
        # Remove common Matrix reply prefixes (fallback formatting)
        lines = cleaned.split('\n')
        non_quote_lines = []
        for line in lines:
            if not line.strip().startswith('> '):
                non_quote_lines.append(line)
        
        cleaned = '\n'.join(non_quote_lines).strip()
        
        # Remove extra whitespace
        cleaned = re.sub(r'\s+', ' ', cleaned).strip()
        
        return cleaned
    
    print("Testing with standalone implementation...")
    
    test_cases = [
        {
            "input": "@askaosus how do I install Ubuntu?",
            "expected": "how do I install Ubuntu?",
            "description": "Remove simple mention"
        },
        {
            "input": "> Original message here\n@askaosus what about this?",
            "expected": "what about this?",
            "description": "Remove quote lines and mention"
        },
        {
            "input": "askaosus please help with this issue",
            "expected": "please help with this issue",
            "description": "Remove mention without @ symbol"
        },
        {
            "input": "   @askaosus   help   me   please   ",
            "expected": "help me please",
            "description": "Clean up extra whitespace"
        },
        {
            "input": "> Quote line 1\n> Quote line 2\nActual question here @askaosus",
            "expected": "Actual question here",
            "description": "Remove multiple quote lines and mention"
        }
    ]
    
    bot_mentions = ["@askaosus", "askaosus"]
    
    for i, test_case in enumerate(test_cases):
        print(f"Test {i+1}: {test_case['description']}")
        
        try:
            result = clean_reply_content(test_case["input"], bot_mentions)
            if result == test_case["expected"]:
                print(f"✓ Input: '{test_case['input']}' -> Output: '{result}'")
            else:
                print(f"✗ Expected: '{test_case['expected']}', Got: '{result}'")
                return False
        except Exception as e:
            print(f"✗ Error cleaning content: {e}")
            return False
    
    print("🎉 All content cleaning tests passed!")
    return True


def main():
    """Run all reply behavior tests."""
    print("Testing BOT_REPLY_BEHAVIOR configuration and functionality...\n")
    
    # Clean up environment first
    env_vars_to_clean = [
        "MATRIX_HOMESERVER_URL", "MATRIX_USER_ID", "MATRIX_PASSWORD", 
        "LLM_API_KEY", "BOT_REPLY_BEHAVIOR"
    ]
    for var in env_vars_to_clean:
        if var in os.environ:
            del os.environ[var]
    
    success = True
    
    # Test configuration
    if not test_reply_behavior_config():
        success = False
    
    # Test content cleaning 
    if not test_clean_reply_content():
        success = False
    
    if success:
        print("\n🎉 All reply behavior tests passed!")
        return 0
    else:
        print("\n❌ Some tests failed!")
        return 1


if __name__ == "__main__":
    sys.exit(main())