Test/test_auth_unit.py

"""
Unit tests for User model and authentication utilities
Tests user creation, password hashing, validation methods, authentication decorators,
utility functions, and MongoDB integration with error handling.
"""

import unittest
from unittest.mock import Mock, patch, MagicMock
from datetime import datetime
from bson import ObjectId
from werkzeug.security import check_password_hash
from flask import Flask
from flask_login import current_user

# Import modules to test
from models import User, ChatSession
from auth import (
    validate_email_format, validate_password_format, validate_registration_data,
    validate_login_data, authenticate_user, create_user_account, login_required,
    get_current_user, is_safe_url, get_redirect_target
)


class TestUserModel(unittest.TestCase):
    """Test cases for User model"""
    
    def setUp(self):
        """Set up test fixtures"""
        self.sample_user_data = {
            '_id': ObjectId(),
            'email': 'test@example.com',
            'password_hash': 'hashed_password',
            'created_at': datetime.utcnow(),
            'is_active': True
        }
        
        self.test_user = User(self.sample_user_data)
    
    def test_user_initialization(self):
        """Test User object initialization from MongoDB document"""
        user = User(self.sample_user_data)
        
        self.assertEqual(user.id, str(self.sample_user_data['_id']))
        self.assertEqual(user.email, self.sample_user_data['email'])
        self.assertEqual(user.password_hash, self.sample_user_data['password_hash'])
        self.assertEqual(user.created_at, self.sample_user_data['created_at'])
        self.assertTrue(user.is_active)
    
    def test_user_initialization_with_defaults(self):
        """Test User initialization with missing optional fields"""
        minimal_data = {
            '_id': ObjectId(),
            'email': 'test@example.com',
            'password_hash': 'hashed_password'
        }
        
        user = User(minimal_data)
        
        self.assertEqual(user.email, 'test@example.com')
        self.assertTrue(user.is_active)  # Default value
        self.assertIsInstance(user.created_at, datetime)
    
    def test_get_id(self):
        """Test get_id method for Flask-Login compatibility"""
        user_id = self.test_user.get_id()
        self.assertEqual(user_id, str(self.sample_user_data['_id']))
        self.assertIsInstance(user_id, str)
    
    @patch('models.get_users_collection')
    def test_find_by_email_success(self, mock_get_collection):
        """Test successful user lookup by email"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.return_value = self.sample_user_data
        
        user = User.find_by_email('test@example.com')
        
        self.assertIsInstance(user, User)
        self.assertEqual(user.email, 'test@example.com')
        mock_collection.find_one.assert_called_once_with({"email": "test@example.com"})
    
    @patch('models.get_users_collection')
    def test_find_by_email_not_found(self, mock_get_collection):
        """Test user lookup by email when user doesn't exist"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.return_value = None
        
        user = User.find_by_email('nonexistent@example.com')
        
        self.assertIsNone(user)
    
    @patch('models.get_users_collection')
    def test_find_by_email_database_error(self, mock_get_collection):
        """Test user lookup by email with database error"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.side_effect = Exception("Database error")
        
        user = User.find_by_email('test@example.com')
        
        self.assertIsNone(user)
    
    @patch('models.get_users_collection')
    def test_find_by_id_success(self, mock_get_collection):
        """Test successful user lookup by ID"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.return_value = self.sample_user_data
        
        user_id = str(self.sample_user_data['_id'])
        user = User.find_by_id(user_id)
        
        self.assertIsInstance(user, User)
        self.assertEqual(user.id, user_id)
        mock_collection.find_one.assert_called_once_with({"_id": self.sample_user_data['_id']})
    
    @patch('models.get_users_collection')
    def test_find_by_id_not_found(self, mock_get_collection):
        """Test user lookup by ID when user doesn't exist"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.return_value = None
        
        user = User.find_by_id(str(ObjectId()))
        
        self.assertIsNone(user)
    
    @patch('models.get_users_collection')
    def test_find_by_id_invalid_id(self, mock_get_collection):
        """Test user lookup by ID with invalid ObjectId"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find_one.side_effect = Exception("Invalid ObjectId")
        
        user = User.find_by_id('invalid_id')
        
        self.assertIsNone(user)
    
    @patch('models.User.validate_password_strength')
    @patch('models.User.validate_email_format')
    @patch('models.User.find_by_email')
    @patch('models.get_users_collection')
    def test_create_user_success(self, mock_get_collection, mock_find_by_email, 
                                mock_validate_email, mock_validate_password):
        """Test successful user creation"""
        # Setup mocks
        mock_validate_email.return_value = True
        mock_validate_password.return_value = True
        mock_find_by_email.return_value = None  # User doesn't exist
        
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.insert_one.return_value = Mock(inserted_id=self.sample_user_data['_id'])
        mock_collection.find_one.return_value = self.sample_user_data
        
        user = User.create_user('test@example.com', 'password123')
        
        self.assertIsInstance(user, User)
        self.assertEqual(user.email, 'test@example.com')
        mock_collection.insert_one.assert_called_once()
    
    @patch('models.User.validate_email_format')
    def test_create_user_invalid_email(self, mock_validate_email):
        """Test user creation with invalid email"""
        mock_validate_email.return_value = False
        
        user = User.create_user('invalid_email', 'password123')
        
        self.assertIsNone(user)
    
    @patch('models.User.validate_password_strength')
    @patch('models.User.validate_email_format')
    def test_create_user_weak_password(self, mock_validate_email, mock_validate_password):
        """Test user creation with weak password"""
        mock_validate_email.return_value = True
        mock_validate_password.return_value = False
        
        user = User.create_user('test@example.com', 'weak')
        
        self.assertIsNone(user)
    
    @patch('models.User.validate_password_strength')
    @patch('models.User.validate_email_format')
    @patch('models.User.find_by_email')
    def test_create_user_already_exists(self, mock_find_by_email, mock_validate_email, mock_validate_password):
        """Test user creation when user already exists"""
        mock_validate_email.return_value = True
        mock_validate_password.return_value = True
        mock_find_by_email.return_value = self.test_user  # User exists
        
        user = User.create_user('test@example.com', 'password123')
        
        self.assertIsNone(user)
    
    def test_check_password_success(self):
        """Test successful password verification"""
        # Create a user with a known password hash
        from werkzeug.security import generate_password_hash
        password = 'testpassword123'
        user_data = self.sample_user_data.copy()
        user_data['password_hash'] = generate_password_hash(password)
        
        user = User(user_data)
        
        self.assertTrue(user.check_password(password))
        self.assertFalse(user.check_password('wrongpassword'))
    
    def test_check_password_error_handling(self):
        """Test password verification with corrupted hash"""
        user_data = self.sample_user_data.copy()
        user_data['password_hash'] = 'corrupted_hash'
        
        user = User(user_data)
        
        # Should return False for any password with corrupted hash
        self.assertFalse(user.check_password('anypassword'))
    
    @patch('models.get_users_collection')
    def test_set_password_success(self, mock_get_collection):
        """Test successful password update"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.update_one.return_value = Mock(modified_count=1)
        
        with patch.object(User, 'validate_password_strength', return_value=True):
            result = self.test_user.set_password('newpassword123')
        
        self.assertTrue(result)
        mock_collection.update_one.assert_called_once()
        # Verify password hash was updated
        self.assertTrue(check_password_hash(self.test_user.password_hash, 'newpassword123'))
    
    @patch('models.get_users_collection')
    def test_set_password_weak_password(self, mock_get_collection):
        """Test password update with weak password"""
        with patch.object(User, 'validate_password_strength', return_value=False):
            result = self.test_user.set_password('weak')
        
        self.assertFalse(result)
    
    @patch('models.get_users_collection')
    def test_set_password_database_error(self, mock_get_collection):
        """Test password update with database error"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.update_one.side_effect = Exception("Database error")
        
        with patch.object(User, 'validate_password_strength', return_value=True):
            result = self.test_user.set_password('newpassword123')
        
        self.assertFalse(result)
    
    def test_validate_email_format_valid(self):
        """Test email format validation with valid emails"""
        valid_emails = [
            'test@example.com',
            'user.name@domain.co.uk',
            'user+tag@example.org',
            'user123@test-domain.com'
        ]
        
        for email in valid_emails:
            with self.subTest(email=email):
                self.assertTrue(User.validate_email_format(email))
    
    def test_validate_email_format_invalid(self):
        """Test email format validation with invalid emails"""
        invalid_emails = [
            'invalid_email',
            '@example.com',
            'user@',
            'user..name@example.com',
            '',
            None
        ]
        
        for email in invalid_emails:
            with self.subTest(email=email):
                if email is not None:
                    self.assertFalse(User.validate_email_format(email))
    
    def test_validate_password_strength_valid(self):
        """Test password strength validation with valid passwords"""
        valid_passwords = [
            'password123',
            'verylongpassword',
            'P@ssw0rd!',
            '12345678'  # Minimum 8 characters
        ]
        
        for password in valid_passwords:
            with self.subTest(password=password):
                self.assertTrue(User.validate_password_strength(password))
    
    def test_validate_password_strength_invalid(self):
        """Test password strength validation with invalid passwords"""
        invalid_passwords = [
            'short',      # Too short
            '1234567',    # 7 characters
            '',           # Empty
            None          # None
        ]
        
        for password in invalid_passwords:
            with self.subTest(password=password):
                if password is not None:
                    self.assertFalse(User.validate_password_strength(password))
    
    @patch('models.get_chat_sessions_collection')
    def test_get_chat_history_success(self, mock_get_collection):
        """Test successful chat history retrieval"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        
        # Mock chat history data
        mock_sessions = [
            {
                '_id': ObjectId(),
                'message': 'Hello',
                'response': 'Hi there!',
                'timestamp': datetime.utcnow(),
                'session_data': {}
            },
            {
                '_id': ObjectId(),
                'message': 'How are you?',
                'response': 'I am doing well!',
                'timestamp': datetime.utcnow(),
                'session_data': {}
            }
        ]
        
        mock_cursor = Mock()
        mock_cursor.__iter__ = Mock(return_value=iter(mock_sessions))
        mock_collection.find.return_value.sort.return_value.limit.return_value = mock_cursor
        
        history = self.test_user.get_chat_history(limit=10)
        
        self.assertEqual(len(history), 2)
        # History is returned in chronological order (oldest first) after reversal
        self.assertEqual(history[0]['message'], 'How are you?')
        self.assertEqual(history[1]['message'], 'Hello')
    
    @patch('models.get_chat_sessions_collection')
    def test_get_chat_history_database_error(self, mock_get_collection):
        """Test chat history retrieval with database error"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find.side_effect = Exception("Database error")
        
        history = self.test_user.get_chat_history()
        
        self.assertEqual(history, [])
    
    def test_to_dict(self):
        """Test user serialization to dictionary"""
        user_dict = self.test_user.to_dict()
        
        expected_keys = {'id', 'email', 'created_at', 'is_active'}
        self.assertEqual(set(user_dict.keys()), expected_keys)
        self.assertNotIn('password_hash', user_dict)  # Should not include password
        self.assertEqual(user_dict['email'], 'test@example.com')
    
    def test_repr(self):
        """Test user string representation"""
        repr_str = repr(self.test_user)
        self.assertEqual(repr_str, '<User test@example.com>')


class TestChatSessionModel(unittest.TestCase):
    """Test cases for ChatSession model"""
    
    def setUp(self):
        """Set up test fixtures"""
        self.user_id = str(ObjectId())
        self.session_id = ObjectId()
    
    @patch('models.get_chat_sessions_collection')
    def test_save_message_success(self, mock_get_collection):
        """Test successful message saving"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.insert_one.return_value = Mock(inserted_id=self.session_id)
        
        result = ChatSession.save_message(
            user_id=self.user_id,
            message="Test message",
            response="Test response",
            session_metadata={"key": "value"}
        )
        
        self.assertTrue(result)
        mock_collection.insert_one.assert_called_once()
        
        # Verify the document structure
        call_args = mock_collection.insert_one.call_args[0][0]
        self.assertEqual(call_args['message'], "Test message")
        self.assertEqual(call_args['response'], "Test response")
        self.assertEqual(call_args['session_data'], {"key": "value"})
        self.assertIsInstance(call_args['user_id'], ObjectId)
    
    @patch('models.get_chat_sessions_collection')
    def test_save_message_database_error(self, mock_get_collection):
        """Test message saving with database error"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.insert_one.side_effect = Exception("Database error")
        
        result = ChatSession.save_message(
            user_id=self.user_id,
            message="Test message",
            response="Test response"
        )
        
        self.assertFalse(result)
    
    @patch('models.get_chat_sessions_collection')
    def test_get_user_history_success(self, mock_get_collection):
        """Test successful user history retrieval"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        
        mock_sessions = [
            {
                '_id': ObjectId(),
                'message': 'First message',
                'response': 'First response',
                'timestamp': datetime.utcnow(),
                'session_data': {}
            }
        ]
        
        mock_cursor = Mock()
        mock_cursor.__iter__ = Mock(return_value=iter(mock_sessions))
        mock_collection.find.return_value.sort.return_value.limit.return_value = mock_cursor
        
        history = ChatSession.get_user_history(self.user_id, limit=10)
        
        self.assertEqual(len(history), 1)
        self.assertEqual(history[0]['message'], 'First message')
    
    @patch('models.get_chat_sessions_collection')
    def test_get_user_history_database_error(self, mock_get_collection):
        """Test user history retrieval with database error"""
        mock_collection = Mock()
        mock_get_collection.return_value = mock_collection
        mock_collection.find.side_effect = Exception("Database error")
        
        history = ChatSession.get_user_history(self.user_id)
        
        self.assertEqual(history, [])
    
    @patch('models.ChatSession.get_user_history')
    def test_get_session_context_success(self, mock_get_history):
        """Test successful session context retrieval"""
        mock_get_history.return_value = [
            {
                'message': 'Hello',
                'response': 'Hi there!'
            },
            {
                'message': 'How are you?',
                'response': 'I am doing well!'
            }
        ]
        
        context = ChatSession.get_session_context(self.user_id, limit=2)
        
        self.assertEqual(len(context), 2)
        self.assertEqual(context[0]['user'], 'Hello')
        self.assertEqual(context[0]['assistant'], 'Hi there!')
    
    @patch('models.ChatSession.get_user_history')
    def test_get_session_context_error(self, mock_get_history):
        """Test session context retrieval with error"""
        mock_get_history.side_effect = Exception("Database error")
        
        context = ChatSession.get_session_context(self.user_id)
        
        self.assertEqual(context, [])


class TestAuthenticationUtilities(unittest.TestCase):
    """Test cases for authentication utility functions"""
    
    def test_validate_email_format_valid(self):
        """Test email format validation with valid emails"""
        valid_emails = [
            'test@example.com',
            'user.name@domain.co.uk',
            'user+tag@example.org'
        ]
        
        for email in valid_emails:
            with self.subTest(email=email):
                is_valid, error = validate_email_format(email)
                self.assertTrue(is_valid)
                self.assertEqual(error, "")
    
    def test_validate_email_format_invalid(self):
        """Test email format validation with invalid emails"""
        invalid_cases = [
            ('', 'Email is required'),
            ('   ', 'Email is required'),
            ('invalid_email', 'Please enter a valid email address'),
            ('@example.com', 'Please enter a valid email address'),
            ('user@', 'Please enter a valid email address')
        ]
        
        for email, expected_error in invalid_cases:
            with self.subTest(email=email):
                is_valid, error = validate_email_format(email)
                self.assertFalse(is_valid)
                self.assertIn(expected_error, error)
    
    def test_validate_password_format_valid(self):
        """Test password format validation with valid passwords"""
        valid_passwords = [
            'password123',
            'verylongpassword',
            'P@ssw0rd!',
            '12345678'  # Minimum 8 characters
        ]
        
        for password in valid_passwords:
            with self.subTest(password=password):
                is_valid, error = validate_password_format(password)
                self.assertTrue(is_valid)
                self.assertEqual(error, "")
    
    def test_validate_password_format_invalid(self):
        """Test password format validation with invalid passwords"""
        invalid_cases = [
            ('', 'Password is required'),
            ('short', 'Password must be at least 8 characters long'),
            ('1234567', 'Password must be at least 8 characters long')
        ]
        
        for password, expected_error in invalid_cases:
            with self.subTest(password=password):
                is_valid, error = validate_password_format(password)
                self.assertFalse(is_valid)
                self.assertEqual(error, expected_error)
    
    @patch('auth.User.find_by_email')
    def test_validate_registration_data_success(self, mock_find_by_email):
        """Test successful registration data validation"""
        mock_find_by_email.return_value = None  # Email doesn't exist
        
        is_valid, errors = validate_registration_data(
            'test@example.com',
            'password123',
            'password123'
        )
        
        self.assertTrue(is_valid)
        self.assertEqual(errors, [])
    
    @patch('auth.User.find_by_email')
    def test_validate_registration_data_email_exists(self, mock_find_by_email):
        """Test registration data validation when email exists"""
        mock_user = Mock()
        mock_find_by_email.return_value = mock_user  # Email exists
        
        is_valid, errors = validate_registration_data(
            'test@example.com',
            'password123',
            'password123'
        )
        
        self.assertFalse(is_valid)
        self.assertIn('Email already registered', errors)
    
    def test_validate_registration_data_password_mismatch(self):
        """Test registration data validation with password mismatch"""
        with patch('auth.User.find_by_email', return_value=None):
            is_valid, errors = validate_registration_data(
                'test@example.com',
                'password123',
                'different_password'
            )
        
        self.assertFalse(is_valid)
        self.assertIn('Passwords do not match', errors)
    
    def test_validate_login_data_success(self):
        """Test successful login data validation"""
        is_valid, errors = validate_login_data('test@example.com', 'password123')
        
        self.assertTrue(is_valid)
        self.assertEqual(errors, [])
    
    def test_validate_login_data_missing_fields(self):
        """Test login data validation with missing fields"""
        test_cases = [
            ('', 'password123', 'Email is required'),
            ('test@example.com', '', 'Password is required'),
            ('', '', 'Email is required')
        ]
        
        for email, password, expected_error in test_cases:
            with self.subTest(email=email, password=password):
                is_valid, errors = validate_login_data(email, password)
                self.assertFalse(is_valid)
                self.assertIn(expected_error, errors)
    
    @patch('auth.User.find_by_email')
    def test_authenticate_user_success(self, mock_find_by_email):
        """Test successful user authentication"""
        mock_user = Mock()
        mock_user.is_active = True
        mock_user.check_password.return_value = True
        mock_find_by_email.return_value = mock_user
        
        is_authenticated, user, error = authenticate_user('test@example.com', 'password123')
        
        self.assertTrue(is_authenticated)
        self.assertEqual(user, mock_user)
        self.assertEqual(error, "")
    
    @patch('auth.User.find_by_email')
    def test_authenticate_user_not_found(self, mock_find_by_email):
        """Test user authentication when user doesn't exist"""
        mock_find_by_email.return_value = None
        
        is_authenticated, user, error = authenticate_user('test@example.com', 'password123')
        
        self.assertFalse(is_authenticated)
        self.assertIsNone(user)
        self.assertEqual(error, "Invalid email or password")
    
    @patch('auth.User.find_by_email')
    def test_authenticate_user_inactive(self, mock_find_by_email):
        """Test user authentication when user is inactive"""
        mock_user = Mock()
        mock_user.is_active = False
        mock_find_by_email.return_value = mock_user
        
        is_authenticated, user, error = authenticate_user('test@example.com', 'password123')
        
        self.assertFalse(is_authenticated)
        self.assertIsNone(user)
        self.assertIn("disabled", error)
    
    @patch('auth.User.find_by_email')
    def test_authenticate_user_wrong_password(self, mock_find_by_email):
        """Test user authentication with wrong password"""
        mock_user = Mock()
        mock_user.is_active = True
        mock_user.check_password.return_value = False
        mock_find_by_email.return_value = mock_user
        
        is_authenticated, user, error = authenticate_user('test@example.com', 'wrongpassword')
        
        self.assertFalse(is_authenticated)
        self.assertIsNone(user)
        self.assertEqual(error, "Invalid email or password")
    
    @patch('auth.User.create_user')
    def test_create_user_account_success(self, mock_create_user):
        """Test successful user account creation"""
        mock_user = Mock()
        mock_create_user.return_value = mock_user
        
        with patch('auth.validate_registration_data', return_value=(True, [])):
            is_created, user, errors = create_user_account(
                'test@example.com',
                'password123',
                'password123'
            )
        
        self.assertTrue(is_created)
        self.assertEqual(user, mock_user)
        self.assertEqual(errors, [])
    
    def test_create_user_account_validation_error(self):
        """Test user account creation with validation errors"""
        validation_errors = ['Email already registered', 'Passwords do not match']
        
        with patch('auth.validate_registration_data', return_value=(False, validation_errors)):
            is_created, user, errors = create_user_account(
                'test@example.com',
                'password123',
                'different_password'
            )
        
        self.assertFalse(is_created)
        self.assertIsNone(user)
        self.assertEqual(errors, validation_errors)
    
    def test_is_safe_url_safe(self):
        """Test safe URL validation with safe URLs"""
        app = Flask(__name__)
        
        with app.test_request_context('http://localhost:5000/'):
            safe_urls = [
                '/dashboard',
                '/chat',
                'http://localhost:5000/profile',
                'https://localhost:5000/settings'
            ]
            
            for url in safe_urls:
                with self.subTest(url=url):
                    self.assertTrue(is_safe_url(url))
    
    def test_is_safe_url_unsafe(self):
        """Test safe URL validation with unsafe URLs"""
        app = Flask(__name__)
        
        with app.test_request_context('http://localhost:5000/'):
            unsafe_urls = [
                'http://evil.com/malicious',
                'https://phishing.site/steal',
                'ftp://localhost:5000/file',
                'javascript:alert("xss")'
            ]
            
            for url in unsafe_urls:
                with self.subTest(url=url):
                    self.assertFalse(is_safe_url(url))


class TestLoginRequiredDecorator(unittest.TestCase):
    """Test cases for login_required decorator"""
    
    def setUp(self):
        """Set up Flask app for testing"""
        self.app = Flask(__name__)
        self.app.config['SECRET_KEY'] = 'test-secret-key'
        self.app.config['WTF_CSRF_ENABLED'] = False
        
        # Initialize Flask-Login
        from auth import init_login_manager
        init_login_manager(self.app)
        
        @self.app.route('/protected')
        @login_required
        def protected_route():
            return 'Protected content'
        
        @self.app.route('/login')
        def login():
            return 'Login page'
        
        self.client = self.app.test_client()
    
    def test_login_required_unauthenticated_redirect(self):
        """Test login_required decorator redirects unauthenticated users"""
        with self.app.test_request_context():
            response = self.client.get('/protected')
            
            self.assertEqual(response.status_code, 302)
            self.assertIn('/login', response.location)
    
    def test_login_required_ajax_unauthenticated(self):
        """Test login_required decorator handles AJAX requests"""
        with self.app.test_request_context():
            response = self.client.get(
                '/protected',
                headers={'Content-Type': 'application/json'}
            )
            
            self.assertEqual(response.status_code, 401)
            data = response.get_json()
            self.assertFalse(data['success'])
            self.assertIn('Authentication required', data['error'])


if __name__ == '__main__':
    # Run the tests
    unittest.main(verbosity=2)