Merge pull request #2 from Aignosi/feature/SIENTIAPDE-1248
SIENTIAPDE-1248: Integrate MinIO, Add Worker Tests, and Refactor Documentation
This commit is contained in:
Bruno Domingues
12
.env.example
12
.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"
|
||||
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"
|
||||
396
README.md
396
README.md
@@ -2,6 +2,67 @@
|
||||
|
||||
A comprehensive AI model management platform for the complete machine learning lifecycle. Handles model training, versioning, deployment, monitoring, and governance. Streamlines MLOps workflows with centralized model registry, automated pipelines, performance tracking, and enterprise-grade compliance features.
|
||||
|
||||
## 📑 Table of Contents
|
||||
|
||||
- [Features](#features)
|
||||
- [Core Functionality](#core-functionality)
|
||||
- [Advanced Capabilities](#advanced-capabilities)
|
||||
- [Development & Quality Assurance](#development--quality-assurance)
|
||||
- [Architecture](#architecture)
|
||||
- [Architecture Principles](#architecture-principles)
|
||||
- [Key Components](#key-components)
|
||||
- [Data Flow Architecture](#data-flow-architecture)
|
||||
- [Security Architecture](#security-architecture)
|
||||
- [Workflows](#workflows)
|
||||
- [Predictions Batch Workflow](#1-predictions-batch-workflow-predictions_batchpy)
|
||||
- [Prediction Process Workflow](#2-prediction-process-workflow-prediction_processpy)
|
||||
- [Format and Export Prediction Workflow](#3-format-and-export-prediction-workflow-format_and_export_predictionpy)
|
||||
- [Minimal Retrain Workflow](#4-minimal-retrain-workflow-minimal_retrainpy)
|
||||
- [Installation & Setup](#installation--setup)
|
||||
- [Prerequisites](#prerequisites)
|
||||
- [Environment Setup](#environment-setup)
|
||||
- [Temporal Namespace Setup](#temporal-namespace-setup)
|
||||
- [Local Development Setup](#local-development-setup)
|
||||
- [How to Run](#how-to-run)
|
||||
- [Running the Model Manager Application](#running-the-model-manager-application)
|
||||
- [Running Tests and Coverage](#running-tests-and-coverage)
|
||||
- [Manual Test Execution](#manual-test-execution)
|
||||
- [Manual Application Execution](#manual-application-execution)
|
||||
- [Code Quality & Validation](#code-quality--validation)
|
||||
- [Overview](#overview)
|
||||
- [Validation Tools](#validation-tools)
|
||||
- [Tools Installation](#tools-installation)
|
||||
- [Complete Validation](#complete-validation)
|
||||
- [Automatic Fixes](#automatic-fixes)
|
||||
- [Configuration](#configuration)
|
||||
- [CI/CD Integration](#cicd-integration)
|
||||
- [Best Practices](#best-practices)
|
||||
- [Testing](#testing)
|
||||
- [Test Structure](#test-structure)
|
||||
- [Test Execution](#test-execution)
|
||||
- [Monitoring and Metrics](#monitoring-and-metrics)
|
||||
- [Application Health Metrics](#application-health-metrics)
|
||||
- [Prediction Operation Metrics](#prediction-operation-metrics)
|
||||
- [Data Quality Metrics](#data-quality-metrics)
|
||||
- [Configuration](#configuration-1)
|
||||
- [Environment Variables](#environment-variables)
|
||||
- [Workflow Configuration](#workflow-configuration)
|
||||
- [Development](#development)
|
||||
- [Code Quality & Testing](#code-quality--testing)
|
||||
- [Project Structure](#project-structure)
|
||||
- [Adding New Features](#adding-new-features)
|
||||
- [Test Coverage Guidelines](#test-coverage-guidelines)
|
||||
- [Troubleshooting](#troubleshooting)
|
||||
- [Common Issues](#common-issues)
|
||||
- [Debug Mode](#debug-mode)
|
||||
- [Performance Tuning](#performance-tuning)
|
||||
- [Key Parameters](#key-parameters)
|
||||
- [Scaling Considerations](#scaling-considerations)
|
||||
- [Contributing](#contributing)
|
||||
- [Code Quality Standards](#code-quality-standards)
|
||||
- [License](#license)
|
||||
- [Support](#support)
|
||||
|
||||
## Features
|
||||
|
||||
### Core Functionality
|
||||
@@ -21,9 +82,11 @@ A comprehensive AI model management platform for the complete machine learning l
|
||||
|
||||
### Development & Quality Assurance
|
||||
- **Code Quality Tools**: Ruff (linting/formatting), mypy (type checking), Bandit (security analysis)
|
||||
- **Automated Validation**: Pre-commit validation script and CI/CD integration
|
||||
- **Comprehensive Testing**: pytest with async support and 70%+ code coverage
|
||||
- **Automated Validation**: Pre-commit validation script (`validate.sh`) and CI/CD integration
|
||||
- **Comprehensive Testing**: pytest with async support and **99%+ code coverage** 🎯
|
||||
- **Type Safety**: Static type checking with mypy for improved code reliability
|
||||
- **Coverage Visualization**: Integration with Coverage Gutters for real-time coverage feedback
|
||||
- **Automated Versioning**: Semantic versioning based on branch patterns (release/*, feature/*, fix/*, rc/*)
|
||||
|
||||
## Architecture
|
||||
|
||||
@@ -142,7 +205,7 @@ Production Update → Notification & Monitoring
|
||||
- **Audit Logging**: Comprehensive access and operation logging
|
||||
- **Data Retention**: Configurable data lifecycle management
|
||||
|
||||
## 🔄 Workflows
|
||||
## Workflows
|
||||
|
||||
### 1. Predictions Batch Workflow (`predictions_batch.py`)
|
||||
|
||||
@@ -331,12 +394,15 @@ flowchart LR
|
||||
D -.-> PostgreSQL[(PostgreSQL)]
|
||||
```
|
||||
|
||||
## 📋 Prerequisites
|
||||
## Installation & Setup
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Python 3.11+
|
||||
- Temporal server/cluster
|
||||
- PostgreSQL database
|
||||
- MLFlow server
|
||||
- MinIO object storage (for MLFlow artifacts)
|
||||
- MongoDB server (for notifications)
|
||||
|
||||
**Note**: External dependencies must be available either through:
|
||||
@@ -345,6 +411,44 @@ flowchart LR
|
||||
- Cloud-managed services
|
||||
- Local installations
|
||||
|
||||
#### MinIO Setup
|
||||
|
||||
MinIO is required for MLFlow artifact storage. For detailed installation and configuration instructions, refer to:
|
||||
|
||||
📚 **[Install MinIO via Helm Chart on K8s](https://aignosi-wiki.atlassian.net/wiki/spaces/IT1/pages/225214465/Install+Minio+via+Helm+Chart+on+K8s)**
|
||||
|
||||
This guide covers:
|
||||
- Helm chart installation on Kubernetes
|
||||
- Storage configuration and persistence
|
||||
- Access credentials setup
|
||||
- Integration with MLFlow
|
||||
|
||||
### Environment Setup
|
||||
|
||||
1. **Clone the repository**:
|
||||
```bash
|
||||
git clone <repository-url>
|
||||
cd sientia-dataops-model-manager
|
||||
```
|
||||
|
||||
2. **Create virtual environment**:
|
||||
```bash
|
||||
python -m venv venv
|
||||
source venv/bin/activate # On Windows: venv\Scripts\activate
|
||||
```
|
||||
|
||||
3. **Install dependencies**:
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
4. **Configure environment variables** (see [Configuration](#-configuration) section)
|
||||
|
||||
5. **Run validation script**:
|
||||
```bash
|
||||
./validate.sh
|
||||
```
|
||||
|
||||
### Temporal Namespace Setup
|
||||
|
||||
The Model Manager requires a dedicated Temporal namespace to isolate workflows and maintain proper execution history. The namespace must be created **before** starting the application.
|
||||
@@ -448,8 +552,6 @@ kubectl exec -n temporal <temporal-admin-tools-pod-name> -- \
|
||||
- Contact cluster administrator for namespace creation
|
||||
- Or request elevated permissions for your service account
|
||||
|
||||
## 🚀 Installation
|
||||
|
||||
### Local Development Setup
|
||||
|
||||
1. **Clone the repository**
|
||||
@@ -512,7 +614,7 @@ kubectl exec -n temporal <temporal-admin-tools-pod-name> -- \
|
||||
# Ensure services are accessible on localhost with appropriate ports
|
||||
```
|
||||
|
||||
## 📦 How to Run
|
||||
## How to Run
|
||||
|
||||
### Running the Model Manager Application
|
||||
|
||||
@@ -585,122 +687,123 @@ fi
|
||||
python -m model_manager.worker.worker
|
||||
```
|
||||
|
||||
## 🔍 Code Quality & Validation
|
||||
## Code Quality & Validation
|
||||
|
||||
### Overview
|
||||
|
||||
Como Python não é uma linguagem compilada, utilizamos um conjunto robusto de ferramentas para validar a qualidade, segurança e correção do código antes da execução. Estas ferramentas detectam erros, problemas de estilo, vulnerabilidades de segurança e garantem a consistência do código.
|
||||
Since Python is not a compiled language, we use a robust set of tools to validate code quality, security, and correctness before execution. These tools detect errors, style issues, security vulnerabilities, and ensure code consistency.
|
||||
|
||||
### Ferramentas de Validação
|
||||
### Validation Tools
|
||||
|
||||
#### 1. **Ruff** - Linting e Formatação ⚡
|
||||
Ferramenta moderna e extremamente rápida (escrita em Rust) que substitui múltiplas ferramentas:
|
||||
- **Linting**: Detecta erros de código, problemas de estilo (PEP 8), bugs comuns
|
||||
- **Formatação**: Formata código automaticamente de forma consistente
|
||||
- **Velocidade**: 10-100x mais rápido que Flake8/Black
|
||||
#### 1. **Ruff** - Linting and Formatting ⚡
|
||||
Modern and extremely fast tool (written in Rust) that replaces multiple tools:
|
||||
- **Linting**: Detects code errors, style issues (PEP 8), common bugs
|
||||
- **Formatting**: Automatically formats code consistently
|
||||
- **Speed**: 10-100x faster than Flake8/Black
|
||||
|
||||
#### 2. **mypy** - Type Checking 🏷️
|
||||
Verificador de tipos estáticos que analisa type hints:
|
||||
- Detecta erros de tipo antes da execução
|
||||
- Melhora a documentação do código
|
||||
- Previne bugs relacionados a tipos incorretos
|
||||
Static type checker that analyzes type hints:
|
||||
- Detects type errors before execution
|
||||
- Improves code documentation
|
||||
- Prevents bugs related to incorrect types
|
||||
|
||||
#### 3. **Bandit** - Análise de Segurança 🔒
|
||||
Scanner de vulnerabilidades de segurança:
|
||||
- Detecta padrões inseguros de código
|
||||
- Identifica hardcoded passwords, SQL injection, etc.
|
||||
- Garante conformidade com práticas de segurança
|
||||
#### 3. **Bandit** - Security Analysis 🔒
|
||||
Security vulnerability scanner:
|
||||
- Detects insecure code patterns
|
||||
- Identifies hardcoded passwords, SQL injection, etc.
|
||||
- Ensures compliance with security practices
|
||||
|
||||
#### 4. **pytest** - Testes Automatizados 🧪
|
||||
Framework de testes com cobertura de código:
|
||||
- Executa testes unitários e de integração
|
||||
- Mede cobertura de código
|
||||
- Suporta testes assíncronos
|
||||
#### 4. **pytest** - Automated Testing 🧪
|
||||
Testing framework with code coverage:
|
||||
- Executes unit and integration tests
|
||||
- Measures code coverage
|
||||
- Supports asynchronous tests
|
||||
|
||||
### Instalação das Ferramentas
|
||||
### Tools Installation
|
||||
|
||||
```bash
|
||||
# Instalar dependências de desenvolvimento
|
||||
# Install development dependencies
|
||||
pip install -r requirements-dev.txt
|
||||
```
|
||||
|
||||
### Validação Completa
|
||||
### Complete Validation
|
||||
|
||||
#### Opção 1: Script Automatizado (Recomendado)
|
||||
#### Option 1: Automated Script (Recommended)
|
||||
```bash
|
||||
# Executar todas as validações de uma vez
|
||||
# Run all validations at once
|
||||
./validate.sh
|
||||
```
|
||||
|
||||
O script `validate.sh` executa automaticamente:
|
||||
1. ✅ Verificação de formatação (Ruff)
|
||||
2. ✅ Linting de código (Ruff)
|
||||
The `validate.sh` script automatically executes:
|
||||
1. ✅ Format checking (Ruff)
|
||||
2. ✅ Code linting (Ruff)
|
||||
3. ✅ Type checking (mypy)
|
||||
4. ✅ Análise de segurança (Bandit)
|
||||
5. ✅ Testes unitários com cobertura (pytest)
|
||||
4. ✅ Security analysis (Bandit)
|
||||
5. ✅ Unit tests with coverage (pytest)
|
||||
|
||||
#### Opção 2: Comandos Individuais
|
||||
#### Option 2: Individual Commands
|
||||
```bash
|
||||
# 1. Verificar formatação
|
||||
# 1. Check formatting
|
||||
ruff format --check model_manager/ tests/
|
||||
|
||||
# 2. Verificar linting
|
||||
# 2. Check linting
|
||||
ruff check model_manager/ tests/
|
||||
|
||||
# 3. Verificar tipos
|
||||
# 3. Check types
|
||||
mypy model_manager/
|
||||
|
||||
# 4. Análise de segurança
|
||||
# 4. Security analysis
|
||||
bandit -r model_manager/ -ll
|
||||
|
||||
# 5. Executar testes
|
||||
# 5. Run tests
|
||||
pytest tests/ --cov=model_manager --cov-report=term-missing
|
||||
```
|
||||
|
||||
### Correção Automática
|
||||
### Automatic Fixes
|
||||
|
||||
Algumas ferramentas podem corrigir problemas automaticamente:
|
||||
Some tools can automatically fix issues:
|
||||
|
||||
```bash
|
||||
# Formatar código automaticamente
|
||||
# Format code automatically
|
||||
ruff format model_manager/ tests/
|
||||
|
||||
# Corrigir problemas de linting automaticamente
|
||||
# Fix linting issues automatically
|
||||
ruff check --fix model_manager/ tests/
|
||||
```
|
||||
|
||||
### Configuração
|
||||
### Configuration
|
||||
|
||||
Todas as ferramentas são configuradas no arquivo `pyproject.toml`:
|
||||
- **Ruff**: Regras de linting, formatação, complexidade
|
||||
- **mypy**: Configurações de type checking
|
||||
- **pytest**: Opções de teste e cobertura
|
||||
- **Bandit**: Regras de segurança
|
||||
All tools are configured in the `pyproject.toml` file:
|
||||
- **Ruff**: Linting rules, formatting, complexity
|
||||
- **mypy**: Type checking settings
|
||||
- **pytest**: Test and coverage options
|
||||
- **Bandit**: Security rules
|
||||
|
||||
### Integração com CI/CD
|
||||
### CI/CD Integration
|
||||
|
||||
O workflow `.github/workflows/quality-gate.yml` executa automaticamente todas as validações em cada push/PR:
|
||||
- ✅ Formatação e linting bloqueiam merge se falharem
|
||||
- ⚠️ Type checking e segurança geram avisos mas não bloqueiam
|
||||
- ✅ Testes devem passar com cobertura mínima de 70%
|
||||
The `.github/workflows/quality-gate.yml` workflow automatically runs all validations on each push/PR:
|
||||
- ✅ Formatting and linting block merge if they fail
|
||||
- ⚠️ Type checking and security generate warnings but don't block
|
||||
- ✅ Tests must pass with minimum 80% coverage
|
||||
|
||||
### Boas Práticas
|
||||
### Best Practices
|
||||
|
||||
1. **Antes de Commit**: Execute `./validate.sh` para garantir qualidade
|
||||
2. **Durante Desenvolvimento**: Use `ruff check --watch` para feedback em tempo real
|
||||
3. **Type Hints**: Adicione type hints em funções novas para melhor validação
|
||||
4. **Testes**: Mantenha cobertura acima de 70%
|
||||
5. **Segurança**: Revise e corrija todos os avisos do Bandit
|
||||
1. **Before Commit**: Run `./validate.sh` to ensure quality
|
||||
2. **During Development**: Use `ruff check --watch` for real-time feedback
|
||||
3. **Type Hints**: Add type hints to new functions for better validation
|
||||
4. **Tests**: Maintain coverage above 80%
|
||||
5. **Security**: Review and fix all Bandit warnings
|
||||
|
||||
## 🧪 Testing
|
||||
## Testing
|
||||
|
||||
### Test Structure
|
||||
```
|
||||
tests/
|
||||
├── activities/ # Activity implementation tests
|
||||
├── workflow/ # Workflow orchestration tests
|
||||
├── utils/ # Utility function tests
|
||||
└── integration/ # End-to-end workflow tests
|
||||
├── laborious/
|
||||
│ ├── activities/ # Activity implementation tests
|
||||
│ ├── workflows/ # Workflow orchestration tests
|
||||
│ ├── utils/ # Utility function tests
|
||||
│ └── worker/ # Worker tests
|
||||
```
|
||||
|
||||
### Test Execution
|
||||
@@ -712,11 +815,11 @@ pip install pytest pytest-cov pytest-asyncio
|
||||
pytest --cov=model_manager --cov-report=html
|
||||
|
||||
# Run specific test modules
|
||||
pytest tests/activities/test_gates.py
|
||||
pytest tests/workflow/test_predictions_batch.py
|
||||
pytest tests/laborious/activities/test_gates.py
|
||||
pytest tests/laborious/workflows/test_predictions_batch.py
|
||||
```
|
||||
|
||||
## 📊 Monitoring and Metrics
|
||||
## Monitoring and Metrics
|
||||
|
||||
The Model Manager system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring:
|
||||
|
||||
@@ -738,7 +841,7 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa
|
||||
- MLFlow API response validation metrics
|
||||
- Data quality gate performance tracking
|
||||
|
||||
## ⚙️ Configuration
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
@@ -757,6 +860,15 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa
|
||||
| `MLFLOW_PORT` | MLFlow server port | `5080` | Yes |
|
||||
| `MLFLOW_USERNAME` | MLFlow username | `aignosi` | Yes |
|
||||
| `MLFLOW_PASSWORD` | MLFlow password | `aignosi` | Yes |
|
||||
| `MINIO_ENDPOINT_URL` | MinIO server endpoint | `http://minio.minio.svc.cluster.local:9000` | Yes |
|
||||
| `MINIO_ACCESS_KEY` | MinIO access key | `minioadmin` | Yes |
|
||||
| `MINIO_SECRET_KEY` | MinIO secret key | `minioadmin` | Yes |
|
||||
| `MINIO_REGION` | MinIO region | `us-east-1` | No |
|
||||
| `MINIO_USE_SSL` | Enable SSL for MinIO | `false` | No |
|
||||
| `MINIO_MAX_RETRY_ATTEMPTS` | Maximum retry attempts | `3` | No |
|
||||
| `MINIO_RETRY_MODE` | Retry mode (standard/adaptive) | `adaptive` | No |
|
||||
| `MINIO_CONNECT_TIMEOUT` | Connection timeout (seconds) | `10` | No |
|
||||
| `MINIO_READ_TIMEOUT` | Read timeout (seconds) | `60` | No |
|
||||
| `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes |
|
||||
| `MONGODB_USERNAME` | MongoDB username | `root` | Yes |
|
||||
| `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes |
|
||||
@@ -856,30 +968,104 @@ This is the configuration created by the Orchestrator in Temporal.
|
||||
}
|
||||
```
|
||||
|
||||
## 🔧 Development
|
||||
## Development
|
||||
|
||||
### Code Quality & Testing
|
||||
|
||||
The project maintains **99%+ code coverage** with comprehensive unit and integration tests.
|
||||
|
||||
#### Running Tests
|
||||
|
||||
```bash
|
||||
# Run all tests with coverage
|
||||
pytest tests/ --cov=model_manager --cov-report=term-missing --cov-report=xml --cov-report=html
|
||||
|
||||
# Run specific test file
|
||||
pytest tests/laborious/activities/test_gates.py -v
|
||||
|
||||
# Run with coverage visualization
|
||||
pytest tests/ --cov=model_manager --cov-report=xml
|
||||
# Then open htmlcov/index.html in browser
|
||||
```
|
||||
|
||||
#### Validation Script
|
||||
|
||||
The `validate.sh` script runs all quality checks before commit:
|
||||
|
||||
```bash
|
||||
./validate.sh
|
||||
```
|
||||
|
||||
This script performs:
|
||||
1. ✅ **Code Formatting** (Ruff)
|
||||
2. ✅ **Code Linting** (Ruff)
|
||||
3. ✅ **Type Checking** (mypy)
|
||||
4. ✅ **Security Analysis** (Bandit)
|
||||
5. ✅ **Unit Tests** (pytest with 80%+ coverage requirement)
|
||||
|
||||
#### Coverage Visualization
|
||||
|
||||
For real-time coverage feedback in VS Code/Windsurf:
|
||||
|
||||
1. **Install Coverage Gutters extension**
|
||||
2. **Configure `.vscode/settings.json`**:
|
||||
```json
|
||||
{
|
||||
"coverage-gutters.coverageBaseDir": "${workspaceFolder}",
|
||||
"coverage-gutters.coverageFileNames": ["coverage.xml"],
|
||||
"coverage-gutters.showLineCoverage": true,
|
||||
"coverage-gutters.showRulerCoverage": true
|
||||
}
|
||||
```
|
||||
3. **Run tests to generate coverage**:
|
||||
```bash
|
||||
pytest tests/ --cov=model_manager --cov-report=xml
|
||||
```
|
||||
4. **Activate Coverage Gutters**: Press `Ctrl+Shift+7` (or `Cmd+Shift+7` on Mac)
|
||||
|
||||
#### Automated Versioning
|
||||
|
||||
The project uses semantic versioning based on branch patterns:
|
||||
|
||||
| Branch Pattern | Version Change | Example |
|
||||
|---------------|----------------|---------|
|
||||
| `release/*` | Major version bump | `2.0.0` |
|
||||
| `feature/*` | Minor version bump | `1.2.0` |
|
||||
| `fix/*` | Patch version bump | `1.1.3` |
|
||||
| `rc/*` | Release candidate | `1.1.2-rc2` |
|
||||
|
||||
Version is calculated automatically in the CI/CD pipeline and passed to SonarQube.
|
||||
|
||||
### Project Structure
|
||||
```
|
||||
model_manager/
|
||||
├── activities/ # Temporal activity implementations
|
||||
│ ├── activities.py # Main activities orchestrator
|
||||
│ ├── gates.py # Data quality gates and filtering
|
||||
│ └── mlflow.py # MLFlow model operations
|
||||
│ ├── __init__.py
|
||||
│ ├── activities.py # Main activities orchestrator (combines all activities)
|
||||
│ ├── gates.py # Data quality gates and filtering logic
|
||||
│ ├── minio.py # MinIO object storage operations
|
||||
│ └── mlflow.py # MLFlow model operations (predict/transform)
|
||||
├── workflows/ # Temporal workflow definitions
|
||||
│ ├── predictions_batch.py # Main batch prediction workflow
|
||||
│ ├── __init__.py
|
||||
│ ├── predictions_batch.py # Main batch prediction workflow entry point
|
||||
│ ├── minimal_retrain.py # Model retraining workflow
|
||||
│ └── sub_workflows/ # Sub-workflow implementations
|
||||
│ ├── prediction_process.py # Core prediction workflow
|
||||
│ └── format_and_export_prediction.py # Export workflow
|
||||
│ ├── __init__.py
|
||||
│ ├── prediction_process.py # Core prediction pipeline
|
||||
│ └── format_and_export_prediction.py # Data export workflow
|
||||
├── worker/ # Worker implementation
|
||||
│ └── worker.py # Main worker orchestrator
|
||||
├── utils/ # Utility functions
|
||||
│ ├── connectors_config.py # Database configuration
|
||||
│ ├── filters/ # Data quality filters
|
||||
│ │ ├── conditional_filters.py # Conditional data filters
|
||||
│ │ └── mlflow_filters.py # MLFlow response filters
|
||||
│ ├── __init__.py
|
||||
│ └── worker.py # Main worker orchestrator (Temporal client setup)
|
||||
├── utils/ # Utility functions and helpers
|
||||
│ ├── __init__.py
|
||||
│ ├── connectors_config.py # Environment-based configuration builders
|
||||
│ ├── filters/ # Data quality validation filters
|
||||
│ │ ├── __init__.py
|
||||
│ │ ├── conditional_filters.py # Input data validation filters
|
||||
│ │ └── mlflow_filters.py # MLFlow response validation filters
|
||||
│ └── repository/ # Data access layer
|
||||
│ └── model_repository.py # MLFlow model operations
|
||||
│ ├── __init__.py
|
||||
│ └── model_repository.py # MLFlow model operations and retraining
|
||||
├── metrics.py # Prometheus metrics definitions
|
||||
└── __init__.py
|
||||
```
|
||||
@@ -889,10 +1075,28 @@ model_manager/
|
||||
1. **Follow Temporal patterns** for new workflows and activities
|
||||
2. **Add comprehensive docstrings** for all public methods
|
||||
3. **Include Prometheus metrics** for monitoring
|
||||
4. **Add unit tests** for new functionality
|
||||
5. **Update this README** with new features and configuration
|
||||
4. **Add unit tests** for new functionality (maintain 80%+ coverage)
|
||||
5. **Run validation script** (`./validate.sh`) before committing
|
||||
6. **Update this README** with new features and configuration
|
||||
|
||||
## 🐛 Troubleshooting
|
||||
### Test Coverage Guidelines
|
||||
|
||||
- **Minimum coverage**: 80% (enforced by CI/CD)
|
||||
- **Current coverage**: 99%+ 🎯
|
||||
- **Test all branches**: Use Coverage Gutters to identify uncovered lines
|
||||
- **Mock external dependencies**: Use `unittest.mock` for external services
|
||||
- **Async testing**: Use `pytest-asyncio` for async activities and workflows
|
||||
- **Test structure**:
|
||||
```
|
||||
tests/
|
||||
├── laborious/
|
||||
│ ├── activities/ # Activity tests
|
||||
│ ├── workflows/ # Workflow tests
|
||||
│ ├── utils/ # Utility tests
|
||||
│ └── worker/ # Worker tests
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
@@ -923,7 +1127,7 @@ Enable debug logging by setting the log level:
|
||||
export LOG_LEVEL=DEBUG
|
||||
```
|
||||
|
||||
## ⚡ Performance Tuning
|
||||
## Performance Tuning
|
||||
|
||||
### Key Parameters
|
||||
|
||||
@@ -939,7 +1143,7 @@ export LOG_LEVEL=DEBUG
|
||||
- **Database Performance**: Optimize indexes and connection pooling
|
||||
- **MLFlow Performance**: Configure appropriate model serving resources
|
||||
|
||||
## 🤝 Contributing
|
||||
## Contributing
|
||||
|
||||
1. Fork the repository
|
||||
2. Create a feature branch
|
||||
@@ -955,11 +1159,11 @@ export LOG_LEVEL=DEBUG
|
||||
- Use type hints where appropriate
|
||||
- Follow Temporal.io best practices
|
||||
|
||||
## 📄 License
|
||||
## License
|
||||
|
||||
This project is licensed under the terms specified in the LICENSE file.
|
||||
|
||||
## 🆘 Support
|
||||
## Support
|
||||
|
||||
For support and questions:
|
||||
- Check the troubleshooting section above
|
||||
|
||||
@@ -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):
|
||||
|
||||
229
model_manager/activities/minio.py
Normal file
229
model_manager/activities/minio.py
Normal file
@@ -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:
|
||||
ConnectionError: 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 ConnectionError(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:
|
||||
OSError: 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 OSError(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:
|
||||
OSError: 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 OSError(error_msg) from e
|
||||
@@ -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')),
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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'
|
||||
|
||||
328
tests/laborious/activities/test_minio.py
Normal file
328
tests/laborious/activities/test_minio.py
Normal file
@@ -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(ConnectionError, 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(OSError, 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(OSError, 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(OSError, 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(OSError, 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''
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
499
tests/laborious/worker/test_worker.py
Normal file
499
tests/laborious/worker/test_worker.py
Normal file
@@ -0,0 +1,499 @@
|
||||
from unittest.mock import ANY, AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_env_vars(monkeypatch):
|
||||
"""Fixture to set up environment variables for tests."""
|
||||
monkeypatch.setenv('POD_ID', 'test-pod-123')
|
||||
monkeypatch.setenv('TEMPORAL_HOST', 'test-temporal:7233')
|
||||
monkeypatch.setenv('TEMPORAL_NAMESPACE', 'test-namespace')
|
||||
monkeypatch.setenv('HTTP_METRICS_PORT', '9090')
|
||||
monkeypatch.setenv('HTTP_SDK_METRICS_PORT', '9091')
|
||||
monkeypatch.setenv('PROJECT_NAME', 'test-project')
|
||||
monkeypatch.setenv('POSTGRES_HOST', 'localhost')
|
||||
monkeypatch.setenv('POSTGRES_PORT', '5432')
|
||||
monkeypatch.setenv('POSTGRES_USER', 'test')
|
||||
monkeypatch.setenv('POSTGRES_PASSWORD', 'test')
|
||||
monkeypatch.setenv('POSTGRES_DBNAME', 'test')
|
||||
monkeypatch.setenv('MLFLOW_HOST', 'http://localhost')
|
||||
monkeypatch.setenv('MLFLOW_PORT', '5000')
|
||||
monkeypatch.setenv('MLFLOW_USERNAME', 'test')
|
||||
monkeypatch.setenv('MLFLOW_PASSWORD', 'test')
|
||||
monkeypatch.setenv('MINIO_ENDPOINT_URL', 'http://localhost:9000')
|
||||
monkeypatch.setenv('MINIO_ACCESS_KEY', 'test')
|
||||
monkeypatch.setenv('MINIO_SECRET_KEY', 'test')
|
||||
monkeypatch.setenv('MONGODB_USERNAME', 'test')
|
||||
monkeypatch.setenv('MONGODB_PASSWORD', 'test')
|
||||
monkeypatch.setenv('MONGODB_URL', 'localhost:27017')
|
||||
monkeypatch.setenv('MONGODB_DATABASE_NAME', 'test')
|
||||
|
||||
|
||||
@patch('model_manager.worker.worker.start_http_server')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
def test_start_prometheus_server_success(mock_metrics, mock_start_http_server, mock_env_vars):
|
||||
"""Test successful Prometheus server startup."""
|
||||
# Import after patching to ensure mocks are in place
|
||||
from model_manager.worker.worker import start_prometheus_server
|
||||
|
||||
# Act
|
||||
start_prometheus_server()
|
||||
|
||||
# Assert
|
||||
mock_start_http_server.assert_called_once_with(9090)
|
||||
mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123')
|
||||
mock_metrics.APP_UP.labels.return_value.set.assert_called_once_with(1)
|
||||
|
||||
|
||||
@patch('model_manager.worker.worker.start_http_server')
|
||||
@patch('model_manager.worker.worker.os._exit')
|
||||
def test_start_prometheus_server_failure(mock_exit, mock_start_http_server, mock_env_vars):
|
||||
"""Test Prometheus server startup failure."""
|
||||
# Arrange
|
||||
mock_start_http_server.side_effect = Exception('Port already in use')
|
||||
|
||||
# Import after patching
|
||||
from model_manager.worker.worker import start_prometheus_server
|
||||
|
||||
# Act
|
||||
start_prometheus_server()
|
||||
|
||||
# Assert
|
||||
mock_start_http_server.assert_called_once_with(9090)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.asyncio.gather')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_success(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_gather,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test successful main function execution."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Mock gather to complete successfully
|
||||
mock_gather.return_value = None
|
||||
|
||||
# Import and run
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert
|
||||
mock_start_prometheus.assert_called_once()
|
||||
mock_notification_handler.assert_called_once()
|
||||
mock_activities.assert_called_once()
|
||||
mock_client_connect.assert_called_once_with(
|
||||
target_host='test-temporal:7233',
|
||||
namespace='test-namespace',
|
||||
runtime=ANY,
|
||||
)
|
||||
assert mock_worker.call_count == 2 # Two workers created
|
||||
mock_gather.assert_called_once()
|
||||
mock_sys_exit.assert_called_once_with(1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.asyncio.gather')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
async def test_main_exception_handling(
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_gather,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test main function exception handling and cleanup."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Mock gather to raise an exception
|
||||
mock_gather.side_effect = Exception('Worker failed')
|
||||
|
||||
# Import and run
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - Verify cleanup was performed
|
||||
mock_logger.custom_error.assert_called_once()
|
||||
mock_handler.shutdown.assert_called_once()
|
||||
mock_activities_instance.shutdown.assert_called_once()
|
||||
mock_metrics.APP_UP.labels.assert_called_once_with(pod_id='test-pod-123')
|
||||
mock_metrics.APP_UP.labels.return_value.set.assert_called_once_with(0)
|
||||
mock_sys_exit.assert_called_once_with(1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_creates_two_workers(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test that main creates two workers with correct configurations."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Import
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Mock gather to prevent infinite wait
|
||||
with patch('model_manager.worker.worker.asyncio.gather', new_callable=AsyncMock):
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - Verify two workers were created
|
||||
assert mock_worker.call_count == 2
|
||||
|
||||
# Verify first worker (minimal_retrain-queue)
|
||||
first_call = mock_worker.call_args_list[0]
|
||||
assert first_call[1]['task_queue'] == 'minimal_retrain-queue'
|
||||
assert 'MinimalRetrain' in str(first_call[1]['workflows'])
|
||||
|
||||
# Verify second worker (predictions_batch-queue)
|
||||
second_call = mock_worker.call_args_list[1]
|
||||
assert second_call[1]['task_queue'] == 'predictions_batch-queue'
|
||||
assert 'PredictionsBatch' in str(second_call[1]['workflows'])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_initializes_activities_with_configs(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test that main initializes Activities with correct configurations."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Import
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Mock gather to prevent infinite wait
|
||||
with patch('model_manager.worker.worker.asyncio.gather', new_callable=AsyncMock):
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - Verify Activities was initialized with correct parameters
|
||||
mock_activities.assert_called_once()
|
||||
call_kwargs = mock_activities.call_args[1]
|
||||
assert 'postgres_config' in call_kwargs
|
||||
assert 'mlflow_config' in call_kwargs
|
||||
assert 'minio_config' in call_kwargs
|
||||
assert call_kwargs['logger'] == mock_logger
|
||||
assert call_kwargs['notification_handler'] == mock_handler
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_uses_environment_variables(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test that main uses environment variables correctly."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Import
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Mock gather to prevent infinite wait
|
||||
with patch('model_manager.worker.worker.asyncio.gather', new_callable=AsyncMock):
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - Verify environment variables were used
|
||||
mock_client_connect.assert_called_once_with(
|
||||
target_host='test-temporal:7233',
|
||||
namespace='test-namespace',
|
||||
runtime=ANY,
|
||||
)
|
||||
|
||||
mock_notification_handler.assert_called_once()
|
||||
notification_call_kwargs = mock_notification_handler.call_args[1]
|
||||
assert notification_call_kwargs['project_name'] == 'test-project'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.asyncio.gather')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_cleanup_with_none_notification_handler(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_gather,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test cleanup when notification_handler is None (line 194 branch False)."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
# Return None for notification_handler
|
||||
mock_notification_handler.return_value = None
|
||||
|
||||
mock_activities_instance = MagicMock()
|
||||
mock_activities_instance.shutdown = AsyncMock()
|
||||
mock_activities.return_value = mock_activities_instance
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Mock gather to complete
|
||||
mock_gather.return_value = None
|
||||
|
||||
# Import and run
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - notification_handler.shutdown() should NOT be called (line 194 False)
|
||||
# Since notification_handler is None, we can't call shutdown on it
|
||||
mock_activities_instance.shutdown.assert_called_once()
|
||||
mock_sys_exit.assert_called_once_with(1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@patch('model_manager.worker.worker.sys.exit')
|
||||
@patch('model_manager.worker.worker.metrics')
|
||||
@patch('model_manager.worker.worker.asyncio.gather')
|
||||
@patch('model_manager.worker.worker.Worker')
|
||||
@patch('model_manager.worker.worker.client.Client.connect')
|
||||
@patch('model_manager.worker.worker.Runtime')
|
||||
@patch('model_manager.worker.worker.Activities')
|
||||
@patch('model_manager.worker.worker.NotificationHandler')
|
||||
@patch('model_manager.worker.worker.start_prometheus_server')
|
||||
@patch('model_manager.worker.worker.get_logger')
|
||||
async def test_main_cleanup_with_falsy_activities(
|
||||
mock_get_logger,
|
||||
mock_start_prometheus,
|
||||
mock_notification_handler,
|
||||
mock_activities,
|
||||
mock_runtime,
|
||||
mock_client_connect,
|
||||
mock_worker,
|
||||
mock_gather,
|
||||
mock_metrics,
|
||||
mock_sys_exit,
|
||||
mock_env_vars,
|
||||
):
|
||||
"""Test cleanup when activities evaluates to False (line 196 branch False)."""
|
||||
# Arrange
|
||||
mock_logger = MagicMock()
|
||||
mock_get_logger.return_value = mock_logger
|
||||
|
||||
mock_handler = MagicMock()
|
||||
mock_notification_handler.return_value = mock_handler
|
||||
|
||||
# Create a falsy activities object (empty list, 0, False, etc.)
|
||||
# Using an object that evaluates to False but doesn't cause AttributeError
|
||||
class FalsyActivities:
|
||||
def __bool__(self):
|
||||
return False
|
||||
|
||||
def __getattr__(self, name):
|
||||
# Return mock methods to avoid AttributeError during worker creation
|
||||
return MagicMock()
|
||||
|
||||
falsy_activities = FalsyActivities()
|
||||
mock_activities.return_value = falsy_activities
|
||||
|
||||
mock_temporal_client = AsyncMock()
|
||||
mock_client_connect.return_value = mock_temporal_client
|
||||
|
||||
mock_worker_instance = MagicMock()
|
||||
mock_worker_instance.run = MagicMock(return_value=AsyncMock())
|
||||
mock_worker.return_value = mock_worker_instance
|
||||
|
||||
# Mock gather to complete
|
||||
mock_gather.return_value = None
|
||||
|
||||
# Import and run
|
||||
from model_manager.worker.worker import main
|
||||
|
||||
# Act
|
||||
await main()
|
||||
|
||||
# Assert - notification_handler.shutdown() is called, but activities.shutdown() is NOT
|
||||
mock_handler.shutdown.assert_called_once()
|
||||
# activities is falsy, so shutdown should NOT be called
|
||||
mock_sys_exit.assert_called_once_with(1)
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user