tests/unit/test_authentication.py

"""
Unit tests for user authentication functionality

Consolidated from:
- test_user_authentication_comprehensive.py (unit test portions)
- test_user_id_validation.py
- Parts of test_chat_integration_user_auth.py (validation portions)
"""

import asyncio

from app import normalize_user_id, validate_user_id
import pytest

from analytics.collectors import create_session, track_message, track_search
from analytics.models import Message, SearchAnalytics, Session
from tests.utilities import MockDataGenerator, TestHelpers, ValidationHelpers, empty_user_ids, invalid_user_ids, valid_user_ids

    TestHelpers, ValidationHelpers, MockDataGenerator,
    valid_user_ids, invalid_user_ids, empty_user_ids
)


class TestUserIdValidation:
    """Test user_id validation functions and model validation"""
    
    def test_normalize_user_id_function(self):
        """Test normalize_user_id function handles various inputs correctly"""
        # Test None input
        assert normalize_user_id(None) is None
        
        # Test empty string
        assert normalize_user_id("") is None
        
        # Test whitespace-only strings
        assert normalize_user_id("   ") is None
        assert normalize_user_id("\t") is None
        assert normalize_user_id("\n") is None
        assert normalize_user_id("  \t\n  ") is None
        
        # Test valid user_id
        assert normalize_user_id("user123") == "user123"
        assert normalize_user_id("  user123  ") == "user123"
    
    def test_validate_user_id_function(self):
        """Test validate_user_id function properly handles all cases"""
        # Test None input (anonymous)
        assert validate_user_id(None) is None
        
        # Test empty string (anonymous)
        assert validate_user_id("") is None
        assert validate_user_id("   ") is None
        
        # Test valid user_id
        assert validate_user_id("user123") == "user123"
        assert validate_user_id("user_123") == "user_123"
        assert validate_user_id("user-123") == "user-123"
        
        # Test invalid user_id raises exception
        with pytest.raises(Exception):
            validate_user_id("user@123")
        
        with pytest.raises(Exception):
            validate_user_id("a" * 256)  # Too long
    
    def test_validate_user_id_with_fixtures(self, valid_user_ids, invalid_user_ids, empty_user_ids):
        """Test validate_user_id with fixture data"""
        # Test valid user IDs
        for user_id in valid_user_ids:
            result = validate_user_id(user_id)
            assert result == user_id.strip()
        
        # Test invalid user IDs
        for user_id in invalid_user_ids:
            with pytest.raises(Exception):
                validate_user_id(user_id)
        
        # Test empty user IDs (should return None)
        for user_id in empty_user_ids:
            result = validate_user_id(user_id)
            assert result is None


class TestSessionModelValidation:
    """Test Session model user_id validation"""
    
    def test_session_valid_user_id(self):
        """Test Session model with valid user_id values"""
        # Valid user_id
        session = Session(user_id="user_123")
        assert session.user_id == "user_123"
        
        # Valid user_id with hyphens and underscores
        session = Session(user_id="user-123_test")
        assert session.user_id == "user-123_test"
        
        # None user_id (anonymous)
        session = Session(user_id=None)
        assert session.user_id is None
        
        # Empty string becomes None
        session = Session(user_id="")
        assert session.user_id is None
        
        # Whitespace-only string becomes None
        session = Session(user_id="   ")
        assert session.user_id is None
    
    def test_session_invalid_user_id(self):
        """Test Session model with invalid user_id values"""
        # Non-string user_id
        with pytest.raises(Exception) as exc_info:
            Session(user_id=123)
        assert "string" in str(exc_info.value).lower()
        
        # Too long user_id
        with pytest.raises(Exception) as exc_info:
            Session(user_id="a" * 256)
        assert "255" in str(exc_info.value)
        
        # Invalid characters
        invalid_ids = ["user@123", "user 123", "user.123"]
        for invalid_id in invalid_ids:
            with pytest.raises(Exception) as exc_info:
                Session(user_id=invalid_id)
            assert "alphanumeric" in str(exc_info.value).lower()
    
    def test_session_to_dict_includes_user_id(self):
        """Test that Session.to_dict() includes user_id field"""
        # Session with user_id
        session = Session(user_id="user_123")
        session_dict = session.to_dict()
        assert "user_id" in session_dict
        assert session_dict["user_id"] == "user_123"
        
        # Anonymous session
        session = Session(user_id=None)
        session_dict = session.to_dict()
        assert "user_id" in session_dict
        assert session_dict["user_id"] is None


class TestMessageModelValidation:
    """Test Message model user_id validation"""
    
    def test_message_valid_user_id(self):
        """Test Message model with valid user_id values"""
        # Valid user_id
        message = Message(
            session_id="test_session",
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_123"
        )
        assert message.user_id == "user_123"
        
        # None user_id (anonymous)
        message = Message(
            session_id="test_session",
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id=None
        )
        assert message.user_id is None
    
    def test_message_invalid_user_id(self):
        """Test Message model with invalid user_id values"""
        # Non-string user_id
        with pytest.raises(Exception) as exc_info:
            Message(
                session_id="test_session",
                prompt_length=50,
                response_length=100,
                response_time_ms=1000,
                user_id=123
            )
        assert "string" in str(exc_info.value).lower()
        
        # Invalid characters
        with pytest.raises(Exception):
            Message(
                session_id="test_session",
                prompt_length=50,
                response_length=100,
                response_time_ms=1000,
                user_id="user@123"
            )
    
    def test_message_to_dict_includes_user_id(self):
        """Test that Message.to_dict() includes user_id field"""
        message = Message(
            session_id="test_session",
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_123"
        )
        message_dict = message.to_dict()
        assert "user_id" in message_dict
        assert message_dict["user_id"] == "user_123"


class TestSearchAnalyticsModelValidation:
    """Test SearchAnalytics model user_id validation"""
    
    def test_search_analytics_valid_user_id(self):
        """Test SearchAnalytics model with valid user_id values"""
        # Valid user_id
        search = SearchAnalytics(
            message_id="test_message",
            search_query="test query",
            user_id="user_123"
        )
        assert search.user_id == "user_123"
        
        # None user_id (anonymous)
        search = SearchAnalytics(
            message_id="test_message",
            search_query="test query",
            user_id=None
        )
        assert search.user_id is None
    
    def test_search_analytics_invalid_user_id(self):
        """Test SearchAnalytics model with invalid user_id values"""
        # Non-string user_id
        with pytest.raises(Exception) as exc_info:
            SearchAnalytics(
                message_id="test_message",
                search_query="test query",
                user_id=123
            )
        assert "string" in str(exc_info.value).lower()
        
        # Invalid characters
        with pytest.raises(Exception):
            SearchAnalytics(
                message_id="test_message",
                search_query="test query",
                user_id="user@123"
            )
    
    def test_search_analytics_to_dict_includes_user_id(self):
        """Test that SearchAnalytics.to_dict() includes user_id field"""
        search = SearchAnalytics(
            message_id="test_message",
            search_query="test query",
            user_id="user_123"
        )
        search_dict = search.to_dict()
        assert "user_id" in search_dict
        assert search_dict["user_id"] == "user_123"


class TestAnalyticsCollectors:
    """Test analytics collectors with user_id support"""
    
    @pytest.mark.asyncio
    async def test_create_session_with_user_id(self):
        """Test create_session function with user_id"""
        # Create session with user_id
        session = await create_session(user_agent="TestAgent", user_id="user_123")
        assert session.user_id == "user_123"
        assert session.user_agent == "TestAgent"
        assert session.session_id is not None
        
        # Create anonymous session
        session = await create_session(user_agent="TestAgent", user_id=None)
        assert session.user_id is None
        
        # Create session without user_id parameter
        session = await create_session(user_agent="TestAgent")
        assert session.user_id is None
    
    @pytest.mark.asyncio
    async def test_track_message_with_user_id(self):
        """Test track_message function with user_id"""
        # Create a session first
        session = await create_session(user_id="user_123")
        
        # Track message with user_id
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_123"
        )
        
        assert message is not None
        assert message.user_id == "user_123"
        assert message.session_id == session.session_id
        assert message.prompt_length == 50
        assert message.response_length == 100
        assert message.response_time_ms == 1000
    
    @pytest.mark.asyncio
    async def test_track_message_user_id_mismatch(self):
        """Test that user_id mismatch between session and message is handled"""
        # Create a session with one user_id
        session = await create_session(user_id="user_123")
        
        # Track message with different user_id
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_456"  # Different user_id
        )
        
        assert message is not None
        assert message.user_id == "user_456"  # Message should use provided user_id
        assert message.session_id == session.session_id
    
    @pytest.mark.asyncio
    async def test_track_search_with_user_id(self):
        """Test track_search function with user_id"""
        # Create session and message first
        session = await create_session(user_id="user_123")
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_123"
        )
        
        # Track search with user_id
        search = await track_search(
            message_id=message.message_id,
            search_query="test query",
            search_terms=["test", "query"],
            brave_results=5,
            duckduckgo_results=3,
            total_unique_results=7,
            user_id="user_123"
        )
        
        assert search is not None
        assert search.user_id == "user_123"
        assert search.message_id == message.message_id
        assert search.search_query == "test query"
    
    @pytest.mark.asyncio
    async def test_track_search_user_id_mismatch(self):
        """Test track_search with user_id mismatch"""
        # Create session and message with one user_id
        session = await create_session(user_id="user_123")
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id="user_123"
        )
        
        # Track search with different user_id
        search = await track_search(
            message_id=message.message_id,
            search_query="test query",
            search_terms=["test", "query"],
            user_id="user_456"  # Different user_id
        )
        
        assert search is not None
        assert search.user_id == "user_456"  # Should use provided user_id
        assert search.message_id == message.message_id


class TestBackwardCompatibility:
    """Test backward compatibility for anonymous users"""
    
    @pytest.mark.asyncio
    async def test_create_session_without_user_id_parameter(self):
        """Test creating sessions without user_id parameter (old way)"""
        # Create session the old way (no user_id parameter)
        session = await create_session(user_agent="TestAgent")
        
        assert session.user_id is None
        assert session.user_agent == "TestAgent"
        assert session.session_id is not None
        assert session.status == "active"
    
    @pytest.mark.asyncio
    async def test_track_message_without_user_id_parameter(self):
        """Test tracking messages without user_id parameter (old way)"""
        # Create session first
        session = await create_session(user_agent="TestAgent")
        
        # Track message the old way (no user_id parameter)
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            used_search=True,
            max_tokens=500,
            temperature=0.7,
            success=True
        )
        
        assert message is not None
        assert message.user_id is None
        assert message.session_id == session.session_id
        assert message.prompt_length == 50
        assert message.used_search is True
    
    @pytest.mark.asyncio
    async def test_track_search_without_user_id_parameter(self):
        """Test tracking search without user_id parameter (old way)"""
        # Create session and message first
        session = await create_session()
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000
        )
        
        # Track search the old way (no user_id parameter)
        search = await track_search(
            message_id=message.message_id,
            search_query="test query",
            search_terms=["test", "query"],
            brave_results=5,
            duckduckgo_results=3,
            total_unique_results=7
        )
        
        assert search is not None
        assert search.user_id is None
        assert search.message_id == message.message_id
        assert search.search_query == "test query"


class TestAuthenticationEdgeCases:
    """Test edge cases and error conditions"""
    
    def test_user_id_length_limits(self):
        """Test user_id length validation"""
        # Test maximum valid length
        max_valid = "a" * 255
        session = Session(user_id=max_valid)
        assert session.user_id == max_valid
        
        # Test too long
        too_long = "a" * 256
        with pytest.raises(Exception) as exc_info:
            Session(user_id=too_long)
        assert "255" in str(exc_info.value)
    
    def test_user_id_character_validation(self):
        """Test user_id character validation"""
        # Valid characters
        valid_chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-"
        session = Session(user_id=valid_chars)
        assert session.user_id == valid_chars
        
        # Invalid characters
        invalid_chars = ["@", " ", ".", "#", "$", "%", "&", "*", "+", "=", "[", "]", "{", "}", "|", "\\", "/", ":", ";", "<", ">", "?", ",", "'", '"', "`", "~", "!"]
        
        for char in invalid_chars:
            user_id_with_invalid = f"user{char}123"
            with pytest.raises(Exception):
                Session(user_id=user_id_with_invalid)
    
    def test_user_id_type_validation(self):
        """Test user_id type validation"""
        # Valid types
        Session(user_id="string_user_id")
        Session(user_id=None)
        
        # Invalid types
        invalid_types = [123, 45.67, True, [], {}, set()]
        
        for invalid_type in invalid_types:
            with pytest.raises(Exception):
                Session(user_id=invalid_type)
    
    @pytest.mark.asyncio
    async def test_collectors_with_invalid_user_id(self):
        """Test that collectors handle invalid user_id gracefully"""
        # Test create_session with invalid user_id
        with pytest.raises(Exception):
            await create_session(user_id="invalid@user")
        
        # Test track_message with invalid user_id
        session = await create_session(user_id="valid_user")
        with pytest.raises(Exception):
            await track_message(
                session_id=session.session_id,
                prompt_length=50,
                response_length=100,
                response_time_ms=1000,
                user_id="invalid@user"
            )


class TestAuthenticationDataConsistency:
    """Test data consistency across authentication features"""
    
    @pytest.mark.asyncio
    async def test_session_message_user_id_consistency(self):
        """Test that session and message user_id values are consistent"""
        user_id = "consistency_test_user"
        
        # Create session
        session = await create_session(user_id=user_id)
        assert session.user_id == user_id
        
        # Create message with same user_id
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id=user_id
        )
        assert message.user_id == user_id
        assert message.session_id == session.session_id
        
        # Verify data persistence
        await TestHelpers.wait_for_data_persistence()
        
        # Verify session data
        session_valid = await TestHelpers.verify_session_data(
            session.session_id, user_id, expected_messages=1
        )
        assert session_valid
        
        # Verify message data
        message_valid = await TestHelpers.verify_message_data(
            message.message_id, user_id
        )
        assert message_valid
    
    @pytest.mark.asyncio
    async def test_anonymous_data_consistency(self):
        """Test that anonymous data is consistently handled"""
        # Create anonymous session
        session = await create_session(user_id=None)
        assert session.user_id is None
        
        # Create anonymous message
        message = await track_message(
            session_id=session.session_id,
            prompt_length=50,
            response_length=100,
            response_time_ms=1000,
            user_id=None
        )
        assert message.user_id is None
        assert message.session_id == session.session_id
        
        # Verify data persistence
        await TestHelpers.wait_for_data_persistence()
        
        # Verify session data
        session_valid = await TestHelpers.verify_session_data(
            session.session_id, None, expected_messages=1
        )
        assert session_valid
        
        # Verify message data
        message_valid = await TestHelpers.verify_message_data(
            message.message_id, None
        )
        assert message_valid


if __name__ == "__main__":
    # Run tests manually for debugging
    async def run_basic_tests():
        test_validation = TestUserIdValidation()
        test_validation.test_normalize_user_id_function()
        test_validation.test_validate_user_id_function()
        print("✅ User ID validation tests passed")
        
        test_session = TestSessionModelValidation()
        test_session.test_session_valid_user_id()
        test_session.test_session_to_dict_includes_user_id()
        print("✅ Session model validation tests passed")
        
        test_collectors = TestAnalyticsCollectors()
        await test_collectors.test_create_session_with_user_id()
        await test_collectors.test_track_message_with_user_id()
        print("✅ Analytics collectors tests passed")
    
    asyncio.run(run_basic_tests())