317 lines
10 KiB
Python
317 lines
10 KiB
Python
"""Unit tests for Activities class with 100% coverage."""
|
|
|
|
import asyncio
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_logger():
|
|
"""Create a mock logger."""
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_notification_handler():
|
|
"""Create a mock notification handler."""
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def postgres_config():
|
|
"""Create a valid PostgreSQL configuration."""
|
|
return {
|
|
'host': 'localhost',
|
|
'port': 5432,
|
|
'user': 'testuser',
|
|
'password': 'testpass',
|
|
'dbname': 'testdb',
|
|
'min_connections': 1,
|
|
'max_connections': 10,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def mlflow_config():
|
|
"""Create a valid MLFlow configuration."""
|
|
return {
|
|
'url': 'http://mlflow:5080',
|
|
'username': 'aignosi',
|
|
'password': 'aignosi',
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def minio_config():
|
|
"""Create a valid MinIO configuration."""
|
|
return {
|
|
'endpoint_url': 'http://minio:9000',
|
|
'access_key': 'minioadmin',
|
|
'secret_key': 'minioadmin',
|
|
'region': 'us-east-1',
|
|
'use_ssl': False,
|
|
'max_retry_attempts': 3,
|
|
'retry_mode': 'standard',
|
|
'connect_timeout': 5,
|
|
'read_timeout': 5,
|
|
}
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_init_success(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test successful initialization of Activities."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
mock_et_init.assert_called_once()
|
|
assert mock_et_init.call_args[1]['host'] == postgres_config['host']
|
|
assert mock_et_init.call_args[1]['port'] == postgres_config['port']
|
|
assert mock_et_init.call_args[1]['user'] == postgres_config['user']
|
|
assert mock_et_init.call_args[1]['password'] == postgres_config['password']
|
|
assert mock_et_init.call_args[1]['dbname'] == postgres_config['dbname']
|
|
assert mock_et_init.call_args[1]['min_connections'] == postgres_config['min_connections']
|
|
assert mock_et_init.call_args[1]['max_connections'] == postgres_config['max_connections']
|
|
assert mock_et_init.call_args[1]['logger'] is mock_logger
|
|
assert mock_et_init.call_args[1]['notification_handler'] is mock_notification_handler
|
|
|
|
mock_model_repo.assert_called_once_with(
|
|
url=mlflow_config['url'],
|
|
username=mlflow_config['username'],
|
|
password=mlflow_config['password'],
|
|
logger=mock_logger,
|
|
)
|
|
|
|
mock_storage_repo.assert_called_once_with(
|
|
endpoint_url=minio_config['endpoint_url'],
|
|
access_key=minio_config['access_key'],
|
|
secret_key=minio_config['secret_key'],
|
|
region=minio_config['region'],
|
|
use_ssl=minio_config['use_ssl'],
|
|
max_retry_attempts=minio_config['max_retry_attempts'],
|
|
retry_mode=minio_config['retry_mode'],
|
|
connect_timeout=minio_config['connect_timeout'],
|
|
read_timeout=minio_config['read_timeout'],
|
|
logger=mock_logger,
|
|
)
|
|
|
|
mock_training_init.assert_called_once()
|
|
assert mock_training_init.call_args[1]['model_repository'] is mock_model_repo.return_value
|
|
assert mock_training_init.call_args[1]['storage_repository'] is mock_storage_repo.return_value
|
|
assert mock_training_init.call_args[1]['logger'] is mock_logger
|
|
assert mock_training_init.call_args[1]['notification_handler'] is mock_notification_handler
|
|
|
|
assert hasattr(activities, 'model_repository')
|
|
assert hasattr(activities, 'storage_repository')
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ExperimentTracking.close')
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_shutdown(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_close,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test Activities.shutdown() calls ExperimentTracking.close()."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
asyncio.run(activities.shutdown())
|
|
|
|
mock_et_close.assert_called_once_with(activities)
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_del_without_engine(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test __del__ when engine attribute does not exist."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
if hasattr(activities, 'engine'):
|
|
delattr(activities, 'engine')
|
|
|
|
activities.__del__()
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_del_with_engine_no_super_del(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test __del__ when engine exists but super has no __del__."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
activities.engine = MagicMock()
|
|
|
|
with patch('builtins.super') as mock_super:
|
|
mock_super_instance = MagicMock()
|
|
del mock_super_instance.__del__
|
|
mock_super.return_value = mock_super_instance
|
|
|
|
activities.__del__()
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_del_with_engine_and_super_del(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test __del__ when engine exists and super has __del__."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
activities.engine = MagicMock()
|
|
|
|
mock_super_del = MagicMock()
|
|
|
|
class MockSuper:
|
|
def __del__(self):
|
|
mock_super_del()
|
|
|
|
with patch('builtins.super', return_value=MockSuper()):
|
|
activities.__del__()
|
|
|
|
mock_super_del.assert_called_once()
|
|
|
|
|
|
@patch('model_manager.activities.activities.ExperimentTracking.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.Training.__init__', return_value=None)
|
|
@patch('model_manager.activities.activities.ModelRepository')
|
|
@patch('model_manager.activities.activities.StorageRepository')
|
|
def test_activities_del_with_engine_exception_caught(
|
|
mock_storage_repo,
|
|
mock_model_repo,
|
|
mock_training_init,
|
|
mock_et_init,
|
|
postgres_config,
|
|
mlflow_config,
|
|
minio_config,
|
|
mock_logger,
|
|
mock_notification_handler,
|
|
):
|
|
"""Test __del__ catches exceptions when super().__del__() raises."""
|
|
from model_manager.activities.activities import Activities
|
|
|
|
activities = Activities(
|
|
postgres_config=postgres_config,
|
|
mlflow_config=mlflow_config,
|
|
minio_config=minio_config,
|
|
logger=mock_logger,
|
|
notification_handler=mock_notification_handler,
|
|
)
|
|
|
|
activities.engine = MagicMock()
|
|
|
|
class MockSuperWithError:
|
|
def __del__(self):
|
|
# Only raise error if not being cleaned up by garbage collector
|
|
# This prevents the PytestUnraisableExceptionWarning
|
|
if hasattr(self, '_should_raise') and self._should_raise:
|
|
raise RuntimeError('Test error')
|
|
|
|
# Suppress the PytestUnraisableExceptionWarning for this specific test
|
|
import warnings
|
|
|
|
warnings.filterwarnings('ignore', category=pytest.PytestUnraisableExceptionWarning)
|
|
|
|
mock_super = MockSuperWithError()
|
|
mock_super._should_raise = True
|
|
try:
|
|
with patch('builtins.super', return_value=mock_super):
|
|
activities.__del__()
|
|
finally:
|
|
# Prevent the exception from being raised during garbage collection
|
|
mock_super._should_raise = False
|