diff --git a/.env.example b/.env.example index 652ce20..3b7e542 100644 --- a/.env.example +++ b/.env.example @@ -23,4 +23,14 @@ MONGODB_USERNAME="mongo_user" MONGODB_PASSWORD="mongo_db_password" MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017" MONGODB_DATABASE="sientia" -MONGODB_TTL_INDEX_HOURS="1" \ No newline at end of file +MONGODB_TTL_INDEX_HOURS="1" + +MINIO_ENDPOINT_URL="http://minio.minio.svc.cluster.local:9000" +MINIO_ACCESS_KEY="minioadmin" +MINIO_SECRET_KEY="minioadmin" +MINIO_REGION="us-east-1" +MINIO_USE_SSL="false" +MINIO_MAX_RETRY_ATTEMPTS="3" +MINIO_RETRY_MODE="adaptive" +MINIO_CONNECT_TIMEOUT="10" +MINIO_READ_TIMEOUT="60" \ No newline at end of file diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 9fedd57..79daef4 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -8,25 +8,28 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.postgres import Postgres from model_manager.activities.gates import Gates + from model_manager.activities.minio import MinIO from model_manager.activities.mlflow import MLFlow -class Activities(Postgres, MLFlow, Gates): +class Activities(Postgres, MLFlow, MinIO, Gates): """ Main activities orchestrator for the Model Manager system. This class combines functionality from multiple activity classes to provide a unified interface for all workflow operations. It manages database connections, - MLFlow model interactions, and data quality validation. + MLFlow model interactions, MinIO storage operations, and data quality validation. The class implements multiple inheritance to combine specialized functionality: - Postgres: Database operations and data persistence - MLFlow: Model inference and transformation operations + - MinIO: Object storage operations (file upload/download/delete) - Gates: Data quality validation and filtering mechanisms Attributes: postgres_config (dict): PostgreSQL connection configuration mlflow_config (dict): MLFlow server configuration + minio_config (dict): MinIO storage configuration logger (Logger): Logging and observability instance notification_handler (NotificationHandler): Notification management instance """ @@ -35,6 +38,7 @@ class Activities(Postgres, MLFlow, Gates): self, postgres_config: dict[str, Any], mlflow_config: dict[str, Any], + minio_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler, ): @@ -49,6 +53,8 @@ class Activities(Postgres, MLFlow, Gates): Required keys: host, port, user, password, dbname, min_connections, max_connections mlflow_config: MLFlow server configuration dictionary Required keys: host, port, username, password + minio_config: MinIO storage configuration dictionary + Required keys: endpoint_url, access_key, secret_key, region, use_ssl logger: Logger instance for observability and debugging notification_handler: Notification handler for alerts and monitoring @@ -79,6 +85,21 @@ class Activities(Postgres, MLFlow, Gates): notification_handler=notification_handler, ) + MinIO.__init__( + self, + 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=logger, + notification_handler=notification_handler, + ) + Gates.__init__(self, logger=logger, notification_handler=notification_handler) async def shutdown(self): diff --git a/model_manager/activities/minio.py b/model_manager/activities/minio.py new file mode 100644 index 0000000..0637e21 --- /dev/null +++ b/model_manager/activities/minio.py @@ -0,0 +1,229 @@ +from temporalio import activity, workflow + +with workflow.unsafe.imports_passed_through(): + import traceback + from io import BytesIO + from typing import Any + + import boto3 # type: ignore[import-untyped] + from botocore.config import Config # type: ignore[import-untyped] + from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler + from sientia_do.notifications.models import NotificationLevel + from sientia_do.observability.logger import Logger + from sientia_do.temporal.activities.base import BaseActivity + + +class MinIO(BaseActivity): + """ + MinIO (S3-compatible) storage activities for file operations. + + This class provides activities for interacting with MinIO object storage, + including file download and deletion operations. It handles authentication, + connection management, and comprehensive error handling. + + The class implements best practices for S3/MinIO operations: + - Connection reuse (boto3 client is thread-safe) + - Automatic retry with exponential backoff + - Comprehensive error handling and logging + - Notification integration for critical errors + + Attributes: + endpoint_url (str): MinIO server endpoint URL + access_key (str): MinIO access key ID + secret_key (str): MinIO secret access key + region (str): MinIO region name + use_ssl (bool): Whether to use SSL/TLS for connections + minio_client: Boto3 S3 client configured for MinIO + """ + + def __init__( + self, + endpoint_url: str, + access_key: str, + secret_key: str, + region: str, + use_ssl: bool, + max_retry_attempts: int, + retry_mode: str, + connect_timeout: int, + read_timeout: int, + logger: Logger, + notification_handler: NotificationHandler, + ): + """ + Initialize MinIO activities with server configuration. + + This constructor creates a persistent boto3 S3 client that will be + reused across all activity calls. The client is thread-safe and + includes automatic retry configuration. + + Args: + endpoint_url: MinIO server endpoint URL (e.g., http://localhost:9000) + access_key: MinIO access key ID for authentication + secret_key: MinIO secret access key for authentication + region: MinIO region name (e.g., us-east-1) + use_ssl: Whether to use SSL/TLS for connections + max_retry_attempts: Maximum number of retry attempts (e.g., 3) + retry_mode: Retry mode - standard, legacy, or adaptive (e.g., adaptive) + connect_timeout: Connection timeout in seconds (e.g., 10) + read_timeout: Read timeout in seconds (e.g., 60) + logger: Logger instance for observability and debugging + notification_handler: Notification handler for alerts and monitoring + + Raises: + Exception: If boto3 client initialization fails + """ + BaseActivity.__init__(self, logger, notification_handler, set_error_counter=True) + self.endpoint_url = endpoint_url + self.access_key = access_key + self.secret_key = secret_key + self.region = region + self.use_ssl = use_ssl + self.max_retry_attempts = max_retry_attempts + self.retry_mode = retry_mode + self.connect_timeout = connect_timeout + self.read_timeout = read_timeout + + # Configure boto3 with retry strategy + # This handles transient network errors and connection issues automatically + boto_config = Config( + region_name=region, + retries={ + 'max_attempts': max_retry_attempts, + 'mode': retry_mode, + }, + connect_timeout=connect_timeout, + read_timeout=read_timeout, + ) + + try: + self.minio_client = boto3.client( + 's3', + endpoint_url=endpoint_url, + aws_access_key_id=access_key, + aws_secret_access_key=secret_key, + config=boto_config, + use_ssl=use_ssl, + ) + self.info(f'MinIO client initialized successfully: {endpoint_url}') + except Exception as e: + error_msg = f'Failed to initialize MinIO client: {str(e)}' + self.error(error_msg) + raise Exception(error_msg) from e + + @activity.defn(name='fetch_file_from_minio') + async def fetch_file_from_minio(self, input_data: dict[str, Any]) -> BytesIO: + """ + Fetch a file from MinIO and return its content as a BytesIO object. + + This activity downloads a file from a MinIO bucket and returns the + content as a BytesIO object, which is a file-like object that can be + used directly with many Python libraries (pandas, PIL, etc.). + + The operation includes: + 1. Input validation + 2. File download from MinIO + 3. Content reading and wrapping in BytesIO + 4. Comprehensive error handling and logging + + Args: + input_data: Configuration for file fetch operation + Required keys: + - metadata (dict): Workflow execution metadata + - bucket_name (str): MinIO bucket name + - file_name (str): File path/key in the bucket + + Returns: + BytesIO: File content as a file-like object + + Raises: + Exception: If file fetch fails due to network, permission, or other errors + """ + metadata = input_data.get('metadata', {}) + bucket_name = input_data['bucket_name'] + file_name = input_data['file_name'] + + self.info(f'Fetching file from MinIO: {bucket_name}/{file_name}', metadata) + + try: + # Download file from MinIO + response = self.minio_client.get_object(Bucket=bucket_name, Key=file_name) + + # Read file content + with response['Body'] as body: + file_content = body.read() + + file_size = len(file_content) + self.info( + f'File fetched successfully: {bucket_name}/{file_name} ({file_size} bytes)', + metadata, + ) + + return BytesIO(file_content) + + except Exception as e: # noqa: BLE001 + error_msg = f'Error fetching file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}' + trace = traceback.format_exc() + self.send_notification( + metadata=metadata, + notification_id='FETCH_FILE_FROM_MINIO_ERROR', + message=error_msg, + block='fetch_file_from_minio', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + self.error(trace, metadata=metadata) + raise Exception(error_msg) from e + + @activity.defn(name='delete_file_from_minio') + async def delete_file_from_minio(self, input_data: dict[str, Any]) -> None: + """ + Delete a file from MinIO storage. + + This activity removes a file from a MinIO bucket. The operation is + idempotent - deleting a non-existent file is considered successful. + + The operation includes: + 1. Input validation + 2. File deletion from MinIO + 3. Comprehensive error handling and logging + + Args: + input_data: Configuration for file deletion operation + Required keys: + - metadata (dict): Workflow execution metadata + - bucket_name (str): MinIO bucket name + - file_name (str): File path/key to delete + + Returns: + None + + Raises: + Exception: If file deletion fails due to permission or other errors + """ + metadata = input_data.get('metadata', {}) + bucket_name = input_data['bucket_name'] + file_name = input_data['file_name'] + + self.info(f'Deleting file from MinIO: {bucket_name}/{file_name}', metadata) + + try: + # Delete file from MinIO + # Note: delete_object is idempotent - no error if file doesn't exist + self.minio_client.delete_object(Bucket=bucket_name, Key=file_name) + + self.info(f'File deleted successfully: {bucket_name}/{file_name}', metadata) + + except Exception as e: # noqa: BLE001 + error_msg = f'Error deleting file from MinIO - Bucket: {bucket_name}, File: {file_name}, Error: {str(e)}' + trace = traceback.format_exc() + self.send_notification( + metadata=metadata, + notification_id='DELETE_FILE_FROM_MINIO_ERROR', + message=error_msg, + block='delete_file_from_minio', + level=NotificationLevel.ERROR, + attachment_content=trace, + ) + self.error(trace, metadata=metadata) + raise Exception(error_msg) from e diff --git a/model_manager/utils/connectors_config.py b/model_manager/utils/connectors_config.py index aec7625..6f338a4 100644 --- a/model_manager/utils/connectors_config.py +++ b/model_manager/utils/connectors_config.py @@ -87,3 +87,38 @@ def build_mongodb_config() -> dict[str, Any]: 'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'), 'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600, } + + +def build_minio_config() -> dict[str, Any]: + """ + Build MinIO (S3-compatible) configuration from environment variables. + + This function constructs a MinIO configuration dictionary from + environment variables with sensible defaults for local development. + It handles endpoint URL, authentication, connection parameters, and retry policies. + + Environment Variables: + MINIO_ENDPOINT_URL: MinIO server endpoint URL (default: http://localhost:9000) + MINIO_ACCESS_KEY: MinIO access key ID (default: minioadmin) + MINIO_SECRET_KEY: MinIO secret access key (default: minioadmin) + MINIO_REGION: MinIO region name (default: us-east-1) + MINIO_USE_SSL: Whether to use SSL/TLS (default: false) + MINIO_MAX_RETRY_ATTEMPTS: Maximum number of retry attempts (default: 3) + MINIO_RETRY_MODE: Retry mode - standard, legacy, or adaptive (default: adaptive) + MINIO_CONNECT_TIMEOUT: Connection timeout in seconds (default: 10) + MINIO_READ_TIMEOUT: Read timeout in seconds (default: 60) + + Returns: + dict: MinIO configuration dictionary with all required parameters + """ + return { + 'endpoint_url': getenv('MINIO_ENDPOINT_URL', 'http://localhost:9000'), + 'access_key': getenv('MINIO_ACCESS_KEY', 'minioadmin'), + 'secret_key': getenv('MINIO_SECRET_KEY', 'minioadmin'), + 'region': getenv('MINIO_REGION', 'us-east-1'), + 'use_ssl': getenv('MINIO_USE_SSL', 'false').lower() == 'true', + 'max_retry_attempts': int(getenv('MINIO_MAX_RETRY_ATTEMPTS', '3')), + 'retry_mode': getenv('MINIO_RETRY_MODE', 'adaptive'), + 'connect_timeout': int(getenv('MINIO_CONNECT_TIMEOUT', '10')), + 'read_timeout': int(getenv('MINIO_READ_TIMEOUT', '60')), + } diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 76fc37e..7a8b771 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -41,6 +41,7 @@ with workflow.unsafe.imports_passed_through(): from model_manager import metrics from model_manager.activities.activities import Activities from model_manager.utils.connectors_config import ( + build_minio_config, build_mlflow_config, build_mongodb_config, build_postgres_config, @@ -106,6 +107,7 @@ async def main(): activities = Activities( postgres_config=build_postgres_config(), mlflow_config=build_mlflow_config(), + minio_config=build_minio_config(), logger=logger, notification_handler=notification_handler, ) diff --git a/requirements.txt b/requirements.txt index 685fa78..57c0af1 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,8 @@ temporalio psycopg2-binary sqlalchemy +boto3 +botocore git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.6 git+ssh://git@github.com/Aignosi/sientia-mlops-library.git@0.39.0 prometheus-client diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index b0cd4e6..99f6083 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -10,8 +10,9 @@ from model_manager.activities.mlflow import MLFlow @patch('model_manager.activities.activities.Postgres.__init__') @patch('model_manager.activities.activities.MLFlow.__init__') +@patch('model_manager.activities.activities.MinIO.__init__') @patch('model_manager.activities.activities.Gates.__init__') -def test___init__(mock_gates_init, mock_mlflow_init, mock_postgres_init): +def test___init__(mock_gates_init, mock_minio_init, mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', 'port': 5432, @@ -24,12 +25,25 @@ def test___init__(mock_gates_init, mock_mlflow_init, mock_postgres_init): mlflow_config = {'host': 'localhost', 'port': 5000, 'username': 'mlflow', 'password': 'mlflow'} + minio_config = { + 'endpoint_url': 'http://localhost:9000', + 'access_key': 'minioadmin', + 'secret_key': 'minioadmin', + 'region': 'us-east-1', + 'use_ssl': False, + 'max_retry_attempts': 3, + 'retry_mode': 'adaptive', + 'connect_timeout': 10, + 'read_timeout': 60, + } + logger = MagicMock() notification_handler = MagicMock() activities = Activities( postgres_config=postgres_config, mlflow_config=mlflow_config, + minio_config=minio_config, logger=logger, notification_handler=notification_handler, ) @@ -62,6 +76,21 @@ def test___init__(mock_gates_init, mock_mlflow_init, mock_postgres_init): notification_handler=notification_handler, ) + mock_minio_init.assert_called_once_with( + ANY, + 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=logger, + notification_handler=notification_handler, + ) + mock_gates_init.assert_called_once_with( ANY, logger=logger, notification_handler=notification_handler ) @@ -83,12 +112,25 @@ async def test_shutdown(_mock_mlflow_init, mock_postgres_init): mlflow_config = {'host': 'localhost', 'port': 5000, 'username': 'mlflow', 'password': 'mlflow'} + minio_config = { + 'endpoint_url': 'http://localhost:9000', + 'access_key': 'minioadmin', + 'secret_key': 'minioadmin', + 'region': 'us-east-1', + 'use_ssl': False, + 'max_retry_attempts': 3, + 'retry_mode': 'adaptive', + 'connect_timeout': 10, + 'read_timeout': 60, + } + logger = MagicMock() notification_handler = MagicMock() activities = Activities( postgres_config=postgres_config, mlflow_config=mlflow_config, + minio_config=minio_config, logger=logger, notification_handler=notification_handler, ) diff --git a/tests/laborious/activities/test_gates.py b/tests/laborious/activities/test_gates.py index ee68007..2b3707b 100644 --- a/tests/laborious/activities/test_gates.py +++ b/tests/laborious/activities/test_gates.py @@ -117,6 +117,25 @@ async def test_input_gate_with_filter(gates_activity): gates_activity.debug.assert_called() +@mark.asyncio +async def test_input_gate_filter_returns_false(gates_activity): + """Test to cover line 129 branch when filter returns False (filter passes).""" + # Arrange - Use data that will NOT trigger EMPTY_DATA filter (has data) + input_data = { + **metadata, + 'filters': {'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, + 'data': {'value': [1, 2, 3, 4, 5]}, # Has data, filter returns False + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + result = await gates_activity.input_gate(input_data) + + # Assert - Filter returns False, so no policy is added to filter_output + assert result == (None, 0, '') # No filter triggered + gates_activity.debug.assert_called() + + @mark.asyncio async def test_mlflow_response_gate_invalid_filter(gates_activity): # Arrange @@ -210,6 +229,29 @@ async def test_mlflow_response_gate_with_filter(gates_activity): gates_activity.send_notification.assert_called() +@mark.asyncio +async def test_mlflow_response_gate_filter_returns_false(gates_activity): + """Test to cover line 208 branch when filter returns False (no API error).""" + # Arrange - Use data that will NOT trigger API_ERROR filter (success=True) + input_data = { + **metadata, + 'filters': {'API_ERROR': {'policy': 'STOP'}}, + 'data': { + 'success': True, # Success=True, filter returns False + 'content': {'message': 'Operation successful', 'result': 'data'}, + }, + 'type': 'transform', + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + result = await gates_activity.mlflow_response_gate(input_data) + + # Assert - Filter returns False, so no policy is added to filter_output + assert result == (None, 0, '') # No filter triggered + gates_activity.debug.assert_called() + + @mark.asyncio async def test_mlflow_content_gate_invalid_filter(gates_activity): # Arrange @@ -304,6 +346,26 @@ async def test_mlflow_content_gate_with_filter(gates_activity): gates_activity.send_notification.assert_called() +@mark.asyncio +async def test_mlflow_content_gate_filter_returns_false(gates_activity): + """Test to cover line 293 branch when filter returns False (no NaN values).""" + # Arrange - Use data that will NOT trigger NAN_VALUES filter (no NaN) + input_data = { + **metadata, + 'filters': {'NAN_VALUES': {'policy': 'STOP', 'config': {}}}, + 'data': {'value': [1.0, 2.0, 3.0, 4.0, 5.0]}, # All valid numbers, no NaN + 'type': 'predict', + 'path_priority': ['STOP', 'CONTINUE', 'REPEAT'], + } + + # Act + result = await gates_activity.mlflow_content_gate(input_data) + + # Assert - Filter returns False, so no policy is added to filter_output + assert result == (None, 0, '') # No filter triggered + gates_activity.debug.assert_called() + + def test_get_prediction_store_policy_invalid_policy(gates_activity): # Arrange prediction_store_policy = 'INVALID_POLICY' diff --git a/tests/laborious/activities/test_minio.py b/tests/laborious/activities/test_minio.py new file mode 100644 index 0000000..89888bd --- /dev/null +++ b/tests/laborious/activities/test_minio.py @@ -0,0 +1,328 @@ +from io import BytesIO +from unittest.mock import MagicMock, patch + +from pytest import fixture, mark, raises + +from model_manager.activities.minio import MinIO + + +@patch('model_manager.activities.minio.boto3.client') +def test___init__(mock_boto3_client): + """Test MinIO initialization with correct configuration.""" + mock_client = MagicMock() + mock_boto3_client.return_value = mock_client + + logger = MagicMock() + notification_handler = MagicMock() + + minio = MinIO( + endpoint_url='http://localhost:9000', + access_key='minioadmin', + secret_key='minioadmin', + region='us-east-1', + use_ssl=False, + max_retry_attempts=3, + retry_mode='adaptive', + connect_timeout=10, + read_timeout=60, + logger=logger, + notification_handler=notification_handler, + ) + + assert minio.endpoint_url == 'http://localhost:9000' + assert minio.access_key == 'minioadmin' + assert minio.secret_key == 'minioadmin' + assert minio.region == 'us-east-1' + assert minio.use_ssl is False + assert minio.max_retry_attempts == 3 + assert minio.retry_mode == 'adaptive' + assert minio.connect_timeout == 10 + assert minio.read_timeout == 60 + + # Verify boto3 client was created with correct parameters + mock_boto3_client.assert_called_once() + call_kwargs = mock_boto3_client.call_args[1] + assert call_kwargs['endpoint_url'] == 'http://localhost:9000' + assert call_kwargs['aws_access_key_id'] == 'minioadmin' + assert call_kwargs['aws_secret_access_key'] == 'minioadmin' + assert call_kwargs['use_ssl'] is False + + +@patch('model_manager.activities.minio.boto3.client') +def test___init___failure(mock_boto3_client): + """Test MinIO initialization failure handling.""" + mock_boto3_client.side_effect = Exception('Connection failed') + + logger = MagicMock() + notification_handler = MagicMock() + + with raises(Exception, match='Failed to initialize MinIO client'): + MinIO( + endpoint_url='http://localhost:9000', + access_key='minioadmin', + secret_key='minioadmin', + region='us-east-1', + use_ssl=False, + max_retry_attempts=3, + retry_mode='adaptive', + connect_timeout=10, + read_timeout=60, + logger=logger, + notification_handler=notification_handler, + ) + + +@fixture +@patch('model_manager.activities.minio.boto3.client') +def minio(mock_boto3_client): + """Fixture to create a MinIO instance for testing.""" + mock_client = MagicMock() + mock_boto3_client.return_value = mock_client + + logger = MagicMock() + notification_handler = MagicMock() + + minio_instance = MinIO( + endpoint_url='http://localhost:9000', + access_key='minioadmin', + secret_key='minioadmin', + region='us-east-1', + use_ssl=False, + max_retry_attempts=3, + retry_mode='adaptive', + connect_timeout=10, + read_timeout=60, + logger=logger, + notification_handler=notification_handler, + ) + + minio_instance.send_notification = MagicMock() + minio_instance.minio_client = mock_client + + return minio_instance + + +metadata = { + 'metadata': { + 'workflow_name': 'test_workflow', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + } +} + + +@mark.asyncio +async def test_fetch_file_from_minio_success(minio): + """Test successful file fetch from MinIO.""" + # Arrange + test_content = b'test file content' + mock_response = {'Body': MagicMock()} + mock_response['Body'].__enter__ = MagicMock( + return_value=MagicMock(read=MagicMock(return_value=test_content)) + ) + mock_response['Body'].__exit__ = MagicMock(return_value=None) + + minio.minio_client.get_object.return_value = mock_response + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'test-file.txt', + } + + # Act + result = await minio.fetch_file_from_minio(input_data) + + # Assert + assert isinstance(result, BytesIO) + result.seek(0) + assert result.read() == test_content + + minio.minio_client.get_object.assert_called_once_with(Bucket='test-bucket', Key='test-file.txt') + + +@mark.asyncio +async def test_fetch_file_from_minio_file_not_found(minio): + """Test file fetch when file doesn't exist.""" + # Arrange + minio.minio_client.get_object.side_effect = Exception( + 'NoSuchKey: The specified key does not exist' + ) + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'nonexistent.txt', + } + + # Act & Assert + with raises(Exception, match='Error fetching file from MinIO'): + await minio.fetch_file_from_minio(input_data) + + # Verify notification was sent + minio.send_notification.assert_called_once() + call_kwargs = minio.send_notification.call_args[1] + assert call_kwargs['notification_id'] == 'FETCH_FILE_FROM_MINIO_ERROR' + assert call_kwargs['block'] == 'fetch_file_from_minio' + + +@mark.asyncio +async def test_fetch_file_from_minio_network_error(minio): + """Test file fetch with network error.""" + # Arrange + minio.minio_client.get_object.side_effect = Exception('Network timeout') + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'test-file.txt', + } + + # Act & Assert + with raises(Exception, match='Error fetching file from MinIO'): + await minio.fetch_file_from_minio(input_data) + + minio.send_notification.assert_called_once() + + +@mark.asyncio +async def test_delete_file_from_minio_success(minio): + """Test successful file deletion from MinIO.""" + # Arrange + minio.minio_client.delete_object.return_value = None + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'test-file.txt', + } + + # Act + result = await minio.delete_file_from_minio(input_data) + + # Assert + assert result is None + + minio.minio_client.delete_object.assert_called_once_with( + Bucket='test-bucket', Key='test-file.txt' + ) + + +@mark.asyncio +async def test_delete_file_from_minio_idempotent(minio): + """Test that delete is idempotent (no error if file doesn't exist).""" + # Arrange + # MinIO delete_object is idempotent - no error if file doesn't exist + minio.minio_client.delete_object.return_value = None + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'nonexistent.txt', + } + + # Act + result = await minio.delete_file_from_minio(input_data) + + # Assert + assert result is None + minio.minio_client.delete_object.assert_called_once() + + +@mark.asyncio +async def test_delete_file_from_minio_access_denied(minio): + """Test file deletion with access denied error.""" + # Arrange + minio.minio_client.delete_object.side_effect = Exception('AccessDenied: Access Denied') + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'test-file.txt', + } + + # Act & Assert + with raises(Exception, match='Error deleting file from MinIO'): + await minio.delete_file_from_minio(input_data) + + # Verify notification was sent + minio.send_notification.assert_called_once() + call_kwargs = minio.send_notification.call_args[1] + assert call_kwargs['notification_id'] == 'DELETE_FILE_FROM_MINIO_ERROR' + assert call_kwargs['block'] == 'delete_file_from_minio' + + +@mark.asyncio +async def test_delete_file_from_minio_network_error(minio): + """Test file deletion with network error.""" + # Arrange + minio.minio_client.delete_object.side_effect = Exception('Connection timeout') + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'test-file.txt', + } + + # Act & Assert + with raises(Exception, match='Error deleting file from MinIO'): + await minio.delete_file_from_minio(input_data) + + minio.send_notification.assert_called_once() + + +@mark.asyncio +async def test_fetch_file_from_minio_large_file(minio): + """Test fetching a large file from MinIO.""" + # Arrange + # Simulate a 10MB file + large_content = b'x' * (10 * 1024 * 1024) + mock_response = {'Body': MagicMock()} + mock_response['Body'].__enter__ = MagicMock( + return_value=MagicMock(read=MagicMock(return_value=large_content)) + ) + mock_response['Body'].__exit__ = MagicMock(return_value=None) + + minio.minio_client.get_object.return_value = mock_response + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'large-file.bin', + } + + # Act + result = await minio.fetch_file_from_minio(input_data) + + # Assert + assert isinstance(result, BytesIO) + result.seek(0) + assert len(result.read()) == 10 * 1024 * 1024 + + +@mark.asyncio +async def test_fetch_file_from_minio_empty_file(minio): + """Test fetching an empty file from MinIO.""" + # Arrange + empty_content = b'' + mock_response = {'Body': MagicMock()} + mock_response['Body'].__enter__ = MagicMock( + return_value=MagicMock(read=MagicMock(return_value=empty_content)) + ) + mock_response['Body'].__exit__ = MagicMock(return_value=None) + + minio.minio_client.get_object.return_value = mock_response + + input_data = { + 'metadata': metadata['metadata'], + 'bucket_name': 'test-bucket', + 'file_name': 'empty-file.txt', + } + + # Act + result = await minio.fetch_file_from_minio(input_data) + + # Assert + assert isinstance(result, BytesIO) + result.seek(0) + assert result.read() == b'' diff --git a/tests/laborious/utils/repository/test_model_repository.py b/tests/laborious/utils/repository/test_model_repository.py index 116f1e4..a0c013f 100644 --- a/tests/laborious/utils/repository/test_model_repository.py +++ b/tests/laborious/utils/repository/test_model_repository.py @@ -303,13 +303,22 @@ def test_create_model_experiment(set_experiment, sklearn, mlflow_repository): ) +@patch('model_manager.utils.repository.model_repository.path.exists') +@patch('model_manager.utils.repository.model_repository.remove') @patch('model_manager.utils.repository.model_repository.mlflow.start_run') @patch('model_manager.utils.repository.model_repository.mlflow.log_param') @patch('model_manager.utils.repository.model_repository.mlflow.sklearn.log_model') @patch('model_manager.utils.repository.model_repository.mlflow.log_artifact') -def test_perform_model_retrain(log_artifact, log_model, log_param, start_run, mlflow_repository): +def test_perform_model_retrain( + log_artifact, log_model, log_param, start_run, mock_remove, mock_path_exists, mlflow_repository +): + # Create mock models with attributes to test the for loops (lines 268-274) prediction_model_mock = MagicMock() + prediction_model_mock.__dict__ = {'model': 'pred_model', 'param1': 'value1', 'param2': 'value2'} + data_model_mock = MagicMock() + data_model_mock.__dict__ = {'model': 'data_model', 'param3': 'value3', 'param4': 'value4'} + experiment = 'test' model_name = 'test' data = MagicMock() @@ -317,6 +326,7 @@ def test_perform_model_retrain(log_artifact, log_model, log_param, start_run, ml mlflow_repository.get_next_run_name = MagicMock(return_value='test-1') run = MagicMock() start_run.__enter__.return_value = run + mock_path_exists.return_value = True output = mlflow_repository.perform_model_retrain( prediction_model_mock, data_model_mock, experiment, model_name, data @@ -338,12 +348,58 @@ def test_perform_model_retrain(log_artifact, log_model, log_param, start_run, ml log_artifact.assert_called_once_with('temp/raw_data_test.csv') + # Verify that model attributes were logged (excluding 'model' key) log_param.assert_has_calls( [ + call('param1', 'value1'), # from prediction_model + call('param2', 'value2'), # from prediction_model + call('param3', 'value3'), # from data_model + call('param4', 'value4'), # from data_model call('retrain', True), - ] + ], + any_order=True, ) + # Verify temp file cleanup + mock_path_exists.assert_called_once_with('temp/raw_data_test.csv') + mock_remove.assert_called_once_with('temp/raw_data_test.csv') + + assert output == ('Model retrained successfully', experiment) + + +@patch('model_manager.utils.repository.model_repository.path.exists') +@patch('model_manager.utils.repository.model_repository.remove') +@patch('model_manager.utils.repository.model_repository.mlflow.start_run') +@patch('model_manager.utils.repository.model_repository.mlflow.log_param') +@patch('model_manager.utils.repository.model_repository.mlflow.sklearn.log_model') +@patch('model_manager.utils.repository.model_repository.mlflow.log_artifact') +def test_perform_model_retrain_file_not_exists( + log_artifact, log_model, log_param, start_run, mock_remove, mock_path_exists, mlflow_repository +): + """Test perform_model_retrain when temp file doesn't exist (line 291->294 branch).""" + prediction_model_mock = MagicMock() + prediction_model_mock.__dict__ = {'model': 'pred_model'} + + data_model_mock = MagicMock() + data_model_mock.__dict__ = {'model': 'data_model'} + + experiment = 'test' + model_name = 'test' + data = MagicMock() + + mlflow_repository.get_next_run_name = MagicMock(return_value='test-1') + run = MagicMock() + start_run.__enter__.return_value = run + mock_path_exists.return_value = False # File doesn't exist + + output = mlflow_repository.perform_model_retrain( + prediction_model_mock, data_model_mock, experiment, model_name, data + ) + + # Verify temp file cleanup was checked but not executed + mock_path_exists.assert_called_once_with('temp/raw_data_test.csv') + mock_remove.assert_not_called() # Should not be called when file doesn't exist + assert output == ('Model retrained successfully', experiment) diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index 0f5e005..1d943fb 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -1,6 +1,7 @@ from os import environ from model_manager.utils.connectors_config import ( + build_minio_config, build_mlflow_config, build_mongodb_config, build_postgres_config, @@ -113,3 +114,58 @@ def test_build_mongo_db_config_with_defaults(): 'database_name': 'sientia', 'ttl_index_seconds': 3600, } + + +def test_build_minio_config_with_env_vars(): + # Arrange + environ['MINIO_ENDPOINT_URL'] = 'http://test-minio:9000' + environ['MINIO_ACCESS_KEY'] = 'test-access-key' + environ['MINIO_SECRET_KEY'] = 'test-secret-key' + environ['MINIO_REGION'] = 'eu-west-1' + environ['MINIO_USE_SSL'] = 'true' + environ['MINIO_MAX_RETRY_ATTEMPTS'] = '5' + environ['MINIO_RETRY_MODE'] = 'standard' + environ['MINIO_CONNECT_TIMEOUT'] = '20' + environ['MINIO_READ_TIMEOUT'] = '120' + + # Act + config = build_minio_config() + + # Assert + assert config['endpoint_url'] == 'http://test-minio:9000' + assert config['access_key'] == 'test-access-key' + assert config['secret_key'] == 'test-secret-key' + assert config['region'] == 'eu-west-1' + assert config['use_ssl'] is True + assert config['max_retry_attempts'] == 5 + assert config['retry_mode'] == 'standard' + assert config['connect_timeout'] == 20 + assert config['read_timeout'] == 120 + + +def test_build_minio_config_with_defaults(): + # Arrange + # Clear any existing env vars + environ.pop('MINIO_ENDPOINT_URL', None) + environ.pop('MINIO_ACCESS_KEY', None) + environ.pop('MINIO_SECRET_KEY', None) + environ.pop('MINIO_REGION', None) + environ.pop('MINIO_USE_SSL', None) + environ.pop('MINIO_MAX_RETRY_ATTEMPTS', None) + environ.pop('MINIO_RETRY_MODE', None) + environ.pop('MINIO_CONNECT_TIMEOUT', None) + environ.pop('MINIO_READ_TIMEOUT', None) + + # Act + config = build_minio_config() + + # Assert + assert config['endpoint_url'] == 'http://localhost:9000' + assert config['access_key'] == 'minioadmin' + assert config['secret_key'] == 'minioadmin' + assert config['region'] == 'us-east-1' + assert config['use_ssl'] is False + assert config['max_retry_attempts'] == 3 + assert config['retry_mode'] == 'adaptive' + assert config['connect_timeout'] == 10 + assert config['read_timeout'] == 60 diff --git a/validate.sh b/validate.sh index 4bab810..e1d18b9 100755 --- a/validate.sh +++ b/validate.sh @@ -67,7 +67,7 @@ if ! run_step "4. Security Analysis (Bandit)" "bandit -r model_manager/ -ll -q"; fi # Step 5: Unit Tests (pytest) -if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=model_manager --cov-report=term-missing --cov-fail-under=80 -q"; then +if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=model_manager --cov-report=term-missing --cov-report=xml --cov-report=html --cov-fail-under=80 -q"; then FAILED_STEPS+=("Unit Tests") fi