From b102f790878d4f300a230691745c9482fc13939f Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 1 Oct 2025 16:25:08 -0300 Subject: [PATCH] SIENTIAPDE-1243: Remove OPC server integration and add code quality tools. This commit removes the OPC server integration from the Model Manager, including related activities, repositories, metrics, and configuration. It also adds code quality tools such as Ruff (linting/formatting), mypy (type checking), and Bandit (security analysis) along with a validation script and CI/CD integration for automated code validation. The README has been updated to reflect these changes. --- .env.example | 3 - .github/workflows/quality-gate.yml | 26 +- README.md | 235 ++++++----- model_manager/activities/activities.py | 17 +- model_manager/activities/opc.py | 356 ---------------- model_manager/metrics.py | 17 +- model_manager/utils/connectors_config.py | 39 -- .../utils/repository/opc_repository.py | 359 ---------------- model_manager/worker/worker.py | 12 +- model_manager/workflows/predictions_batch.py | 4 +- .../format_and_export_prediction.py | 24 +- .../sub_workflows/prediction_process.py | 4 +- pyproject.toml | 154 +++++++ requirements-dev.txt | 17 + requirements.txt | 1 - tests/laborious/activities/test_activities.py | 31 +- tests/laborious/activities/test_opc.py | 369 ----------------- .../utils/repository/test_opc_repository.py | 388 ------------------ .../laborious/utils/test_connectors_config.py | 49 --- .../test_format_and_export_prediction.py | 32 +- .../subworkflows/test_prediction_process.py | 18 +- .../workflows/test_predictions_batch.py | 4 +- validate.sh | 99 +++++ values.yaml | 5 - 24 files changed, 453 insertions(+), 1810 deletions(-) delete mode 100644 model_manager/activities/opc.py delete mode 100644 model_manager/utils/repository/opc_repository.py create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt delete mode 100644 tests/laborious/activities/test_opc.py delete mode 100644 tests/laborious/utils/repository/test_opc_repository.py create mode 100755 validate.sh diff --git a/.env.example b/.env.example index 8bd75e5..652ce20 100644 --- a/.env.example +++ b/.env.example @@ -11,9 +11,6 @@ MLFLOW_PORT="80" MLFLOW_USERNAME="aignosi" MLFLOW_PASSWORD="mlflow_password" -OPC_ID="1" -OPC_URL="opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840" - LOG_LEVEL="DEBUG" HTTP_METRICS_PORT="9090" HTTP_SDK_METRICS_PORT="9091" diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 3eaa27c..997202a 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -193,7 +193,31 @@ jobs: run: | python -m pip install --upgrade pip pip install -r ${{ steps.prepare-requirements.outputs.PROCESSED_REQUIREMENTS_FILE }} - pip install pytest pytest-cov pytest-asyncio + pip install -r requirements-dev.txt + + - name: 📝 Code Formatting Check (Ruff) + run: | + echo "Checking code formatting..." + ruff format --check model_manager/ tests/ + continue-on-error: false + + - name: 🔎 Code Linting (Ruff) + run: | + echo "Running linting checks..." + ruff check model_manager/ tests/ + continue-on-error: false + + - name: 🏷️ Type Checking (mypy) + run: | + echo "Running type checks..." + mypy model_manager/ + continue-on-error: true + + - name: 🔒 Security Analysis (Bandit) + run: | + echo "Running security analysis..." + bandit -r model_manager/ -ll -q + continue-on-error: true - name: 🧪 Run Tests with Pytest run: | diff --git a/README.md b/README.md index 61a50b9..28a38c3 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ A comprehensive AI model management platform for the complete machine learning l - **Temporal Workflow Orchestration**: Robust workflow management with automatic retry policies and fault tolerance - **Data Quality Gates**: Configurable filtering for data validation, MLFlow API responses, and custom validation rules - **Multi-Model Support**: Flexible ML model management with retention policies and versioning -- **Real-time Data Export**: PostgreSQL persistence and OPC server integration for industrial systems +- **Real-time Data Export**: PostgreSQL persistence for data storage - **Comprehensive Monitoring**: Prometheus metrics and detailed logging for operational visibility ### Advanced Capabilities @@ -19,6 +19,12 @@ A comprehensive AI model management platform for the complete machine learning l - **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support - **Model Retraining**: Automated model retraining workflows with production model updates +### 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 +- **Type Safety**: Static type checking with mypy for improved code reliability + ## Architecture The Model Manager system uses a Temporal-based workflow architecture with clear separation of concerns and robust error handling. The architecture is designed for high availability, scalability, and operational excellence in production ML environments. @@ -60,7 +66,6 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Task queue configuration and load balancing - Prometheus metrics server initialization - Notification handler setup and configuration - - OPC server connection management - **Key Features**: - Automatic scaling with `PollerBehaviorAutoscaling` - Health check endpoints for Kubernetes liveness/readiness probes @@ -83,20 +88,16 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - **Activities**: Main activity orchestrator combining all functionality through multiple inheritance - **Gates**: Data quality validation and filtering mechanisms - **MLFlow**: Model transformation and prediction operations -- **OPC**: Real-time data export to industrial OPC servers - **Key Features**: - Multiple inheritance pattern for unified activity interface - Configurable filter policies and validation rules - MLFlow model serving integration with configurable flavors - - OPC UA client with certificate-based authentication - Comprehensive error handling and notification integration - - Support for multiple OPC servers with independent configurations #### **Data Services (`model_manager/utils/`)** - **Connectors Config**: Environment variable-based configuration management -- **Repository**: Data access layer for MLFlow and OPC operations +- **Repository**: Data access layer for MLFlow operations - `model_repository.py`: MLFlow model operations and retraining - - `opc_repository.py`: OPC server communication and data writing - **Filters**: Data quality validation and MLFlow response filtering - `conditional_filters.py`: Input data validation filters - `mlflow_filters.py`: MLFlow API response validation filters @@ -105,14 +106,14 @@ The Model Manager system uses a Temporal-based workflow architecture with clear - Connection pool management and optimization - Security credential management - Configuration validation and error handling - - Support for multiple OPC servers and MLFlow model flavors + - Support for MLFlow model flavors ### Data Flow Architecture #### **1. Batch Prediction Pipeline** ``` Input Data (PostgreSQL) → Data Quality Gates → MLFlow Transform → -MLFlow Prediction → Response Validation → Export (PostgreSQL + OPC) +MLFlow Prediction → Response Validation → Export (PostgreSQL) ``` #### **2. Model Retraining Pipeline** @@ -121,16 +122,10 @@ Training Data → Model Retraining → Quality Validation → Production Update → Notification & Monitoring ``` -#### **3. Real-time Export Pipeline** -``` -Prediction Results → Data Formatting → OPC Server Write → -Success/Failure Metrics → Notification System -``` - ### Security Architecture #### **Authentication & Authorization** -- **Certificate-based OPC Authentication**: Secure industrial communication + - **MLFlow API Authentication**: Username/password with secure transmission - **Database Connection Security**: Encrypted connections with credential management - **Kubernetes Secrets Integration**: Secure credential storage and access @@ -195,11 +190,7 @@ The **PredictionsBatch** workflow is the main entry point for batch prediction p "API_ERROR": {"POLICY": "STOP"} }, "model_retention": 60, - "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "opc_output_config": { - "server_id": "opc_server_1", - "tags": ["prediction_output"] - } + "path_priority": ["STOP", "CONTINUE", "REPEAT"] } ``` @@ -267,8 +258,7 @@ The **PredictionProcess** workflow implements the core prediction pipeline for M "NAN_VALUES": {"POLICY": "STOP"} }, "model_retention": 60, - "path_priority": ["STOP", "CONTINUE", "REPEAT"], - "opc_output_config": {...} + "path_priority": ["STOP", "CONTINUE", "REPEAT"] } ``` @@ -288,33 +278,30 @@ flowchart LR The **FormatAndExportPrediction** workflow handles prediction data formatting and export operations to multiple destinations. #### Purpose -- **Data Formatting**: Formats prediction data for different output destinations +- **Data Formatting**: Formats prediction data for database storage - **PostgreSQL Export**: Persists predictions to database with metrics -- **OPC Integration**: Writes predictions to OPC servers for real-time access - **Metrics Recording**: Tracks export operations and performance metrics #### Execution Flow 1. **Path Decision**: Determines formatting path based on configuration 2. **Data Formatting**: Formats prediction data for specific output requirements 3. **PostgreSQL Export**: Writes formatted predictions to database -4. **OPC Export**: Writes predictions to OPC servers -5. **Metrics Recording**: Records export performance and success metrics +4. **Metrics Recording**: Records export performance and success metrics #### Key Features - **Flexible Formatting**: Configurable output formats for different destinations -- **Multi-Destination Export**: PostgreSQL and OPC server integration +- **Database Export**: PostgreSQL integration for data persistence - **Performance Monitoring**: Comprehensive metrics for export operations - **Error Handling**: Robust error handling with notification integration #### Architecture Diagram ```mermaid flowchart LR - A[1. format_prediction/format_default_prediction] --> B[2. write_opc_data] --> C[3. export_data_to_postgres] --> D[4. write_metrics] + A[1. format_prediction/format_default_prediction] --> B[2. export_data_to_postgres] --> C[3. write_metrics] A -.-> Format[Data Formatting] - B -.-> OPC[OPC Servers] - C -.-> PostgreSQL[(PostgreSQL)] - D -.-> Prometheus[Prometheus] + B -.-> PostgreSQL[(PostgreSQL)] + C -.-> Prometheus[Prometheus] ``` ### 4. Minimal Retrain Workflow (`minimal_retrain.py`) @@ -351,7 +338,6 @@ flowchart LR - Temporal server/cluster - PostgreSQL database - MLFlow server -- OPC server(s) - MongoDB server (for notifications) **Note**: External dependencies must be available either through: @@ -398,12 +384,10 @@ flowchart LR 4. **Install Python dependencies** ```bash python -m pip install --upgrade pip + # Install production dependencies pip install -r requirements.txt - ``` - - 5. **Install test libraries** - ```bash - pip install pytest pytest-cov pytest-asyncio + # Install development and testing tools + pip install -r requirements-dev.txt ``` 4. **Create environment configuration file** @@ -499,6 +483,113 @@ fi python -m model_manager.worker.worker ``` +## 🔍 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. + +### Ferramentas de Validação + +#### 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 + +#### 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 + +#### 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 + +#### 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 + +### Instalação das Ferramentas + +```bash +# Instalar dependências de desenvolvimento +pip install -r requirements-dev.txt +``` + +### Validação Completa + +#### Opção 1: Script Automatizado (Recomendado) +```bash +# Executar todas as validações de uma vez +./validate.sh +``` + +O script `validate.sh` executa automaticamente: +1. ✅ Verificação de formatação (Ruff) +2. ✅ Linting de código (Ruff) +3. ✅ Type checking (mypy) +4. ✅ Análise de segurança (Bandit) +5. ✅ Testes unitários com cobertura (pytest) + +#### Opção 2: Comandos Individuais +```bash +# 1. Verificar formatação +ruff format --check model_manager/ tests/ + +# 2. Verificar linting +ruff check model_manager/ tests/ + +# 3. Verificar tipos +mypy model_manager/ + +# 4. Análise de segurança +bandit -r model_manager/ -ll + +# 5. Executar testes +pytest tests/ --cov=model_manager --cov-report=term-missing +``` + +### Correção Automática + +Algumas ferramentas podem corrigir problemas automaticamente: + +```bash +# Formatar código automaticamente +ruff format model_manager/ tests/ + +# Corrigir problemas de linting automaticamente +ruff check --fix model_manager/ tests/ +``` + +### Configuração + +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 + +### Integração com CI/CD + +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% + +### Boas Práticas + +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 + ## 🧪 Testing ### Test Structure @@ -540,13 +631,6 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa - Labels: `pod_id`, `model_name`, `pipeline_name` - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] -### OPC Export Metrics -- `model_manager_prediction_opc_writing_count`: Counter for OPC server write operations - - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` -- `model_manager_prediction_opc_writing_response_time_monitor`: Histogram for OPC write response times - - Labels: `pod_id`, `model_name`, `pipeline_name`, `opc_server_id` - - Buckets: [0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] - ### Data Quality Metrics - Filter pass/fail rates through notification system - MLFlow API response validation metrics @@ -571,14 +655,6 @@ 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 | -| `OPC_CONFIG` | OPC server configuration (JSON) | `{}` | No | -| `OPC_ID` | OPC server identifier | `1` | No | -| `OPC_URL` | OPC server URL | `opc.tcp://localhost:4840` | No | -| `OPC_SERVER_URI` | OPC server URI | `opc.tcp://localhost:4840` | No | -| `OPC_CERT_PATH` | OPC client certificate path | `None` | No | -| `OPC_PRIVATE_KEY_PATH` | OPC private key path | `None` | No | -| `OPC_SERVER_CERT_PATH` | OPC server certificate path | `None` | No | -| `OPC_RECONNECTION_INTERVAL` | OPC reconnection interval (ms) | `120` | No | | `MONGODB_URL` | MongoDB connection URI | `localhost:27018` | Yes | | `MONGODB_USERNAME` | MongoDB username | `root` | Yes | | `MONGODB_PASSWORD` | MongoDB password | `wKZDbMNU1c` | Yes | @@ -590,45 +666,6 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa | `HTTP_SDK_METRICS_PORT` | Temporal SDK metrics port | `9091` | No | | `POD_ID` | Kubernetes pod identifier | `None` | No | - - - -### OPC Configuration - -For multiple OPC servers, use the `OPC_CONFIG` environment variable: - -```json -{ - "opc_server_1": { - "url": "opc.tcp://server1:4840", - "name": "Server1", - "server_uri": "urn:server1:opcua", - "cert_path": "/path/to/cert.pem", - "private_key_path": "/path/to/key.pem", - "server_cert_path": "/path/to/server_cert.pem", - "reconnection_interval": 5000 - }, - "opc_server_2": { - "url": "opc.tcp://server2:4840", - "name": "Server2", - "server_uri": "urn:server2:opcua", - "cert_path": "/path/to/cert.pem", - "private_key_path": "/path/to/key.pem", - "server_cert_path": "/path/to/server_cert.pem", - "reconnection_interval": 5000 - } -} -``` - -For single OPC server, use individual environment variables: -- `OPC_URL` -- `OPC_NAME` -- `OPC_SERVER_URI` -- `OPC_CERT_PATH` -- `OPC_PRIVATE_KEY_PATH` -- `OPC_SERVER_CERT_PATH` -- `OPC_RECONNECTION_INTERVAL` - ### Workflow Configuration MongoDB pipeline configuration: @@ -705,7 +742,6 @@ This is the configuration created by the Orchestrator in Temporal. }, "model_id":"352", "model_name":"courier", - "opc_output_config":{}, "path_priority":["STOP","CONTINUE","REPEAT"], "predictions_storage_policy":"lts:1", "query":"select * from sientia_data.laborious_data where model_id = 352 order by \"timestamp\" desc limit 300;", @@ -726,8 +762,7 @@ model_manager/ ├── activities/ # Temporal activity implementations │ ├── activities.py # Main activities orchestrator │ ├── gates.py # Data quality gates and filtering -│ ├── mlflow.py # MLFlow model operations -│ └── opc.py # OPC server operations +│ └── mlflow.py # MLFlow model operations ├── workflows/ # Temporal workflow definitions │ ├── predictions_batch.py # Main batch prediction workflow │ ├── minimal_retrain.py # Model retraining workflow @@ -742,8 +777,7 @@ model_manager/ │ │ ├── conditional_filters.py # Conditional data filters │ │ └── mlflow_filters.py # MLFlow response filters │ └── repository/ # Data access layer -│ ├── model_repository.py # MLFlow model operations -│ └── opc_repository.py # OPC server operations +│ └── model_repository.py # MLFlow model operations ├── metrics.py # Prometheus metrics definitions └── __init__.py ``` @@ -775,12 +809,7 @@ model_manager/ - Check connection credentials and network access - Ensure proper connection pool configuration -4. **OPC Connection Failures** - - Verify OPC server is accessible - - Check certificate and key file paths - - Review OPC server logs for connection issues - -5. **Workflow Execution Failures** +4. **Workflow Execution Failures** - Review activity error logs and notifications - Check data quality filter configurations - Verify input data format and required fields diff --git a/model_manager/activities/activities.py b/model_manager/activities/activities.py index 6a8656d..81407a7 100644 --- a/model_manager/activities/activities.py +++ b/model_manager/activities/activities.py @@ -6,28 +6,25 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.observability.logger import Logger from model_manager.activities.mlflow import MLFlow from model_manager.activities.gates import Gates - from model_manager.activities.opc import OPC from typing import Any -class Activities(Postgres, MLFlow, Gates, OPC): +class Activities(Postgres, MLFlow, 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, data quality validation, and OPC server communications. + MLFlow model interactions, 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 - Gates: Data quality validation and filtering mechanisms - - OPC: Real-time data export to OPC servers Attributes: postgres_config (dict): PostgreSQL connection configuration mlflow_config (dict): MLFlow server configuration - opc_config (dict): OPC server configuration logger (Logger): Logging and observability instance notification_handler (NotificationHandler): Notification management instance """ @@ -35,7 +32,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): def __init__(self, postgres_config: dict[str, Any], mlflow_config: dict[str, Any], - opc_config: dict[str, Any], logger: Logger, notification_handler: NotificationHandler): """ @@ -49,8 +45,6 @@ class Activities(Postgres, MLFlow, Gates, OPC): Required keys: host, port, user, password, dbname, min_connections, max_connections mlflow_config: MLFlow server configuration dictionary Required keys: host, port, username, password - opc_config: OPC server configuration dictionary - Can contain multiple server configurations logger: Logger instance for observability and debugging notification_handler: Notification handler for alerts and monitoring @@ -78,22 +72,15 @@ class Activities(Postgres, MLFlow, Gates, OPC): Gates.__init__(self, logger=logger, notification_handler=notification_handler) - OPC.__init__(self, - opc_servers=opc_config, - logger=logger, - notification_handler=notification_handler) - async def shutdown(self): """ Gracefully shutdown all activities and clean up resources. This method ensures proper cleanup of all resources including: - PostgreSQL connection pools - - OPC server connections - Any other resources that need explicit cleanup The method should be called before the application terminates to ensure proper resource cleanup and prevent resource leaks. """ Postgres.close(self) - await OPC.shutdown(self) diff --git a/model_manager/activities/opc.py b/model_manager/activities/opc.py deleted file mode 100644 index 2222eed..0000000 --- a/model_manager/activities/opc.py +++ /dev/null @@ -1,356 +0,0 @@ -from temporalio import activity, workflow - - -with workflow.unsafe.imports_passed_through(): - from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler - from sientia_do.notifications.models import NotificationLevel - from sientia_do.temporal.activities.base import BaseActivity - from sientia_do.observability.logger import Logger - from model_manager.utils.repository.opc_repository import OpcRepository - from typing import Any - import traceback - from pandas import DataFrame - -OPC_WRITTING_ERROR_CONFIDENCE = 12 - - -class OPC(BaseActivity): - """ - OPC server integration activities for real-time data export. - - This class provides comprehensive OPC UA client functionality for connecting - to multiple OPC servers and writing prediction data in real-time. It implements - secure communication with certificate-based authentication and automatic - reconnection capabilities. - - The class supports multiple OPC servers with individual configurations and - provides robust error handling and monitoring for production environments. - - Attributes: - opc_servers (dict): Configuration for multiple OPC servers - opc_repository (dict): Active OPC repository connections - logger (Logger): Logging and observability instance - notification_handler (NotificationHandler): Notification management instance - """ - - def __init__(self, opc_servers: dict[str, dict[str, Any]], - logger: Logger, notification_handler: NotificationHandler): - - self.logger = logger - self.notification_handler = notification_handler - self.opc_servers = opc_servers - - BaseActivity.__init__( - self, logger, notification_handler, set_error_counter=True) - - self.opc_repository: dict[str, OpcRepository] = {} - self.opc_servers = opc_servers - - async def init_opc(self): - """ - Initialize OPC server connections and establish communication channels. - - This method iterates through all configured OPC servers and attempts to - establish secure connections using certificate-based authentication. - Each server connection is managed independently, and connection failures - are reported through the notification system. - - The method performs the following operations: - 1. Creates OpcRepository instances for each configured server - 2. Establishes secure connections with certificate validation - 3. Reports connection success/failure through notifications - 4. Logs connection status for operational visibility - - Raises: - Exception: If OPC repository initialization fails or connection - establishment encounters critical errors - - Note: - Connection failures are logged and reported but do not prevent - the initialization of other OPC servers. Each server is handled - independently to ensure maximum availability. - """ - self.logger.info("Initializing OPC servers...") - for id, server in self.opc_servers.items(): - self.opc_repository[id] = OpcRepository( - id=server['id'], - url=server['url'], - logger=self.logger, - server_uri=server['server_uri'], - cert_path=server['cert_path'], - private_key_path=server['private_key_path'], - server_cert_path=server['server_cert_path'], - notification_handler=self.notification_handler, - reconnection_interval=server['reconnection_interval'], - pod_id=self.pod_id - ) - is_connected, error_data = await self.opc_repository[id].connect() - if not is_connected: - self.send_notification( - metadata={ - 'model_id': '-', - 'model_name': '-', - 'workflow_name': '-', - 'schedule_name': 'INITIALIZATION' - }, - notification_id=error_data['notification_id'], - message=error_data['message'], - block=error_data['block'], - level=error_data.get('level', NotificationLevel.ERROR), - attachment_content=error_data.get( - 'attachment_content', None) - ) - else: - self.logger.info( - f"OPC server {id} connected successfully.") - - async def write_data(self, server_id: str, tag: str, data: Any, - data_type: str, tag_type: str, metadata: dict[str, Any]) -> bool: - """ - Write data to a specific OPC server tag with comprehensive error handling. - - This method provides a secure and reliable way to write data to OPC servers - with automatic error handling, notification integration, and detailed logging. - It validates server availability before attempting write operations and - provides comprehensive error reporting for operational monitoring. - - Args: - - server_id (str): The id of the OPC server. - - tag (str): The tag to write to. - - data (Any): The data to write. - - data_type (str): The data type. - - tag_type (str): The tag type. - - Returns: - - bool: True if the data was written successfully, False otherwise. - """ - - try: - is_success, error_data = await self.opc_repository[server_id].write_data( - tag, data, data_type, self.logger, metadata) - if not is_success: - self.send_notification( - metadata=metadata, - notification_id=error_data['notification_id'], - message=error_data['message'], - block=error_data['block'], - level=error_data.get('level', NotificationLevel.ERROR), - attachment_content=error_data.get( - 'attachment_content', None) - ) - return False - return True - except Exception as e: - trace = traceback.format_exc() - self.send_notification( - metadata=metadata, - notification_id=f"WRITE_OPC_{tag_type.upper()}_ERROR", - message=f"Error writing data to OPC server: {e}", - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=trace - ) - raise e - - def validate_server(self, server_id: str, metadata: dict[str, Any]) -> bool: - """ - Validate that an OPC server is available and configured for write operations. - - This method checks if the specified OPC server exists in the active - repository and is available for data writing operations. It provides - immediate feedback for server availability and logs validation failures - for operational monitoring. - - Args: - server_id (str): Unique identifier for the OPC server to validate - metadata (dict[str, Any]): Context metadata for logging and notifications - - Returns: - bool: True if server is available, False otherwise - - Note: - Server validation failures are automatically reported through the - notification system with detailed information about available servers. - This helps operators quickly identify configuration issues. - """ - if self.opc_repository.get(server_id) is None: - message = f"OPC server {server_id} not found to perform write operation." - self.send_notification( - metadata=metadata, - notification_id="OPC_SERVER_NOT_FOUND", - message=message, - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=f"OPC servers: {list(self.opc_repository.keys())}" - ) - return False - return True - - async def manage_output_tags( - self, server_id: str, config: dict[str, Any], data: DataFrame, - metadata: dict[str, Any], success: bool) -> tuple[bool, int]: - """ - Manage the writing of prediction and confidence data to OPC server tags. - - This method orchestrates the writing of multiple data types to OPC servers - based on configuration. It handles both prediction data and confidence - values independently, allowing for flexible tag configuration and - comprehensive error handling. - - The method supports two main tag types: - 1. Prediction tags: Write actual prediction values to configured OPC tags - 2. Confidence tags: Write confidence scores to separate OPC tags - - Args: - server_id (str): Unique identifier for the target OPC server - config (dict[str, Any]): OPC tag configuration containing: - - prediction_tags (dict, optional): Prediction tag configurations - - confidence_tags (dict, optional): Confidence tag configurations - data (DataFrame): DataFrame containing prediction and confidence data - metadata (dict[str, Any]): Context metadata for logging and notifications - success (bool): Current success status to maintain across operations - - Returns: - tuple[bool, int]: (overall_success, total_tags_written) - - overall_success: True if all configured tags were written successfully - - total_tags_written: Count of successfully written tags - """ - - count = 0 - if 'prediction_tags' in config: - for tag, tag_config in config['prediction_tags'].items(): - local_success = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction'].values[0], - data_type=tag_config['data_type'], - tag_type='prediction', - metadata=metadata - ) - if local_success: - self.info( - f"Prediction data written to OPC server {server_id} for tag {tag}.", metadata) - count += 1 - success = success and local_success - - if 'confidence_tags' in config: - for tag, tag_config in config['confidence_tags'].items(): - local_success = await self.write_data( - server_id=server_id, - tag=tag, - data=data.head(1)['prediction_confidence'].values[0], - data_type=tag_config['data_type'], - tag_type='confidence', - metadata=metadata - ) - if local_success: - self.info( - f"Confidence data written to OPC server {server_id} for tag {tag}.", metadata) - count += 1 - success = success and local_success - - return success, count - - @activity.defn(name='write_opc_data') - async def write_opc_data(self, input_data: dict[str, Any]) -> dict[Any, Any]: - """ - Write prediction and confidence data to OPC servers. The two writing - operations are optional and independent of each other. - - Args: - - input_data(dict[str, Any]): The input data. Contains the following keys: - - data(dict[str, Any]): The dataframe that contains the data to write - to the OPC servers. - - opc_output_config(dict[str, Any]): The OPC writing configuration. - The keys are the OPC server names and the values contain: - - prediction_tags(dict[str, Any]): The tags to write to the OPC servers. - - confidence_tags(dict[str, Any]): The tags to write to the OPC servers. - - Returns: - - dict[Any, Any]: The data that was written to the OPC servers. - - """ - metadata = input_data['metadata'] - self.info("Writing data to OPC servers...", metadata) - data = DataFrame(input_data['data']) - opc_output_config = input_data['opc_output_config'] - self.info(f"Data to write: {data.size} rows", metadata) - - success = True - - for server_id, config in opc_output_config.items(): - - if not self.validate_server(server_id, metadata): - success = False - continue - - local_success, local_count = await self.manage_output_tags( - server_id, config, data, metadata, success) - success = success and local_success - - self.info( - f"Process completed for OPC server {server_id}: {local_count} of {len(config['prediction_tags'])} prediction tags and {len(config['confidence_tags'])} confidence tags", metadata) - - return self.process_confidence(data, success, metadata) - - def process_confidence(self, data: DataFrame, success: bool, metadata: dict[str, Any]) -> dict[Any, Any]: - """ - Process prediction confidence based on OPC write operation success. - - This method updates the prediction confidence values in the DataFrame - based on the success status of OPC server write operations. If any - write operations failed, it sets the confidence to a predefined error - value to indicate data quality issues. - - The method implements a confidence degradation strategy: - - Success: Maintains original confidence values - - Failure: Sets confidence to error value for operational awareness - - Args: - data (DataFrame): DataFrame containing prediction and confidence data - success (bool): Overall success status of OPC write operations - metadata (dict[str, Any]): Context metadata for logging and notifications - - Returns: - dict[Any, Any]: Processed data as a dictionary with updated confidence values - - Note: - The error confidence value (OPC_WRITTING_ERROR_CONFIDENCE = 12) is - used to indicate that data was not successfully exported to OPC servers. - This allows downstream systems to handle data quality appropriately. - """ - - if not success: - data['prediction_confidence'] = OPC_WRITTING_ERROR_CONFIDENCE - self.debug( - f"Some data could not be written to OPC servers, setting confidence to {OPC_WRITTING_ERROR_CONFIDENCE}.", - metadata - ) - - else: - self.debug("Data written to OPC servers successfully.", metadata) - - return data.to_dict() - - async def shutdown(self): - """ - Gracefully shutdown all OPC server connections and cleanup resources. - - This method ensures proper cleanup of all active OPC server connections - by calling the disconnect method on each repository instance. It's - designed to be called during application shutdown to prevent resource - leaks and ensure clean termination. - - The method performs the following cleanup operations: - 1. Iterates through all active OPC repository connections - 2. Calls disconnect() on each repository instance - 3. Allows for graceful connection termination - 4. Prevents resource leaks and connection hanging - - Note: - This method should be called during application shutdown to ensure - proper cleanup. It handles all active connections regardless of - their current state and provides a clean shutdown experience. - """ - for opc in self.opc_repository.values(): - await opc.disconnect() diff --git a/model_manager/metrics.py b/model_manager/metrics.py index 96b806f..a19b03c 100644 --- a/model_manager/metrics.py +++ b/model_manager/metrics.py @@ -13,14 +13,13 @@ Key Metric Categories: - Application Health: Overall system status and availability - Prediction Operations: Count and performance of prediction operations - Data Quality: Confidence levels and validation results -- Export Operations: Database and OPC export performance +- Export Operations: Database export performance - Response Times: Performance monitoring for various operations Metric Labels: - pod_id: Kubernetes pod identifier for multi-instance deployments - model_name: Name of the ML model being used - pipeline_name: Name of the prediction pipeline -- opc_server_id: Identifier for OPC server operations """ from prometheus_client import Gauge, Counter, Histogram @@ -56,17 +55,3 @@ PREDICTION_RESPONSE_TIME_MONITOR = Histogram( CORE_LABELS, buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] ) - -# OPC export metrics -PREDICTION_OPC_WRITING_COUNT = Counter( - "model_manager_prediction_opc_writing_count", - "Number of predictions written to the OPC server", - [*CORE_LABELS, "opc_server_id"], -) - -PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR = Histogram( - "model_manager_prediction_opc_writing_response_time_monitor", - "Current response time of each prediction written to the OPC server", - [*CORE_LABELS, "opc_server_id"], - buckets=[0.01, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0] -) diff --git a/model_manager/utils/connectors_config.py b/model_manager/utils/connectors_config.py index 145f248..40e7c07 100644 --- a/model_manager/utils/connectors_config.py +++ b/model_manager/utils/connectors_config.py @@ -59,45 +59,6 @@ def build_mlflow_config() -> Dict[str, Any]: } -def build_opc_config() -> Dict[str, Any]: - """ - Build OPC server configuration from environment variables. - - This function constructs an OPC server configuration dictionary from - environment variables. It supports both single server and multi-server - configurations with flexible parameter handling. - - Environment Variables: - OPC_CONFIG: JSON string containing multiple OPC server configurations - OPC_ID: OPC server ID (fallback, default: 1) - OPC_URL: Single OPC server URL (fallback, default: opc.tcp://localhost:4840) - OPC_SERVER_URI: Single OPC server URI (fallback, default: opc.tcp://localhost:4840) - OPC_CERT_PATH: Client certificate path (fallback, default: None) - OPC_PRIVATE_KEY_PATH: Client private key path (fallback, default: None) - OPC_SERVER_CERT_PATH: Server certificate path (fallback, default: None) - OPC_RECONNECTION_INTERVAL: Reconnection interval in milliseconds (fallback, default: 120) - - Returns: - dict: OPC server configuration dictionary - """ - opc_raw = getenv('OPC_CONFIG', None) - - if opc_raw: - return json.loads(opc_raw) - - return { - getenv('OPC_ID', '1'): { - 'id': getenv('OPC_ID', '1'), - 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), - 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), - 'cert_path': getenv('OPC_CERT_PATH', None), - 'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None), - 'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None), - 'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')) - } - } - - def build_mongodb_config() -> Dict[str, Any]: """ Build MongoDB configuration from environment variables. diff --git a/model_manager/utils/repository/opc_repository.py b/model_manager/utils/repository/opc_repository.py deleted file mode 100644 index 94b6c46..0000000 --- a/model_manager/utils/repository/opc_repository.py +++ /dev/null @@ -1,359 +0,0 @@ -import asyncio -import traceback -import time -from datetime import datetime -from pathlib import Path -from typing import Any -from asyncua import Client -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from asyncua.ua import DataValue, Variant, VariantType, DateTime -from regex import F -from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler -from sientia_do.notifications.models import NotificationLevel -from sientia_do.observability.logger import Logger -from model_manager import metrics - -data_type_map = { - 'float': { - 'converter': float, - 'opc_type': VariantType.Float, - }, - 'double': { - 'converter': float, - 'opc_type': VariantType.Double, - }, - 'int': { - 'converter': int, - 'opc_type': VariantType.Int32, - }, - 'bool': { - 'converter': bool, - 'opc_type': VariantType.Boolean, - }, - 'str': { - 'converter': str, - 'opc_type': VariantType.String, - } -} - - -class OpcRepository(): - def __init__(self, id: str, url: str, logger: Logger, - notification_handler: NotificationHandler, - reconnection_interval: int = 60, server_uri: str = None, cert_path: str = None, - private_key_path: str = None, server_cert_path: str = None, pod_id: str = None): - self.url = url - self.id = id - self.server_uri = server_uri - self.cert_path = cert_path - self.private_key_path = private_key_path - self.server_cert_path = server_cert_path - self.logger = logger - self.error_count = 0 - self.reconnection_interval = reconnection_interval - self.last_reconnection_time = None - self.notification_handler = notification_handler - self.client = None - self.pod_id = pod_id - - self.metadata = { - 'model_name': '-', - 'model_id': '-', - 'workflow_name': 'opc_repository', - 'schedule_name': '-' - } - - async def set_security(self): - """ - Configures the security settings for the OPC UA client. - This method sets up the security policy, certificates, and timeouts - required for establishing a secure connection with the OPC UA server. - Raises: - ValueError: If either the certificate path or private key path is not provided. - Attributes: - - cert_path (str): Path to the client's certificate file. - - private_key_path (str): Path to the client's private key file. - - server_cert_path (str, optional): Path to the server's certificate file. - - server_uri (str): The URI of the server to be used as the application URI. - - client (opcua.Client): The OPC UA client instance. - - logger (logging.Logger): Logger instance for logging information. - Security Settings: - - Security Policy: Basic256 - - Secure Channel Timeout: 10,000,000 ms - - Session Timeout: 10,000,000 ms - """ - - if not all([self.cert_path, self.private_key_path]): - raise ValueError( - "Certificate and private key paths must be provided for secure connection.") - cert = Path(self.cert_path) - private_key = Path(self.private_key_path) - server_cert = Path( - self.server_cert_path) if self.server_cert_path else None - - self.client.application_uri = self.server_uri - self.logger.custom_info('Setting security...', self.metadata) - await self.client.set_security( - SecurityPolicyBasic256, - certificate=str(cert), - private_key=str(private_key), - server_certificate=str(server_cert) - ) - self.client.secure_channel_timeout = 10000000 - self.client.session_timeout = 10000000 - - async def connect(self) -> tuple[bool, dict[str, Any]]: - """ - Establishes a connection to the OPC server. - This method initializes the OPC client using the provided URL and - sets up security if a certificate path is specified. It then - attempts to connect to the server and logs the connection status. - Raises: - Exception: If the connection to the OPC server fails. - """ - - self.client = Client(self.url) - if self.cert_path: - await self.set_security() - self.logger.custom_info( - f'Starting connection to OPC server {self.id}...', self.metadata) - return await self.try_connect() - - async def try_connect(self) -> tuple[bool, dict[str, Any]]: - """ - Attempt to establish connection to the OPC server. - - This method performs the actual connection attempt to the OPC server - and handles connection failures with comprehensive error reporting. - It updates reconnection timing and provides detailed error information - for operational monitoring and debugging. - - Returns: - tuple[bool, dict[str, Any]]: Connection result - - bool: True if connection successful, False otherwise - - dict: Error information if connection failed - """ - - try: - self.last_reconnection_time = datetime.now() - await self.client.connect() - return True, {} - except Exception as e: - trace = traceback.format_exc() - self.logger.custom_error(trace, self.metadata) - - return False, { - "notification_id": f"OPC_CONNECTION_ERROR_{self.id}", - "message": f"Failed to connect to OPC server: {e}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - async def disconnect(self): - """ - Gracefully disconnect from the OPC server. - - This method safely terminates the connection to the OPC server - and cleans up client resources. It handles disconnection errors - gracefully and ensures proper resource cleanup. - """ - if self.client is None: - return - try: - await self.client.disconnect() - self.logger.custom_info( - 'Disconnected from OPC server', self.metadata) - except Exception as e: - self.logger.custom_error( - f"Failed to disconnect from OPC server: {e}", self.metadata) - self.client = None - - async def validate_connection(self) -> tuple[bool, dict[str, Any]]: - """ - Validate and maintain OPC server connection health. - - This method performs comprehensive connection validation and - implements automatic reconnection logic for production reliability. - It handles various connection states and implements intelligent - reconnection strategies with error counting and timing controls. - - Connection Validation: - 1. Checks client existence and connection state - 2. Implements error counting with automatic disconnection - 3. Enforces reconnection timing windows - 4. Provides detailed error reporting and notifications - - Reconnection Strategy: - - Error Count Threshold: Disconnects after 5 consecutive errors - - Reconnection Window: Enforces minimum intervals between attempts - - Automatic Recovery: Attempts reconnection when conditions allow - - State Monitoring: Continuously monitors connection health - - Args: - None - - Returns: - tuple[bool, dict[str, Any]]: Connection validation result - - bool: True if connection is healthy, False otherwise - - dict: Error information if validation fails - """ - if self.client is None: - return await self.connect() - - if self.error_count > 5: - self.logger.custom_warning( - f"OPC server {self.id} will be disconnected due to multiple errors", self.metadata) - try: - await self.disconnect() - except Exception as e: - trace = traceback.format_exc() - self.logger.custom_error( - f"Failed to disconnect from OPC server: {e}", self.metadata) - self.logger.custom_error(trace, self.metadata) - self.logger.custom_info( - f"Attempting to reconnect to OPC server {self.id}...", self.metadata) - return await self.connect() - - # Check if client is connected using asyncua's connection state - try: - if self.client.uaclient.protocol is None or self.client.uaclient.protocol.state == "closed": - # OPC server is not connected - self.logger.custom_error( - f"OPC server {self.id} is not connected", self.metadata) - if self.last_reconnection_time is None or (datetime.now() - self.last_reconnection_time).total_seconds( - ) > self.reconnection_interval: - await self.disconnect() - self.logger.custom_info( - f"Trying to reconnect to OPC server {self.id}...", self.metadata) - return await self.connect() - - return False, { - "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{self.id}", - "message": f"OPC server {self.id} is not connected, waiting for next reconnection window...", - "block": "opc_repository", - "level": NotificationLevel.WARNING - } - return True, {} - except Exception as e: - trace = traceback.format_exc() - message = f"Failed to validate connection to OPC server: {e}" - self.logger.custom_error(message, self.metadata) - return False, { - "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{self.id}", - "message": message, - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - async def write_data(self, node: str, value: Any, data_type: str, - logger: Logger, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]: - """ - Write data to OPC server with comprehensive validation and monitoring. - - This method provides secure and reliable data writing to OPC servers - with automatic connection validation, data type conversion, and - comprehensive error handling. It implements performance monitoring - and metrics collection for operational visibility. - - Data Writing Process: - 1. Connection validation and automatic reconnection - 2. Node validation and error handling - 3. Data type conversion and validation - 4. OPC data writing with timestamp - 5. Performance metrics collection - 6. Error handling and notification - - Args: - node (str): OPC node identifier to write data to - value (Any): Data value to write to the OPC node - data_type (str): Data type for OPC conversion - logger (Logger): Logger instance for operation logging - metadata (dict[str, Any]): Context metadata for logging and metrics - - Returns: - tuple[bool, dict[str, Any]]: Write operation result - - bool: True if write successful, False otherwise - - dict: Error information if write failed - """ - - is_connected, error = await self.validate_connection() - - if not is_connected: - return False, error - - start_time = time.time() - - try: - node_obj = self.client.get_node(node) - except Exception as e: - trace = traceback.format_exc() - logger.custom_error(trace, metadata.get('schedule_name', 'N/A')) - self.error_count += 1 - return False, { - "notification_id": f"OPC_WRITE_GET_NODE_ERROR_{self.id}", - "message": f"Failed to get node from OPC server: {e} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - - if data_type not in data_type_map: - return False, { - "notification_id": f"OPC_WRITE_DATA_TYPE_ERROR_{self.id}", - "message": f"Unsupported data type: {data_type} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR - } - - data = data_type_map[data_type]['converter'](value) - logger.custom_info( - f'Writing {data} - {type(data)} to {node}', metadata) - now = datetime.now() - ua_data = DataValue( - Variant(data, data_type_map[data_type]['opc_type']), - SourceTimestamp=DateTime( - now.year, - now.month, - now.day, - now.hour, - now.minute, - now.second, - now.microsecond - ) - ) - - try: - await node_obj.write_value(ua_data) - - metrics.PREDICTION_OPC_WRITING_COUNT.labels( - pod_id=self.pod_id, - model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], - opc_server_id=self.id - ).inc() - - end_time = time.time() - response_time = end_time - start_time - metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels( - pod_id=self.pod_id, - model_name=metadata['model_name'], - pipeline_name=metadata['workflow_name'], - opc_server_id=self.id - ).observe(response_time) - - except Exception as e: - trace = traceback.format_exc() - logger.custom_error(trace, metadata) - self.error_count += 1 - return False, { - "notification_id": f"OPC_WRITE_DATA_ERROR_{self.id}", - "message": f"Failed to write data to OPC server: {e} | metadata: {metadata}", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": trace - } - self.error_count = 0 - - return True, {} diff --git a/model_manager/worker/worker.py b/model_manager/worker/worker.py index 5d1d0b1..92631ab 100644 --- a/model_manager/worker/worker.py +++ b/model_manager/worker/worker.py @@ -42,7 +42,6 @@ with workflow.unsafe.imports_passed_through(): from model_manager.utils.connectors_config import ( build_postgres_config, build_mlflow_config, - build_opc_config, build_mongodb_config ) from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler @@ -63,9 +62,8 @@ async def main(): 2. Starts Prometheus metrics server 3. Initializes notification handler 4. Creates and configures activities - 5. Initializes OPC connections - 6. Starts Temporal client and workers - 7. Manages worker lifecycle and graceful shutdown + 5. Starts Temporal client and workers + 6. Manages worker lifecycle and graceful shutdown The function runs indefinitely until interrupted or an error occurs. On error, it performs cleanup and exits with a non-zero status code. @@ -105,14 +103,10 @@ async def main(): activities = Activities( postgres_config=build_postgres_config(), mlflow_config=build_mlflow_config(), - opc_config=build_opc_config(), logger=logger, notification_handler=notification_handler ) - logger.custom_info('Initializing OPC...', metadata) - await activities.init_opc() - logger.custom_info( f'Starting SDK Metrics Server on port {SDK_METRICS_PORT}...', metadata) @@ -167,8 +161,6 @@ async def main(): activities.format_prediction, activities.format_default_prediction, activities.get_last_timestamp, - # OPC - activities.write_opc_data, # Postgres activities.load_custom_query, activities.repeat_last_prediction, diff --git a/model_manager/workflows/predictions_batch.py b/model_manager/workflows/predictions_batch.py index 281b187..dec0fb7 100644 --- a/model_manager/workflows/predictions_batch.py +++ b/model_manager/workflows/predictions_batch.py @@ -58,7 +58,7 @@ class PredictionsBatch(): - mlflow_predict_filters (dict, optional): MLFlow prediction filters - model_retention (int, optional): Model retention period in minutes - path_priority (list[str]): Decision path priority configuration - - opc_output_config (dict, optional): OPC server export configuration + - datetime_columns (list[str], optional): Columns to treat as datetime Returns: @@ -115,7 +115,7 @@ class PredictionsBatch(): }), 'model_config': input_data.get('model_config', {}), 'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']), - 'opc_output_config': input_data.get('opc_output_config', {}), + 'prediction_store_policy': input_data.get( 'prediction_store_policy', 'lts:1') } diff --git a/model_manager/workflows/sub_workflows/format_and_export_prediction.py b/model_manager/workflows/sub_workflows/format_and_export_prediction.py index ef195c4..60fad76 100644 --- a/model_manager/workflows/sub_workflows/format_and_export_prediction.py +++ b/model_manager/workflows/sub_workflows/format_and_export_prediction.py @@ -14,9 +14,9 @@ class FormatAndExportPrediction(): Data formatting and export workflow for prediction results. This workflow handles the final stages of the prediction pipeline, including - data formatting, database persistence, OPC server export, and metrics recording. + data formatting, database persistence, and metrics recording. It implements flexible formatting based on prediction quality and provides - comprehensive export capabilities to multiple destinations. + comprehensive export capabilities. The workflow supports two main prediction paths: 1. Normal Prediction: Formats and exports successful prediction results @@ -24,7 +24,6 @@ class FormatAndExportPrediction(): Export Destinations: - PostgreSQL Database: Persistent storage with timestamp conversion - - OPC Servers: Real-time industrial system integration - Prometheus Metrics: Performance monitoring and operational visibility """ @@ -36,14 +35,12 @@ class FormatAndExportPrediction(): This method orchestrates the complete data export process by: 1. Determining the appropriate formatting strategy based on path_flag 2. Formatting prediction data according to quality and requirements - 3. Exporting data to OPC servers for real-time industrial access - 4. Persisting data to PostgreSQL database with comprehensive metadata - 5. Recording performance metrics for operational monitoring + 3. Persisting data to PostgreSQL database with comprehensive metadata + 4. Recording performance metrics for operational monitoring The method implements flexible formatting strategies: - Normal predictions: Full data formatting with confidence scores - Error predictions: Default formatting with error indicators - - Comprehensive export: Multi-destination data distribution Args: input_data: Complete configuration for the export workflow @@ -58,7 +55,6 @@ class FormatAndExportPrediction(): - comment (str): Operational comment or error description - schema (str): Database schema for data storage - table_name (str): Target table for data persistence - - opc_output_config (dict[str, Any]): OPC server export configuration - prediction_store_policy (str, optional): Data retention policy Returns: @@ -100,18 +96,6 @@ class FormatAndExportPrediction(): start_to_close_timeout=timedelta(seconds=60) ) - # write to opc - prediction = await workflow.execute_activity_method( - Activities.write_opc_data, - { - **metadata, - 'opc_output_config': input_data['opc_output_config'], - 'data': prediction - }, - retry_policy=retry_policy, - start_to_close_timeout=timedelta(seconds=60) - ) - # write to postgres await workflow.execute_activity_method( Activities.export_data_to_postgres, diff --git a/model_manager/workflows/sub_workflows/prediction_process.py b/model_manager/workflows/sub_workflows/prediction_process.py index e613ab5..c5407ec 100644 --- a/model_manager/workflows/sub_workflows/prediction_process.py +++ b/model_manager/workflows/sub_workflows/prediction_process.py @@ -64,7 +64,7 @@ class PredictionProcess(): - mlflow_predict_filters (dict): MLFlow prediction filters - model_retention (int): Model retention period in minutes - path_priority (list[str]): Decision path priority configuration - - opc_output_config (dict): OPC server export configuration + Returns: None: The workflow completes successfully when export workflow finishes @@ -210,7 +210,6 @@ class PredictionProcess(): 'model_id': model_id, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': input_data['opc_output_config'], 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'comment': comment, @@ -287,7 +286,6 @@ class PredictionProcess(): 'schema': schema, 'table_name': table_name, 'comment': comment, - 'opc_output_config': input_data['opc_output_config'], 'prediction_store_policy': input_data['prediction_store_policy'] } ) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..7b577c6 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,154 @@ +[build-system] +requires = ["setuptools>=61.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "model-manager" +version = "0.0.0" +description = "Sientia DataOps Model Manager - ML Model Orchestration System" +readme = "README.md" +requires-python = ">=3.11" +authors = [ + {name = "Aignosi", email = "dev@aignosi.com"} +] + +[tool.ruff] +line-length = 100 +target-version = "py311" +exclude = [ + ".git", + ".venv", + "venv", + "__pycache__", + "*.pyc", + ".pytest_cache", + "htmlcov", +] + +[tool.ruff.lint] +select = [ + "E", # pycodestyle errors + "W", # pycodestyle warnings + "F", # pyflakes + "I", # isort + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "UP", # pyupgrade + "N", # pep8-naming + "YTT", # flake8-2020 + "S", # flake8-bandit + "BLE", # flake8-blind-except + "A", # flake8-builtins + "C90", # mccabe complexity +] + +ignore = [ + "E501", # line too long (handled by formatter) + "S101", # use of assert (needed for tests) + "S105", # possible hardcoded password (false positives) + "S106", # possible hardcoded password (false positives) + "N802", # function name should be lowercase (temporal decorators) + "N806", # variable in function should be lowercase +] + +[tool.ruff.lint.per-file-ignores] +"tests/**/*.py" = [ + "S101", # assert allowed in tests + "S105", # hardcoded passwords ok in tests + "S106", # hardcoded passwords ok in tests +] + +[tool.ruff.lint.mccabe] +max-complexity = 15 + +[tool.ruff.format] +quote-style = "single" +indent-style = "space" +line-ending = "auto" + +[tool.mypy] +python_version = "3.11" +warn_return_any = true +warn_unused_configs = true +disallow_untyped_defs = false +disallow_incomplete_defs = false +check_untyped_defs = true +no_implicit_optional = true +warn_redundant_casts = true +warn_unused_ignores = true +warn_no_return = true +strict_equality = true +ignore_missing_imports = false + +# Ignore missing imports for external packages +[[tool.mypy.overrides]] +module = "temporalio.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "sientia_do.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "mlflow.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "prometheus_client.*" +ignore_missing_imports = true + +[[tool.mypy.overrides]] +module = "redis.*" +ignore_missing_imports = true + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +addopts = [ + "-v", + "--strict-markers", + "--cov=model_manager", + "--cov-report=term-missing", + "--cov-report=html", + "--cov-report=xml", +] +markers = [ + "asyncio: marks tests as async", + "integration: marks tests as integration tests", + "unit: marks tests as unit tests", +] + +[tool.coverage.run] +source = ["model_manager"] +omit = [ + "*/tests/*", + "*/venv/*", + "*/__pycache__/*", + "*/site-packages/*", +] +branch = true + +[tool.coverage.report] +precision = 2 +show_missing = true +skip_covered = false +exclude_lines = [ + "pragma: no cover", + "def __repr__", + "def __str__", + "raise AssertionError", + "raise NotImplementedError", + "if __name__ == .__main__.:", + "if TYPE_CHECKING:", + "class .*\\bProtocol\\):", + "@(abc\\.)?abstractmethod", +] + +[tool.coverage.html] +directory = "htmlcov" + +[tool.bandit] +exclude_dirs = ["tests", "venv", ".venv"] +skips = ["B101", "B601"] # Skip assert and shell injection in controlled environments diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..e65402b --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,17 @@ +# Development and Testing Dependencies +# These packages are only needed for development, testing, and code quality checks +# Install with: pip install -r requirements-dev.txt + +# Code Quality & Linting +ruff>=0.1.0 # Fast Python linter and formatter (replaces flake8, black, isort) +mypy>=1.7.0 # Static type checker +bandit>=1.7.5 # Security vulnerability scanner + +# Testing +pytest>=7.4.0 # Testing framework +pytest-cov>=4.1.0 # Coverage plugin for pytest +pytest-asyncio>=0.21.0 # Async test support (already in main requirements) + +# Development Tools +ipython>=8.12.0 # Enhanced Python shell +ipdb>=0.13.13 # IPython debugger diff --git a/requirements.txt b/requirements.txt index 1ce9f9c..8fcde0c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,6 @@ temporalio psycopg2-binary sqlalchemy -asyncua redis 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 diff --git a/tests/laborious/activities/test_activities.py b/tests/laborious/activities/test_activities.py index 106ccf7..b6e53df 100644 --- a/tests/laborious/activities/test_activities.py +++ b/tests/laborious/activities/test_activities.py @@ -4,14 +4,12 @@ from sientia_do.temporal.activities.postgres import Postgres from model_manager.activities.activities import Activities from model_manager.activities.mlflow import MLFlow from model_manager.activities.gates import Gates -from model_manager.activities.opc import OPC @patch('model_manager.activities.activities.Postgres.__init__') @patch('model_manager.activities.activities.MLFlow.__init__') -@patch('model_manager.activities.activities.OPC.__init__') @patch('model_manager.activities.activities.Gates.__init__') -def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgres_init): +def test___init__(mock_gates_init, mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', @@ -30,19 +28,12 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre 'password': 'mlflow' } - opc_config = { - 'bootstrap_servers': 'localhost:9092', - 'polling_time': 1000, - 'group_id': 'test-group' - } - logger = MagicMock() notification_handler = MagicMock() activities = Activities( postgres_config=postgres_config, mlflow_config=mlflow_config, - opc_config=opc_config, logger=logger, notification_handler=notification_handler ) @@ -50,7 +41,6 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre assert isinstance(activities, Activities) assert isinstance(activities, Postgres) assert isinstance(activities, MLFlow) - assert isinstance(activities, OPC) assert isinstance(activities, Gates) mock_postgres_init.assert_called_once_with( @@ -76,13 +66,6 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre notification_handler=notification_handler ) - mock_opc_init.assert_called_once_with( - ANY, - opc_servers=opc_config, - logger=logger, - notification_handler=notification_handler - ) - mock_gates_init.assert_called_once_with( ANY, logger=logger, @@ -93,9 +76,7 @@ def test___init__(mock_gates_init, mock_opc_init, mock_mlflow_init, mock_postgre @mark.asyncio @patch('model_manager.activities.activities.Postgres', return_value=MagicMock()) @patch('model_manager.activities.activities.MLFlow', return_value=MagicMock()) -@patch('model_manager.activities.activities.OPC', return_value=MagicMock()) -async def test_shutdown(mock_opc_init, - _mock_mlflow_init, mock_postgres_init): +async def test_shutdown(_mock_mlflow_init, mock_postgres_init): postgres_config = { 'host': 'localhost', 'port': 5432, @@ -113,23 +94,15 @@ async def test_shutdown(mock_opc_init, 'password': 'mlflow' } - opc_config = { - 'bootstrap_servers': 'localhost:9092', - 'polling_time': 1000, - 'group_id': 'test-group' - } - logger = MagicMock() notification_handler = MagicMock() activities = Activities( postgres_config=postgres_config, mlflow_config=mlflow_config, - opc_config=opc_config, logger=logger, notification_handler=notification_handler ) await activities.shutdown() - mock_opc_init.shutdown.assert_called_once() mock_postgres_init.close.assert_called_once() diff --git a/tests/laborious/activities/test_opc.py b/tests/laborious/activities/test_opc.py deleted file mode 100644 index dcff019..0000000 --- a/tests/laborious/activities/test_opc.py +++ /dev/null @@ -1,369 +0,0 @@ -from unittest.mock import patch, MagicMock, ANY, call, AsyncMock -from pandas import DataFrame -from pytest import fixture, mark -import pytest_asyncio -from sientia_do.notifications.models import NotificationLevel - -from model_manager.activities.opc import OPC - -metadata = { - "metadata": { - "model_id": "test_model", - "model_name": "test_model", - "workflow_name": "test_workflow", - "schema_name": "test_schedule", - }, -} - - -def test__init__(): - servers = { - 'server1': 'config' - } - opc = OPC( - opc_servers=servers, - logger=MagicMock(), - notification_handler=MagicMock() - ) - - assert opc.opc_servers == servers - assert opc.opc_repository == {} - - -@mark.asyncio -@patch("model_manager.activities.opc.OpcRepository") -@patch("model_manager.activities.opc.OPC.send_notification") -async def test_init_opc(mock_send_notification, mock_opc_repository): - mock_logger = MagicMock() - server1 = MagicMock( - connect=AsyncMock(return_value=(True, {})), - write_data=AsyncMock(return_value=(True, {})) - ) - server2 = MagicMock( - connect=AsyncMock(return_value=(True, {})), - write_data=AsyncMock(return_value=(True, {})) - ) - server3 = MagicMock( - connect=AsyncMock(return_value=(False, { - 'notification_id': 'OPC_CONNECTION_ERROR_server3', - 'message': 'Failed to connect to OPC server: Test error', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': 'Test error' - })), - write_data=AsyncMock(return_value=(True, {})) - ) - mock_opc_repository.side_effect = [server1, server2, server3] - mock_notification_handler = MagicMock() - servers = { - 'server1': { - 'id': 'server1', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - }, - 'server2': { - 'id': 'server2', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - }, - 'server3': { - 'id': 'server3', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - } - } - opc = OPC( - opc_servers=servers, - logger=mock_logger, - notification_handler=mock_notification_handler - ) - await opc.init_opc() - - assert opc.opc_servers == servers - assert opc.logger == mock_logger - assert opc.notification_handler == mock_notification_handler - assert opc.opc_repository['server1'] == server1 - assert opc.opc_repository['server2'] == server2 - - mock_opc_repository.assert_has_calls([ - call( - id="server1", - url="http://localhost:8080", - logger=mock_logger, - server_uri="opc.tcp://localhost:4840", - cert_path="", - private_key_path="", - server_cert_path="", - notification_handler=mock_notification_handler, - reconnection_interval=60, - pod_id='localhost' - ), - ]) - mock_opc_repository.assert_has_calls([ - call( - id="server2", - url="http://localhost:8080", - logger=mock_logger, - server_uri="opc.tcp://localhost:4840", - cert_path="", - private_key_path="", - server_cert_path="", - notification_handler=mock_notification_handler, - reconnection_interval=60, - pod_id='localhost' - ) - ]) - - server1.connect.assert_called_once() - server2.connect.assert_called_once() - - mock_send_notification.assert_has_calls([ - call( - metadata={ - 'model_id': '-', - 'model_name': '-', - 'workflow_name': '-', - 'schedule_name': 'INITIALIZATION' - }, - notification_id="OPC_CONNECTION_ERROR_server3", - message="Failed to connect to OPC server: Test error", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - ]) - - -@pytest_asyncio.fixture -@patch("model_manager.activities.opc.OpcRepository") -async def opc(mock_opc_repository): - servers = { - 'server1': { - 'id': 'server1', - 'url': 'http://localhost:8080', - 'server_uri': 'opc.tcp://localhost:4840', - 'cert_path': '', - 'private_key_path': '', - 'server_cert_path': '', - 'reconnection_interval': 60, - } - } - - mock_opc_repository.return_value.write_data = AsyncMock( - return_value=(True, {}) - ) - mock_opc_repository.return_value.connect = AsyncMock( - return_value=(True, {}) - ) - opc = OPC( - opc_servers=servers, - logger=MagicMock(), - notification_handler=MagicMock() - ) - await opc.init_opc() - opc.send_notification = MagicMock() - return opc - - -WRITE_DATA_CASES = [ - ('tag1', 'int', 50), - ('tag2', 'float', 50.5), - ('tag3', 'bool', True), - ('tag4', 'string', 'test'), -] - - -@mark.parametrize('tag,data_type,data', WRITE_DATA_CASES) -@mark.asyncio -async def test_write_data_success(opc, tag, data_type, data): - result = await opc.write_data(server_id='server1', tag=tag, data=data, - data_type=data_type, tag_type='prediction', metadata=metadata) - assert result is True - opc.opc_repository['server1'].write_data.assert_called_once_with( - tag, data, data_type, opc.logger, metadata) - - -@mark.asyncio -async def test_write_data_failed(opc): - opc.opc_repository['server1'].write_data.return_value = (False, { - 'notification_id': 'OPC_WRITE_DATA_ERROR_server1', - 'message': 'Failed to write data to OPC server: Test error', - 'block': 'opc_repository', - 'level': NotificationLevel.ERROR, - 'attachment_content': 'Test error' - }) - - result = await opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) - assert result is False - - opc.send_notification.assert_called_once_with( - metadata=metadata, - notification_id="OPC_WRITE_DATA_ERROR_server1", - message="Failed to write data to OPC server: Test error", - block="opc_repository", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - - -@mark.asyncio -async def test_write_data_exception(opc): - opc.opc_repository['server1'].write_data.side_effect = Exception( - "Test error") - - try: - await opc.write_data(server_id='server1', tag='tag1', data=50, - data_type='int', tag_type='prediction', metadata=metadata) - - except Exception: - opc.send_notification.assert_called_once_with( - metadata=metadata, - notification_id="WRITE_OPC_PREDICTION_ERROR", - message="Error writing data to OPC server: Test error", - block="write_opc_data", - level=NotificationLevel.ERROR, - attachment_content=ANY - ) - - else: - assert False, "Expected an exception to be raised" - - -@mark.asyncio -async def test_write_opc_data_success(opc): - # Arrange - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_output_config': { - 'server1': { - 'prediction_tags': { - 'tag1': {'data_type': 'float'} - }, - 'confidence_tags': { - 'tag2': {'data_type': 'float'} - } - } - } - } - - # Act - opc.write_data = AsyncMock(return_value=True) - opc.process_confidence = MagicMock(return_value={'data': 'data'}) - output = await opc.write_opc_data(input_data) - - # Assert - assert output == {'data': 'data'} - opc.write_data.assert_has_calls([ - call( - server_id='server1', - tag='tag1', - data=0.75, - data_type='float', - tag_type='prediction', - metadata=metadata['metadata'] - )]) - opc.write_data.assert_has_calls([ - call( - server_id='server1', - tag='tag2', - data=0.95, - data_type='float', - tag_type='confidence', - metadata=metadata['metadata'] - ) - ]) - assert opc.write_data.call_count == 2 - - -@mark.asyncio -async def test_write_opc_data_empty_config(opc): - # Arrange - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_servers': ['server1'], - 'opc_output_config': { - 'server1': { - 'prediction_tags': {}, - 'confidence_tags': {} - } - } - } - - # Act - await opc.write_opc_data(input_data) - - # Assert - opc.opc_repository['server1'].write_data.assert_not_called() - - -@mark.asyncio -async def test_write_opc_data_no_validate_server(opc): - opc.validate_server = MagicMock(return_value=False) - input_data = { - **metadata, - 'data': { - 'prediction': [0.75], - 'prediction_confidence': [0.95] - }, - 'opc_output_config': { - 'server1': { - 'prediction_tags': { - 'tag1': {'data_type': 'float'} - }, - 'confidence_tags': { - 'tag2': {'data_type': 'float'} - } - } - } - } - - # Act - await opc.write_opc_data(input_data) - - # Assert - opc.opc_repository['server1'].write_data.assert_not_called() - - -@mark.parametrize('data,success,expected', [ - (DataFrame({'prediction_confidence': [0]}), True, 0), - (DataFrame({'prediction_confidence': [0]}), False, 12), -]) -def test_process_confidence(opc, data, success, expected): - # Act - result = opc.process_confidence(data, success, metadata) - - # Assert - assert result['prediction_confidence'][0] == expected - - -def test_validate_server(opc): - assert opc.validate_server('server1', metadata) is True - assert opc.validate_server('server2', metadata) is False - - -@mark.asyncio -async def test_shutdown(opc): - opc.opc_repository['server1'].disconnect = AsyncMock(return_value=True) - await opc.shutdown() - opc.opc_repository['server1'].disconnect.assert_called_once() diff --git a/tests/laborious/utils/repository/test_opc_repository.py b/tests/laborious/utils/repository/test_opc_repository.py deleted file mode 100644 index 9795a10..0000000 --- a/tests/laborious/utils/repository/test_opc_repository.py +++ /dev/null @@ -1,388 +0,0 @@ -import pytest -from unittest.mock import AsyncMock, Mock, patch, MagicMock, ANY, call -from asyncua.crypto.security_policies import SecurityPolicyBasic256 -from model_manager.utils.repository.opc_repository import OpcRepository -from sientia_do.notifications.models import NotificationLevel -from datetime import datetime - - -@pytest.fixture -def mock_logger(): - return Mock() - - -@pytest.fixture -def opc_repository(mock_logger): - return OpcRepository( - id="test_repo", - url="opc.tcp://localhost:4840", - logger=mock_logger, - notification_handler=Mock(), - reconnection_interval=60, - server_uri="urn:test:server", - cert_path="/path/to/cert.pem", - private_key_path="/path/to/key.pem", - server_cert_path="/path/to/server_cert.pem" - ) - - -@pytest.fixture -def mock_client(): - with patch('model_manager.utils.repository.opc_repository.Client') as mock: - client_instance = AsyncMock() - mock.return_value = client_instance - yield client_instance - - -metadata = { - "metadata": { - "model_id": "test_model", - "model_name": "test_model", - "workflow_name": "test_workflow", - "schema_name": "test_schedule", - }, -} - - -def test_init(opc_repository): - assert opc_repository.id == "test_repo" - assert opc_repository.url == "opc.tcp://localhost:4840" - assert opc_repository.server_uri == "urn:test:server" - assert opc_repository.cert_path == "/path/to/cert.pem" - assert opc_repository.private_key_path == "/path/to/key.pem" - assert opc_repository.server_cert_path == "/path/to/server_cert.pem" - assert opc_repository.reconnection_interval == 60 - assert opc_repository.client is None - assert opc_repository.last_reconnection_time is None - assert opc_repository.error_count == 0 - - -@pytest.mark.asyncio -async def test_set_security(opc_repository, mock_client): - opc_repository.client = mock_client - await opc_repository.set_security() - - mock_client.application_uri = "urn:test:server" - mock_client.set_security.assert_called_once_with( - SecurityPolicyBasic256, - certificate="/path/to/cert.pem", - private_key="/path/to/key.pem", - server_certificate="/path/to/server_cert.pem" - ) - assert mock_client.secure_channel_timeout == 10000000 - assert mock_client.session_timeout == 10000000 - - -@pytest.mark.asyncio -async def test_set_security_missing_certificates(opc_repository): - opc_repository.cert_path = None - opc_repository.private_key_path = None - - try: - await opc_repository.set_security() - except ValueError as e: - assert str( - e) == "Certificate and private key paths must be provided for secure connection." - - -@pytest.mark.asyncio -async def test_connect_with_security(opc_repository, mock_client): - opc_repository.try_connect = AsyncMock(return_value=(True, {})) - result = await opc_repository.connect() - - opc_repository.try_connect.assert_called_once() - assert opc_repository.client == mock_client - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_connect_without_security(opc_repository, mock_client): - opc_repository.cert_path = None - opc_repository.try_connect = AsyncMock(return_value=(True, {})) - opc_repository.set_security = AsyncMock() - result = await opc_repository.connect() - - opc_repository.try_connect.assert_called_once() - opc_repository.set_security.assert_not_called() - assert opc_repository.client == mock_client - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_try_connect_success(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.client = AsyncMock() - result = await opc_repository.try_connect() - - opc_repository.client.connect.assert_called_once() - assert opc_repository.last_reconnection_time is not None - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_try_connect_fail(opc_repository): - opc_repository.last_reconnection_time = None - opc_repository.client = MagicMock() - opc_repository.client.connect.side_effect = Exception("Test error") - - is_connected, error_data = await opc_repository.try_connect() - - opc_repository.client.connect.assert_called_once() - assert is_connected is False - assert error_data['notification_id'] == f"OPC_CONNECTION_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to connect to OPC server: Test error" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None - - -@pytest.mark.asyncio -async def test_disconnect(opc_repository, mock_client): - opc_repository.client = mock_client - await opc_repository.disconnect() - - mock_client.disconnect.assert_called_once() - assert opc_repository.client is None - - -@pytest.mark.asyncio -async def test_disconnect_no_client(opc_repository): - opc_repository.client = None - assert await opc_repository.disconnect() is None - - -@pytest.mark.asyncio -async def test_disconnect_error(opc_repository, mock_client): - opc_repository.client = mock_client - mock_client.disconnect.side_effect = Exception("Test error") - await opc_repository.disconnect() - - opc_repository.logger.custom_error.assert_called_once_with( - "Failed to disconnect from OPC server: Test error", - ANY - ) - assert opc_repository.client is None - - -@pytest.mark.asyncio -async def test_validate_connection_none_client(opc_repository): - opc_repository.client = None - opc_repository.connect = AsyncMock(return_value=(True, {})) - response = await opc_repository.validate_connection() - assert response == (True, {}) - opc_repository.connect.assert_called_once() - - -@pytest.mark.asyncio -async def test_validate_connection_error_count_disconnect_error(opc_repository): - opc_repository.error_count = 6 - opc_repository.client = AsyncMock() - opc_repository.disconnect = AsyncMock( - side_effect=Exception("Test error") - ) - opc_repository.connect = AsyncMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - assert response == opc_repository.connect.return_value - opc_repository.disconnect.assert_called_once() - opc_repository.connect.assert_called_once() - opc_repository.logger.custom_error.assert_has_calls( - [ - call("Failed to disconnect from OPC server: Test error", ANY), - ] - ) - - -@pytest.mark.asyncio -async def test_validate_connection_error_validate_connection_error(opc_repository): - opc_repository.client = MagicMock( - uaclient=Exception("Test error") - ) - opc_repository.error_count = 0 - - response = await opc_repository.validate_connection() - - assert response == (False, { - "notification_id": f"OPC_CONNECTION_CHECK_ERROR_{opc_repository.id}", - "message": "Failed to validate connection to OPC server: 'Exception' object has no attribute 'protocol'", - "block": "opc_repository", - "level": NotificationLevel.ERROR, - "attachment_content": ANY - }) - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_not_time_to_reconnect(_mock_datetime, opc_repository): - _mock_datetime.now = MagicMock( - return_value=datetime(2025, 1, 1, 0, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = MagicMock() - opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = MagicMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - opc_repository.connect.assert_not_called() - assert response == (False, { - "notification_id": f"OPC_CONNECTION_AWAITING_RECONNECTION_WINDOW_{opc_repository.id}", - "message": f"OPC server {opc_repository.id} is not connected, waiting for next reconnection window...", - "block": "opc_repository", - "level": NotificationLevel.WARNING - }) - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.datetime') -async def test_validate_connection_lost_time_to_reconnect(mock_datetime, opc_repository): - mock_datetime.now = MagicMock( - return_value=datetime(2025, 1, 1, 1, 0, 0)) - opc_repository.error_count = 0 - opc_repository.client = AsyncMock() - opc_repository.client.uaclient.protocol = None - opc_repository.last_reconnection_time = datetime(2025, 1, 1, 0, 0, 0) - opc_repository.connect = AsyncMock(return_value=(True, {})) - - response = await opc_repository.validate_connection() - opc_repository.connect.assert_called_once() - assert response == opc_repository.connect.return_value - - -@pytest.mark.asyncio -async def test_validate_connection_success(opc_repository): - opc_repository.client = MagicMock() - opc_repository.error_count = 0 - opc_repository.client.uaclient.protocol = MagicMock() - opc_repository.client.uaclient.protocol.state = "open" - - output = await opc_repository.validate_connection() - assert output == (True, {}) - - -@pytest.mark.asyncio -async def test_write_data_validate_connection_do_nothing(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = AsyncMock( - get_node=MagicMock() - ) - mock_node = AsyncMock() - opc_repository.client.get_node.return_value = mock_node - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") - assert result == (True, {}) - - -@pytest.mark.asyncio -async def test_write_data_validate_connection_failed(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(False, {})) - opc_repository.client = AsyncMock() - opc_repository.error_count = 0 - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_not_called() - assert result == (False, {}) - - -@pytest.mark.asyncio -async def test_write_data_get_node_failed(opc_repository): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = AsyncMock() - opc_repository.error_count = 0 - opc_repository.client.get_node = MagicMock( - side_effect=Exception("Test error")) - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - opc_repository.client.get_node.assert_called_once_with("ns=2;s=TestNode") - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_GET_NODE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to get node from OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None - - -@pytest.mark.asyncio -async def test_write_data_invalid_data_type(opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - mock_client.get_node = MagicMock(return_value=mock_node) - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "invalid_type", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_DATA_TYPE_ERROR_{opc_repository.id}" - assert error_data['message'] == "Unsupported data type: invalid_type | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data.get('attachment_content') is None - - -@pytest.mark.asyncio -@patch('model_manager.utils.repository.opc_repository.metrics') -async def test_write_data(mock_metrics, opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - mock_client.get_node = MagicMock(return_value=mock_node) - - result = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - mock_node.write_value.assert_called_once() - assert result == (True, {}) - - mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.assert_called_once_with( - pod_id=opc_repository.pod_id, - model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], - opc_server_id=opc_repository.id - ) - mock_metrics.PREDICTION_OPC_WRITING_COUNT.labels.return_value.inc.assert_called_once_with() - - mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.assert_called_once_with( - pod_id=opc_repository.pod_id, - model_name=metadata['metadata']['model_name'], - pipeline_name=metadata['metadata']['workflow_name'], - opc_server_id=opc_repository.id - ) - mock_metrics.PREDICTION_OPC_WRITING_RESPONSE_TIME_MONITOR.labels.return_value.observe.assert_called_once_with( - ANY) - - -@pytest.mark.asyncio -async def test_write_data_write_value_failed(opc_repository, mock_client): - opc_repository.validate_connection = AsyncMock(return_value=(True, {})) - opc_repository.client = mock_client - mock_node = AsyncMock() - opc_repository.error_count = 0 - mock_client.get_node = MagicMock(return_value=mock_node) - mock_node.write_value.side_effect = Exception("Test error") - - is_success, error_data = await opc_repository.write_data("ns=2;s=TestNode", 42.0, - "float", opc_repository.logger, metadata['metadata']) - - opc_repository.validate_connection.assert_called_once() - mock_client.get_node.assert_called_once_with("ns=2;s=TestNode") - mock_node.write_value.assert_called_once() - assert is_success is False - assert error_data['notification_id'] == f"OPC_WRITE_DATA_ERROR_{opc_repository.id}" - assert error_data['message'] == "Failed to write data to OPC server: Test error | metadata: {'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule'}" - assert error_data['block'] == "opc_repository" - assert error_data['level'] == NotificationLevel.ERROR - assert error_data['attachment_content'] is not None diff --git a/tests/laborious/utils/test_connectors_config.py b/tests/laborious/utils/test_connectors_config.py index 43a2ec9..6179b33 100644 --- a/tests/laborious/utils/test_connectors_config.py +++ b/tests/laborious/utils/test_connectors_config.py @@ -1,6 +1,5 @@ from os import environ from model_manager.utils.connectors_config import (build_mlflow_config, - build_opc_config, build_postgres_config, build_mongodb_config) @@ -40,54 +39,6 @@ def test_build_mlflow_config_with_defaults(): assert config['password'] == 'aignosi' -def test_build_opc_config_with_env_vars(): - # Arrange - environ['OPC_CONFIG'] = '{"opc": {"name": "test-opc", "url": "opc.tcp://test:4840"}}' - - # Act - config = build_opc_config() - - # Assert - assert config['opc']['name'] == 'test-opc' - assert config['opc']['url'] == 'opc.tcp://test:4840' - - -def test_build_opc_config_with_individual_env_vars(): - # Arrange - environ.pop('OPC_CONFIG', None) - environ['OPC_ID'] = '1' - environ['OPC_URL'] = 'opc.tcp://test:4840' - environ['OPC_SERVER_URI'] = 'opc.tcp://test:4840' - environ['OPC_RECONNECTION_INTERVAL'] = '300' - - # Act - config = build_opc_config() - - # Assert - assert config['1']['id'] == '1' - assert config['1']['url'] == 'opc.tcp://test:4840' - assert config['1']['server_uri'] == 'opc.tcp://test:4840' - assert config['1']['reconnection_interval'] == 300 - - -def test_build_opc_config_with_defaults(): - # Arrange - environ.pop('OPC_CONFIG', None) - environ.pop('OPC_ID', None) - environ.pop('OPC_URL', None) - environ.pop('OPC_SERVER_URI', None) - environ.pop('OPC_RECONNECTION_INTERVAL', None) - - # Act - config = build_opc_config() - - # Assert - assert config['1']['id'] == '1' - assert config['1']['url'] == 'opc.tcp://localhost:4840' - assert config['1']['server_uri'] == 'opc.tcp://localhost:4840' - assert config['1']['reconnection_interval'] == 120 - - def test_build_postgres_config_with_env_vars(): # Arrange environ['POSTGRES_HOST'] = 'test-host' diff --git a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py index 356c32b..3606d5d 100644 --- a/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py +++ b/tests/laborious/workflows/subworkflows/test_format_and_export_prediction.py @@ -34,8 +34,6 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): "prediction_confidence": 0, "schema": "test_schema", "table_name": "test_table", - "opc_servers": ["test_server"], - "opc_output_config": {"test": "config"}, "prediction_store_policy": "erl:1" } @@ -56,19 +54,6 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): start_to_close_timeout=ANY )]) - workflow_mock.execute_activity_method.assert_has_calls([ - call( - Activities.write_opc_data, - { - 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value, - **metadata - }, - retry_policy=ANY, - start_to_close_timeout=ANY - ) - ]) - workflow_mock.execute_activity_method.assert_has_calls([ call( Activities.export_data_to_postgres, @@ -86,7 +71,7 @@ async def test_run_none_path_flag(workflow_mock, format_and_export_prediction): start_to_close_timeout=ANY )]) - assert workflow_mock.execute_activity_method.call_count == 3 + assert workflow_mock.execute_activity_method.call_count == 2 assert workflow_mock.execute_local_activity_method.call_count == 1 @@ -103,8 +88,6 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction "prediction_confidence": 0, "schema": "test_schema", "table_name": "test_table", - "opc_servers": ["test_server"], - "opc_output_config": {"test": "config"}, "comment": "test_comment" } @@ -125,19 +108,6 @@ async def test_run_default_path_flag(workflow_mock, format_and_export_prediction ) ]) - workflow_mock.execute_activity_method.assert_has_calls([ - call( - Activities.write_opc_data, - { - 'opc_output_config': input_data['opc_output_config'], - 'data': workflow_mock.execute_local_activity_method.return_value, - **metadata - }, - retry_policy=ANY, - start_to_close_timeout=ANY - ) - ]) - workflow_mock.execute_activity_method.assert_has_calls([ call( Activities.export_data_to_postgres, diff --git a/tests/laborious/workflows/subworkflows/test_prediction_process.py b/tests/laborious/workflows/subworkflows/test_prediction_process.py index 46b1565..6922ef1 100644 --- a/tests/laborious/workflows/subworkflows/test_prediction_process.py +++ b/tests/laborious/workflows/subworkflows/test_prediction_process.py @@ -38,7 +38,7 @@ async def test_run(workflow_mock, prediction_process): 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': 'lts:1' } @@ -125,7 +125,7 @@ async def test_run(workflow_mock, prediction_process): 'model_id': 1, 'model_name': 'test_model_name', 'model_config': input_data['model_config'], - 'opc_output_config': input_data['opc_output_config'], + 'schema': input_data['schema'], 'table_name': input_data['table_name'], 'comment': 'Error', @@ -153,7 +153,7 @@ async def test_run_stop_at_input_gate(workflow_mock, prediction_process): 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -201,7 +201,7 @@ async def test_run_stop_at_first_mlflow_response_gate(workflow_mock, prediction_ 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -272,7 +272,7 @@ async def test_run_stop_at_mlflow_content_gate(workflow_mock, prediction_process 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -352,7 +352,7 @@ async def test_run_stop_at_mlflow_last_response_gate(workflow_mock, prediction_p 'retention': '30' }, 'path_priority': ['continue', 'repeat', 'stop'], - 'opc_output_config': {'test': 'config'} + } # Mock the activity responses @@ -536,7 +536,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'last_timestamp': last_timestamp, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, 'Prediction Process' ) @@ -558,7 +558,7 @@ async def test_path_flag_handler_continue(workflow_mock, prediction_process): 'schema': schema, 'table_name': table_name, 'comment': 'Prediction Process', - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy } ) @@ -590,7 +590,7 @@ async def test_path_flag_handler_unknown(workflow_mock, prediction_process): 'last_timestamp': last_timestamp, 'model_name': model_name, 'model_config': model_config, - 'opc_output_config': {'test': 'config'}, + 'prediction_store_policy': prediction_store_policy }, confidence, last_timestamp, "" ) diff --git a/tests/laborious/workflows/test_predictions_batch.py b/tests/laborious/workflows/test_predictions_batch.py index 0d7d477..e71a462 100644 --- a/tests/laborious/workflows/test_predictions_batch.py +++ b/tests/laborious/workflows/test_predictions_batch.py @@ -32,7 +32,7 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch 'query': 'SELECT * FROM test', 'schema': 'test_schema', 'table_name': 'test_table', - 'opc_output_config': 'test_opc_output_config', + 'datetime_columns': ['timestamp', 'created_at'], 'prediction_store_policy': 'erl:1', 'model_config': { @@ -78,7 +78,7 @@ async def test_run(workflow_mock: AsyncMock, predictions_batch: PredictionsBatch }), 'model_config': input_data.get('model_config', {}), 'path_priority': input_data.get('path_priority', ['STOP', 'CONTINUE', 'REPEAT']), - 'opc_output_config': input_data.get('opc_output_config', {}), + 'prediction_store_policy': input_data.get('prediction_store_policy', 'erl:1') } diff --git a/validate.sh b/validate.sh new file mode 100755 index 0000000..6d5ab14 --- /dev/null +++ b/validate.sh @@ -0,0 +1,99 @@ +#!/bin/bash +# Model Manager Code Validation Script +# This script runs all code quality checks before committing or deploying + +set -e # Exit on any error + +# Colors for output +RED='\033[0;31m' +GREEN='\033[0;32m' +YELLOW='\033[1;33m' +BLUE='\033[0;34m' +NC='\033[0m' # No Color + +echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" +echo -e "${BLUE}║ Model Manager - Code Validation Suite ║${NC}" +echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" +echo "" + +# Check if virtual environment is activated +if [[ -z "${VIRTUAL_ENV}" ]] && [[ -z "${CONDA_DEFAULT_ENV}" ]]; then + echo -e "${YELLOW}⚠️ Warning: No virtual environment detected${NC}" + echo -e "${YELLOW} Consider activating your venv/conda environment${NC}" + echo "" +fi + +# Function to run a validation step +run_step() { + local step_name=$1 + local step_command=$2 + + echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + echo -e "${BLUE}▶ ${step_name}${NC}" + echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" + + if eval "$step_command"; then + echo -e "${GREEN}✅ ${step_name} - PASSED${NC}" + echo "" + return 0 + else + echo -e "${RED}❌ ${step_name} - FAILED${NC}" + echo "" + return 1 + fi +} + +# Track failures +FAILED_STEPS=() + +# Step 1: Code Formatting Check (Ruff) +if ! run_step "1. Code Formatting (Ruff)" "ruff format --check model_manager/ tests/"; then + FAILED_STEPS+=("Code Formatting") +fi + +# Step 2: Linting (Ruff) +if ! run_step "2. Code Linting (Ruff)" "ruff check model_manager/ tests/"; then + FAILED_STEPS+=("Linting") +fi + +# Step 3: Type Checking (mypy) +if ! run_step "3. Type Checking (mypy)" "mypy model_manager/"; then + FAILED_STEPS+=("Type Checking") +fi + +# Step 4: Security Analysis (Bandit) +if ! run_step "4. Security Analysis (Bandit)" "bandit -r model_manager/ -ll -q"; then + FAILED_STEPS+=("Security Analysis") +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=70 -q"; then + FAILED_STEPS+=("Unit Tests") +fi + +# Summary +echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" +echo -e "${BLUE}║ Validation Summary ║${NC}" +echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" +echo "" + +if [ ${#FAILED_STEPS[@]} -eq 0 ]; then + echo -e "${GREEN}✅ All validation checks passed!${NC}" + echo -e "${GREEN} Your code is ready for commit/deployment.${NC}" + echo "" + exit 0 +else + echo -e "${RED}❌ Validation failed for the following steps:${NC}" + for step in "${FAILED_STEPS[@]}"; do + echo -e "${RED} • ${step}${NC}" + done + echo "" + echo -e "${YELLOW}💡 Tips:${NC}" + echo -e "${YELLOW} • Run 'ruff format model_manager/ tests/' to auto-fix formatting${NC}" + echo -e "${YELLOW} • Run 'ruff check --fix model_manager/ tests/' to auto-fix linting issues${NC}" + echo -e "${YELLOW} • Review mypy errors and add type hints where needed${NC}" + echo -e "${YELLOW} • Check bandit warnings for security issues${NC}" + echo -e "${YELLOW} • Fix failing tests or improve test coverage${NC}" + echo "" + exit 1 +fi diff --git a/values.yaml b/values.yaml index 744c0c0..fc71161 100644 --- a/values.yaml +++ b/values.yaml @@ -180,11 +180,6 @@ env: - name: MLFLOW_PASSWORD value: "1L0FP50j3ncp123" - - name: OPC_ID - value: "1" - - name: OPC_URL - value: "opc.tcp://sientia-opc-simulator-opc.sientia.svc.cluster.local:4840" - - name: KAFKA_BOOTSTRAP_SERVERS value: "kafka.kafka.svc.cluster.local:9092"