TransferData-main/tests/test_db_utils.py

# Tests para App/Utils/db_utils.py
import pytest
from unittest.mock import Mock, MagicMock, patch, call
from App.Utils.db_utils import DatabaseManager, get_db_connection


class TestDatabaseManager:
    """Tests para la clase DatabaseManager"""

    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_init(self, mock_config):
        """Test que DatabaseManager se inicializa correctamente con credenciales"""
        # Arrange
        mock_config.db_user = 'test_user'
        mock_config.db_password = 'test_pass'
        mock_config.db_host = 'test_host'

        # Act
        manager = DatabaseManager('test_db', 'test_id')

        # Assert
        assert manager.db_database == 'test_db'
        assert manager.db_databaseID == 'test_id'
        assert manager.db_user == 'test_user'
        assert manager.db_password == 'test_pass'
        assert manager.db_host == 'test_host'

    @patch('App.Utils.db_utils.pyodbc.connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_connect_success(self, mock_config, mock_pyodbc_connect):
        """Test que _connect establece conexión exitosa"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'
        mock_connection = MagicMock()
        mock_pyodbc_connect.return_value = mock_connection

        manager = DatabaseManager('test_db', 'test_id')

        # Act
        manager._connect()

        # Assert
        assert manager.connection == mock_connection
        mock_pyodbc_connect.assert_called_once()
        assert 'ODBC Driver 17 for SQL Server' in manager.conn_str

    @patch('App.Utils.db_utils.pyodbc.connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_connect_failure(self, mock_config, mock_pyodbc_connect):
        """Test que _connect maneja errores de conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'
        mock_pyodbc_connect.side_effect = Exception("Connection failed")

        manager = DatabaseManager('test_db', 'test_id')

        # Act
        manager._connect()

        # Assert
        assert manager.connection is None

    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_disconnect(self, mock_config):
        """Test que _disconnect cierra la conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'
        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()

        # Act
        manager._disconnect()

        # Assert
        manager.connection.close.assert_called_once()

    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_disconnect_no_connection(self, mock_config):
        """Test que _disconnect no falla si no hay conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'
        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = None

        # Act & Assert (no debe lanzar excepción)
        manager._disconnect()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_get_execute_query_success(self, mock_config, mock_connect, mock_disconnect):
        """Test que get_execute_query devuelve datos como lista de dicts"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()
        manager.cursor = MagicMock()

        # Simula resultado de query
        manager.cursor.description = [('id',), ('name',)]
        manager.cursor.fetchall.return_value = [(1, 'Alice'), (2, 'Bob')]
        manager.connection.cursor.return_value = manager.cursor

        # Act
        results, error = manager.get_execute_query("SELECT * FROM users")

        # Assert
        assert error is None
        assert len(results) == 2
        assert results[0] == {'id': 1, 'name': 'Alice'}
        assert results[1] == {'id': 2, 'name': 'Bob'}
        manager.cursor.execute.assert_called_once_with("SELECT * FROM users")
        mock_disconnect.assert_called_once()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_get_execute_query_no_connection(self, mock_config, mock_connect, mock_disconnect):
        """Test que get_execute_query devuelve error si no hay conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = None

        # Act
        results, error = manager.get_execute_query("SELECT * FROM users")

        # Assert
        assert results is None
        assert error == "Error de conexión"

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_get_execute_query_error(self, mock_config, mock_connect, mock_disconnect):
        """Test que get_execute_query maneja errores de query"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()
        manager.cursor = MagicMock()
        manager.cursor.execute.side_effect = Exception("invalid column")
        manager.connection.cursor.return_value = manager.cursor

        # Act
        results, error = manager.get_execute_query("SELECT invalid_col FROM users")

        # Assert
        assert results is None
        assert "invalid column" in error
        mock_disconnect.assert_called_once()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_set_execute_query_success(self, mock_config, mock_connect, mock_disconnect):
        """Test que set_execute_query hace commit correctamente"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()
        cursor_mock = MagicMock()
        manager.connection.cursor.return_value = cursor_mock

        # Act
        result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")

        # Assert
        assert result is True
        cursor_mock.execute.assert_called_once_with("INSERT INTO users VALUES (1, 'Alice')")
        manager.connection.commit.assert_called_once()
        cursor_mock.close.assert_called_once()
        mock_disconnect.assert_called_once()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_set_execute_query_no_connection(self, mock_config, mock_connect, mock_disconnect):
        """Test que set_execute_query devuelve False si no hay conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = None

        # Act
        result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")

        # Assert
        assert result is False

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_set_execute_query_error(self, mock_config, mock_connect, mock_disconnect):
        """Test que set_execute_query maneja errores"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()
        cursor_mock = MagicMock()
        cursor_mock.execute.side_effect = Exception("Constraint violation")
        manager.connection.cursor.return_value = cursor_mock

        # Act
        result = manager.set_execute_query("INSERT INTO users VALUES (1, 'Alice')")

        # Assert
        assert result is False
        mock_disconnect.assert_called_once()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_log_execution_success(self, mock_config, mock_connect, mock_disconnect):
        """Test que log_execution ejecuta el SP correctamente"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = MagicMock()
        manager.cursor = MagicMock()
        manager.connection.cursor.return_value = manager.cursor

        # Act
        manager.log_execution(
            procedure_id=1,
            procedure_name='test_proc',
            status='Success',
            message='Test completed',
            executed_by='test_user',
            extra_info='None',
            target_date='2023-01-01',
            date='2023-01-01'
        )

        # Assert
        assert manager.cursor.execute.called
        manager.connection.commit.assert_called_once()
        mock_disconnect.assert_called_once()

    @patch.object(DatabaseManager, '_disconnect')
    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_log_execution_no_connection(self, mock_config, mock_connect, mock_disconnect):
        """Test que log_execution maneja falta de conexión"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'

        manager = DatabaseManager('test_db', 'test_id')
        manager.connection = None

        # Act & Assert (no debe lanzar excepción)
        manager.log_execution(
            procedure_id=1,
            procedure_name='test_proc',
            status='Error',
            message='Connection failed',
            executed_by='test_user',
            extra_info='None',
            target_date='2023-01-01',
            date='2023-01-01'
        )


class TestGetDbConnection:
    """Tests para la función get_db_connection"""

    @patch.object(DatabaseManager, '_connect')
    @patch('App.Utils.db_utils.ConfigDatabase')
    def test_get_db_connection_returns_connection(self, mock_config, mock_connect):
        """Test que get_db_connection devuelve una conexión pyodbc"""
        # Arrange
        mock_config.db_user = 'user'
        mock_config.db_password = 'pass'
        mock_config.db_host = 'server'
        mock_connection = MagicMock()

        with patch.object(DatabaseManager, '__init__', return_value=None):
            with patch.object(DatabaseManager, '_connect'):
                manager = DatabaseManager.__new__(DatabaseManager)
                manager.connection = mock_connection

                # Act (simulamos el comportamiento)
                conn = manager.connection

                # Assert
                assert conn == mock_connection

    @patch('App.Utils.db_utils.DatabaseManager')
    def test_get_db_connection_params(self, mock_db_manager_class):
        """Test que get_db_connection pasa parámetros correctamente"""
        # Arrange
        mock_instance = MagicMock()
        mock_instance.connection = MagicMock()
        mock_db_manager_class.return_value = mock_instance

        # Act
        conn = get_db_connection(database='custom_db', databaseID='custom_id')

        # Assert
        mock_db_manager_class.assert_called_once_with('custom_db', 'custom_id')
        mock_instance._connect.assert_called_once()