gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #29 from Aignosi/feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas

Browse Source 
SIENTIAPDE-1325: Refactor Activities, Remove Couchbase, Update Release Workflow and Dependencies
This commit is contained in:
vitor-aignosi
2025-11-10 10:09:36 -03:00
committed by GitHub
parent f099366bed 92e87530b8
commit b45aa33188
21 changed files with 469 additions and 631 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
.env.example
View File

@@ -3,10 +3,6 @@ REDIS_PORT="6379"
REDIS_USERNAME="redis_username"
REDIS_PASSWORD="redis_password"
COUCHBASE_CONNECTION_STRING="couchbase://sientia.couchbase.svc.cluster.local"
COUCHBASE_USERNAME="sientia"
COUCHBASE_PASSWORD="sientia"
MONGODB_USERNAME="mongo_username"
MONGODB_PASSWORD="mongo_password"
MONGODB_URL="my-release-mongodb.mongodb.svc.cluster.local:27017"

1
.github/workflows/release.yml vendored
View File

@@ -8,6 +8,7 @@ on:
jobs:
release:
if: github.event.pull_request.merged == true
uses: Aignosi/github_workflow_templates/.github/workflows/dataops-module-release.yml@main
permissions: write-all
with:

1
.gitignore vendored
View File

@@ -13,7 +13,6 @@ docker-compose.override.yml
scouter/.file_versions/
scouter/pipelines/**/triggers.yaml
**/postgres_data/**
**/couchbase_data/**
**/redis_data/**
# Ignorar arquivos e diretórios de cache do Python
__pycache__/

3
README.md
View File

@@ -408,7 +408,6 @@ The orchestrator includes an advanced notification filtering system that prevent
- **Email**: SMTP operations with HTML generation and attachment support
- **Formatters**: Configuration processing, slot distribution algorithms, and notification filtering for reports
- **Postgres**: PostgreSQL operations for audit logging and data export (via sientia-dataops-library)
- **Couchbase**: Database operations (currently unused but maintained for future use)
#### **Utilities (`orchestrator/utils/`)**
- **Connectors Configuration**: Database and service configuration management
@@ -642,7 +641,6 @@ The Orchestrator system exposes comprehensive Prometheus metrics:
tests/
├── orchestrator/ # Orchestrator workflow tests
│ ├── test_activities.py
│ ├── test_couchbase.py
│ ├── test_email.py
│ ├── test_formatters.py
│ ├── test_mongo_db.py
@@ -755,7 +753,6 @@ orchestrator/
│ ├── mongo_db.py # MongoDB operations
│ ├── email.py # Email service operations
│ ├── formatters.py # Configuration formatting and report filtering
│ └── couchbase.py # Couchbase operations (currently unused)
├── workflows/ # Temporal workflow definitions
│ ├── orchestrator.py # Main orchestration workflow
│ ├── alerts.py # Error alert workflow

25
orchestrator/activities/activities.py
View File

@@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through():
from typing import Any
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.temporal.activities.postgres import Postgres
from orchestrator.activities.email import Email
@@ -14,9 +15,7 @@ with workflow.unsafe.imports_passed_through():
from orchestrator.activities.temporal_manager import TemporalManager
class Activities( # Couchbase,
TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres
):
class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres):
"""
Central activities orchestrator for Temporal workflow operations.
@@ -47,11 +46,8 @@ class Activities( # Couchbase,
notification_handler: NotificationHandler,
):
# Initialize parent classes
# Couchbase.__init__(self, connection_string=couchbase_config['connection_string'],
# username=couchbase_config['username'],
# password=couchbase_config['password'],
# logger=logger,
# notification_handler=notification_handler)
metrics_controller = MetricsController(logger=logger)
TemporalManager.__init__(
self,
@@ -60,6 +56,7 @@ class Activities( # Couchbase,
laborious_namespace=temporal_config['temporal_laborious_namespace'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
SlotManager.__init__(
@@ -70,6 +67,7 @@ class Activities( # Couchbase,
password=redis_config['password'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Formatters.__init__(
@@ -78,6 +76,7 @@ class Activities( # Couchbase,
laborious_namespace=temporal_config['temporal_laborious_namespace'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
MongoDB.__init__(
@@ -87,6 +86,7 @@ class Activities( # Couchbase,
ttl_index_seconds=mongodb_config['ttl_index_seconds'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Email.__init__(
@@ -97,6 +97,7 @@ class Activities( # Couchbase,
smtp_port=email_config['smtp_port'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
Postgres.__init__(
@@ -110,6 +111,7 @@ class Activities( # Couchbase,
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def shutdown(self):
@@ -120,6 +122,9 @@ class Activities( # Couchbase,
services, and other resources to ensure proper cleanup when the
application terminates.
"""
MongoDB.shutdown(self)
MongoDB.close(self)
Postgres.close(self)
Email.shutdown(self)
Email.close(self)
Formatters.close(self)
SlotManager.close(self)
TemporalManager.close(self)

112
orchestrator/activities/couchbase.py
View File

@@ -1,112 +0,0 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
import json
import traceback
from datetime import timedelta
from logging import Logger
from typing import Any
from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.base import BaseActivity
class Couchbase(BaseActivity):
"""
Couchbase database operations activity (currently unused).
This class provides Couchbase database connectivity and query operations
for Temporal workflows. It handles connection management, query execution,
and error reporting with automatic connection lifecycle management.
Args:
connection_string (str): Couchbase cluster connection string
username (str): Couchbase authentication username
password (str): Couchbase authentication password
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
Note:
This class is currently commented out in the main Activities class
but maintained for potential future use.
"""
def __init__(
self,
connection_string: str,
username: str,
password: str,
logger: Logger,
notification_handler: NotificationHandler,
):
self.connection_string = connection_string
self.username = username
self.password = password
logger.info('Initializing Couchbase connection...')
self.cluster = Cluster(
connection_string,
ClusterOptions(
authenticator=PasswordAuthenticator(username=username, password=password)
),
)
logger.info('Awaiting Couchbase connection...')
self.cluster.wait_until_ready(timeout=timedelta(seconds=10))
logger.info('Couchbase connection ready')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def shutdown(self):
try:
self.cluster.close()
except Exception as e:
self.logger.error(f'Failed to close Couchbase connection: {e}')
def __del__(self):
self.shutdown()
@activity.defn(name='load_query_from_couchbase')
async def load_query_from_couchbase(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Load a query from couchbase
Args:
input_data (dict[str, Any]): The input data containing the query to execute
Returns:
list[dict[str, Any]]: The result of the query
"""
query = input_data['query']
self.logger.info(f'Executing couchbase query: {query}')
try:
result = self.cluster.query(query)
except Exception as e:
trace = traceback.format_exc()
self.notification_handler.build_and_send_notification(
notification_id='COUCHBASE_LOAD_QUERY_ERROR',
message=f'Failed to execute couchbase query: {e}',
block='load_query_from_couchbase',
level=NotificationLevel.ERROR,
attachment_content=trace,
)
self.logger.error(trace)
raise e
rows = []
for row in result.rows():
rows.append(row)
self.logger.info('Fetched %d rows from couchbase', len(rows))
self.logger.debug('Rows: \n %s', json.dumps(rows, indent=4, sort_keys=True))
return rows

23
orchestrator/activities/email.py
View File

@@ -12,13 +12,13 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.notifications.handlers import NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from orchestrator import metrics
from orchestrator.utils.email_builder import EmailBuilder
class Email(BaseActivity):
class Email(SientiaMonitoring):
"""
Email service activity for sending workflow notifications.
@@ -43,6 +43,7 @@ class Email(BaseActivity):
smtp_port: int,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
self.email_builder = EmailBuilder(logger=logger)
@@ -60,13 +61,25 @@ class Email(BaseActivity):
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def shutdown(self):
def close(self):
"""
Shutdown the Email connection and clean up resources.
Close the Email connection and clean up resources.
"""
self.server.quit()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the Email connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='build_email_html')
async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:

52
orchestrator/activities/formatters.py
View File

@@ -10,7 +10,7 @@ with workflow.unsafe.imports_passed_through():
from pandas import DataFrame
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.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
from orchestrator.utils.orchestrator_functions import (
@@ -24,7 +24,7 @@ with workflow.unsafe.imports_passed_through():
topic_separator = '\n ========== \n'
class Formatters(BaseActivity):
class Formatters(SientiaMonitoring):
"""
Schedule and slot configuration formatting and notification filtering activity.
@@ -53,10 +53,28 @@ class Formatters(BaseActivity):
laborious_namespace: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
self.scouter_namespace = scouter_namespace
self.laborious_namespace = laborious_namespace
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self):
"""
Close the Formatters connection and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the Formatters connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='process_schedules')
async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -166,7 +184,7 @@ class Formatters(BaseActivity):
slot_config[f'{i}'], notifications = build_tag_config(slot_tags, opc_servers)
if notifications:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message=f'Servers {", ".join(notifications)} not found in opc_servers',
@@ -179,7 +197,7 @@ class Formatters(BaseActivity):
slot_tags = tags[last_index:]
slot_config[f'{number_of_slots}'], notifications = build_tag_config(slot_tags, opc_servers)
if notifications:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message=f'Servers {", ".join(notifications)} not found in opc_servers',
@@ -357,7 +375,7 @@ class Formatters(BaseActivity):
return output
def send_success_report(
async def send_success_report(
self,
metadata: dict[str, Any],
message: str,
@@ -373,7 +391,7 @@ class Formatters(BaseActivity):
notification_id (str): The ID of the notification to send.
attachment (Any | None, optional): Optional attachment content to include.
"""
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=notification_id,
message=message,
@@ -382,7 +400,7 @@ class Formatters(BaseActivity):
attachment_content=json.dumps(attachment, indent=4, sort_keys=True),
)
def send_error_report(
async def send_error_report(
self, metadata: dict[str, Any], message: str, notification_id: str, attachment: str
) -> None:
"""
@@ -394,7 +412,7 @@ class Formatters(BaseActivity):
notification_id (str): The ID of the notification.
attachment (str): The attachment content for the notification.
"""
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id=notification_id,
message=message,
@@ -455,7 +473,7 @@ class Formatters(BaseActivity):
return success_keys, error_keys
def manage_and_send_report(
async def manage_and_send_report(
self,
metadata: dict[str, Any],
success_keys: list[str],
@@ -474,7 +492,7 @@ class Formatters(BaseActivity):
schedule_data (dict[str, Any]): The schedule data containing items and notification ID.
"""
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Successfully {schedule_type}: \n {", ".join(success_keys)}',
notification_id=schedule_data['id'],
@@ -489,7 +507,7 @@ class Formatters(BaseActivity):
else:
attachment.append(f'{key}:\n{value["message"]}')
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Fails on {schedule_type}: \n {", ".join(error_keys)}',
notification_id=f'{schedule_data["id"]}_ERROR',
@@ -535,7 +553,7 @@ class Formatters(BaseActivity):
if len(schedule_data['items']) > 0:
success_keys, error_keys = self.parse_report_schedule(schedule_data['items'])
self.manage_and_send_report(
await self.manage_and_send_report(
metadata=metadata,
success_keys=success_keys,
error_keys=error_keys,
@@ -566,14 +584,14 @@ class Formatters(BaseActivity):
success_keys, error_keys = self.parse_report(inserted_slots)
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Inserted slots: \n {", ".join(success_keys)}',
notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
)
if len(error_keys) > 0:
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Failed to insert slots: \n {", ".join(error_keys)}',
notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
@@ -584,14 +602,14 @@ class Formatters(BaseActivity):
success_keys, error_keys = self.parse_report(deleted_slots)
if len(success_keys) > 0:
self.send_success_report(
await self.send_success_report(
metadata=metadata,
message=f'Deleted slots: \n {", ".join(success_keys)}',
notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',
)
if len(error_keys) > 0:
self.send_error_report(
await self.send_error_report(
metadata=metadata,
message=f'Failed to delete slots: \n {", ".join(error_keys)}',
notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',

161
orchestrator/activities/mongo_db.py
View File

@@ -6,41 +6,18 @@ with workflow.unsafe.imports_passed_through():
from logging import Logger
from typing import Any
from pymongo import MongoClient
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.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.repository.mongodb_repository import MongoDBRepository
from sientia_do.temporal.constants import (
DATETIME_FORMAT_MS_WITH_TZ,
DATETIME_FORMAT_WITH_TZ,
now,
)
def clear_mongo_id(docs: list) -> list:
"""
Remove MongoDB internal `_id` fields from nested structures.
Args:
docs (list): The list of documents or nested structures to clean.
Returns:
list: The cleaned documents with `_id` fields removed wherever present.
"""
for doc in docs:
if isinstance(doc, list):
clear_mongo_id(doc)
elif isinstance(doc, dict):
if '_id' in doc:
del doc['_id']
for _key, value in doc.items():
if isinstance(value, list):
clear_mongo_id(value)
elif isinstance(value, dict):
clear_mongo_id([value])
return docs
class MongoDB(BaseActivity):
class MongoDB(SientiaMonitoring):
"""
MongoDB operations activity for Temporal workflows.
@@ -64,60 +41,43 @@ class MongoDB(BaseActivity):
ttl_index_seconds: int,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.connection_string = connection_string
self.database_name = database_name
self.client: MongoClient = MongoClient(
self.connection_string, serverSelectionTimeoutMS=5000
self.mongo_db_repository = MongoDBRepository(
connection_string=connection_string,
database_name=database_name,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.client.server_info() # Force early failure if connection is invalid
self.database = self.client[self.database_name]
self.ttl_index_seconds = ttl_index_seconds
# Initialize MongoDB client here (omitted for brevity)
logger.info('MongoDB connection initialized')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def shutdown(self):
def close(self):
"""
Shutdown the MongoDB client and clean up resources.
"""
try:
if self.client:
self.logger.info('Closing MongoDB connection...')
self.client.close()
self.logger.info('MongoDB connection closed successfully')
except Exception as e:
self.logger.error(f'Failed to close MongoDB connection: {e}')
self.mongo_db_repository.close()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the MongoDB client is closed when the object is garbage-collected.
"""
self.shutdown()
def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]:
"""
Find documents in a MongoDB collection based on the provided filters.
Args:
collection_name (str): The name of the collection to search in.
filters (dict[str, Any]): The query filters to apply.
Returns:
list[dict[str, Any]]: Documents matching the filters (with `_id` removed).
"""
collection = self.database[collection_name]
documents = list(collection.find(filters, {'_id': 0}))
documents = clear_mongo_id(documents)
return documents
self.close()
@activity.defn(
name='find_documents_in_mongodb',
@@ -151,7 +111,7 @@ class MongoDB(BaseActivity):
)
try:
documents = self.find(collection_name, filters)
documents = await self.mongo_db_repository.find(collection_name, filters, metadata)
self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'",
@@ -173,7 +133,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_QUERY_ERROR',
message=f'Failed to execute MongoDB query: {e}',
@@ -219,11 +179,9 @@ class MongoDB(BaseActivity):
)
try:
collection = self.database[collection_name]
aggregated_documents = list(collection.aggregate(aggregation))
aggregated_documents = clear_mongo_id(aggregated_documents)
aggregated_documents = await self.mongo_db_repository.aggregate(
collection_name, aggregation, metadata
)
self.info(
f"Aggregated {len(aggregated_documents)} documents from collection '{collection_name}'",
@@ -245,7 +203,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_AGGREGATION_ERROR',
message=f'Failed to execute MongoDB aggregation: {e}',
@@ -268,12 +226,9 @@ class MongoDB(BaseActivity):
updated_pipelines = input_data.get('updated_pipelines', [])
metadata = input_data.get('metadata', {})
date_now = now()
collection = self.database['orchestrated_schedules']
self.info('Updating pipelines timestamps...', metadata=metadata)
success_count = 0
argument = [
{'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
for pipeline in updated_pipelines
@@ -282,11 +237,12 @@ class MongoDB(BaseActivity):
data_filter = {'$or': argument} if argument else {}
try:
collection.update_many(data_filter, {'$set': {'updated_at': date_now}})
success_count += 1
await self.mongo_db_repository.update_many(
'orchestrated_schedules', data_filter, {'$set': {'updated_at': date_now}}, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_UPDATE_PIPELINES_ERROR',
message=f'Failed to update pipelines timestamps: {e}',
@@ -298,7 +254,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Updated {success_count} of {len(updated_pipelines)} pipelines timestamps',
f'Updated {len(updated_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -312,12 +268,9 @@ class MongoDB(BaseActivity):
"""
created_pipelines = input_data.get('created_pipelines', [])
metadata = input_data.get('metadata', {})
collection = self.database['orchestrated_schedules']
self.info('Creating pipelines timestamps...', metadata=metadata)
success_count = 0
date_now = now()
argument = [
@@ -333,11 +286,12 @@ class MongoDB(BaseActivity):
try:
if data_filter:
collection.insert_many(data_filter)
success_count += 1
await self.mongo_db_repository.insert_many(
'orchestrated_schedules', data_filter, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_CREATE_PIPELINES_ERROR',
message=f'Failed to create pipelines timestamps: {e}',
@@ -349,7 +303,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Created {success_count} of {len(created_pipelines)} pipelines timestamps',
f'Created {len(created_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -363,12 +317,9 @@ class MongoDB(BaseActivity):
"""
deleted_pipelines = input_data.get('deleted_pipelines', [])
metadata = input_data.get('metadata', {})
collection = self.database['orchestrated_schedules']
self.info('Deleting pipelines timestamps...', metadata=metadata)
success_count = 0
argument = [
{'schedule_name': pipeline['schedule_name'], 'namespace': pipeline['namespace']}
for pipeline in deleted_pipelines
@@ -377,11 +328,12 @@ class MongoDB(BaseActivity):
data_filter = {'$or': argument} if argument else {}
try:
collection.delete_many(data_filter)
success_count += 1
await self.mongo_db_repository.delete_many(
'orchestrated_schedules', data_filter, metadata
)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_DELETE_PIPELINES_ERROR',
message=f'Failed to delete pipelines timestamps: {e}',
@@ -393,7 +345,7 @@ class MongoDB(BaseActivity):
raise e
self.info(
f'Deleted {success_count} of {len(deleted_pipelines)} pipelines timestamps',
f'Deleted {len(deleted_pipelines)} pipelines timestamps',
metadata=metadata,
)
@@ -413,7 +365,7 @@ class MongoDB(BaseActivity):
metadata=metadata,
)
collection_names = self.database.list_collection_names()
collection_names = self.mongo_db_repository.database.list_collection_names()
created_collections = []
created_indexes = []
@@ -424,10 +376,10 @@ class MongoDB(BaseActivity):
try:
# Check if collection exists
if collection not in collection_names:
self.database.create_collection(collection)
self.mongo_db_repository.database.create_collection(collection)
created_collections.append(collection)
collection = self.database[collection]
collection = self.mongo_db_repository.database[collection]
# Check if TTL index exists
existing_indexes = collection.list_indexes()
ttl_index_exists = False
@@ -448,7 +400,7 @@ class MongoDB(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGODB_CREATE_COLLECTION_ERROR',
message=f'Failed to create collection {collection} with TTL index: {e}',
@@ -504,21 +456,16 @@ class MongoDB(BaseActivity):
data_filter = {
**base_data_filter,
'timestamp': {
'$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_MS_WITH_TZ)
'$gt': datetime.strptime(last_data_timestamp, DATETIME_FORMAT_WITH_TZ)
},
}
self.debug(f'Data filter: {data_filter}', metadata=metadata)
data = self.find(collection_name, data_filter)
data = await self.mongo_db_repository.find(collection_name, data_filter, metadata)
self.debug(f'Collected: {data}', metadata=metadata)
for item in data:
item['timestamp'] = (
item['timestamp'].replace(tzinfo=UTC).strftime(DATETIME_FORMAT_MS_WITH_TZ)
)
self.info(f'Loaded {len(data)} documents from MongoDB', metadata=metadata)
self.debug(f'Loaded data: {data}', metadata=metadata)
@@ -526,7 +473,7 @@ class MongoDB(BaseActivity):
return data
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='MONGO_LOAD_ERROR',
message=f'Error loading data from MongoDB: {e}',

62
orchestrator/activities/slot_manager.py
View File

@@ -10,11 +10,12 @@ with workflow.unsafe.imports_passed_through():
from pandas import DataFrame
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.activities.redis_base import Redis
from sientia_do.observability.sientia_monitoring import MetricsController, SientiaMonitoring
from sientia_do.repository.redis_repository import RedisRepository
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, now
class SlotManager(Redis):
class SlotManager(SientiaMonitoring):
"""
Redis-based OPC slot management and notification filtering activity.
@@ -48,8 +49,37 @@ class SlotManager(Redis):
password: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
Redis.__init__(self, host, port, username, password, logger, notification_handler)
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.redis_repository = RedisRepository(
host=host,
port=port,
username=username,
password=password,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self):
"""
Close the SlotManager connection and clean up resources.
"""
self.redis_repository.close()
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the SlotManager connection is closed when the object is garbage-collected.
"""
self.close()
@activity.defn(name='load_opc_slots')
async def load_opc_slots(self, input_data: dict[str, Any]) -> dict[str, Any]:
@@ -80,7 +110,7 @@ class SlotManager(Redis):
opc_slots = {}
try:
slot_keys = self.redis_client.keys('slot:opc_tags:*')
slot_keys = await self.redis_repository.keys('slot:opc_tags:*')
self.debug(f'Slot keys: {slot_keys}', metadata=metadata)
@@ -91,10 +121,10 @@ class SlotManager(Redis):
decoded_keys = slot_keys
for key in decoded_keys:
opc_slots[key] = self.get(key)
opc_slots[key] = await self.redis_repository.get(key)
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Failed to load OPC slots: {e}',
@@ -123,7 +153,7 @@ class SlotManager(Redis):
self.info('Loading active ingestors...', metadata=metadata)
try:
active_ingestors = self.redis_client.keys('heartbeat:ingestor:*')
active_ingestors = await self.redis_repository.keys('heartbeat:ingestor:*')
self.info(f'Loaded {len(active_ingestors)} active ingestors', metadata=metadata)
@@ -140,7 +170,7 @@ class SlotManager(Redis):
return ingestors
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Failed to load active ingestors: {e}',
@@ -175,7 +205,7 @@ class SlotManager(Redis):
for slot in to_insert:
try:
self.set(f'slot:opc_tags:{slot}', to_insert[slot], ttl=None)
await self.redis_repository.set(f'slot:opc_tags:{slot}', to_insert[slot], ttl=None)
report[slot] = {'success': True, 'message': 'Slot updated successfully'}
success_count += 1
except Exception as e:
@@ -213,7 +243,7 @@ class SlotManager(Redis):
for slot in to_delete:
try:
self.redis_client.delete(f'slot:opc_tags:{slot}')
await self.redis_repository.delete(f'slot:opc_tags:{slot}')
report[slot] = {'success': True, 'message': 'Slot deleted successfully'}
success_count += 1
except Exception as e:
@@ -243,9 +273,9 @@ class SlotManager(Redis):
key = f'notification_last_timestamp:{input_data["mail_type"]}'
try:
data_hold = self.get(key)
data_hold = await self.redis_repository.get(key)
except Exception as e:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_GET_ERROR',
message=f'Error getting last data timestamp: {e}',
@@ -290,9 +320,9 @@ class SlotManager(Redis):
self.debug(f'Last collected timestamp to insert: {last_data_timestamp}', metadata=metadata)
try:
self.set(key, last_data_timestamp, ttl=60 * 60 * 5)
await self.redis_repository.set(key, last_data_timestamp, ttl=60 * 60 * 5)
except Exception as e:
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='REDIS_SET_ERROR',
message=f'Error setting last data timestamp: {e}',
@@ -361,7 +391,7 @@ class SlotManager(Redis):
# Check if notification was recently sent
key = f'{notification["trigger"]}:{notification_id}'
last_sent = self.get(key)
last_sent = await self.redis_repository.get(key)
if last_sent is None:
alert_type = 'core_alerts'
@@ -414,6 +444,6 @@ class SlotManager(Redis):
status = row['status']
if status == 'sent':
key = f'{row["schedule"]}:{row["notification_id"]}'
self.set(key, date_now, ttl=sent_ttl)
await self.redis_repository.set(key, date_now, ttl=sent_ttl)
self.info('Notification cache stored...', metadata=metadata)

25
orchestrator/activities/temporal_manager.py
View File

@@ -19,12 +19,12 @@ 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.sientia_monitoring import MetricsController, SientiaMonitoring
from orchestrator.utils.converters import parse_frequency
class TemporalManager(BaseActivity):
class TemporalManager(SientiaMonitoring):
"""
Temporal workflow and schedule management activity.
@@ -48,7 +48,14 @@ class TemporalManager(BaseActivity):
laborious_namespace: str,
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
):
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
self.temporal_host = host
self.scouter_namespace = scouter_namespace
self.laborious_namespace = laborious_namespace
@@ -58,7 +65,17 @@ class TemporalManager(BaseActivity):
self.model_name_id_key = SearchAttributeKey.for_keyword('model_name')
self.orchestrated_id_key = SearchAttributeKey.for_keyword('orchestrated')
BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler)
def close(self):
"""
Close the TemporalManager connection and clean up resources.
"""
SientiaMonitoring.shutdown(self)
def __del__(self):
"""
Ensure the TemporalManager connection is closed when the object is garbage-collected.
"""
self.close()
async def connect_to_temporal(self):
"""
@@ -127,7 +144,7 @@ class TemporalManager(BaseActivity):
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
await self.send_notification_async(
metadata=metadata,
notification_id='SCHEDULER_NORMALIZE_SCHEDULES_ERROR',
message=f'Failed to normalize schedules: {e}',

6
pyproject.toml
View File

@@ -114,11 +114,7 @@ python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"-v",
"--strict-markers",
"--cov=model_manager",
"--cov-report=term-missing",
"--cov-report=html",
"--cov-report=xml",
"--strict-markers"
]
markers = [
"asyncio: marks tests as async",

3
requirements.txt
View File

@@ -2,8 +2,7 @@ temporalio
psycopg2-binary
sqlalchemy
redis
couchbase
pymongo
jinja2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.4.7
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.5.3
prometheus-client

45
tests/orchestrator/activities/test_activities.py
View File

@@ -13,7 +13,9 @@ from orchestrator.activities.temporal_manager import TemporalManager
@patch('orchestrator.activities.formatters.Formatters.__init__')
@patch('orchestrator.activities.email.Email.__init__')
@patch('sientia_do.temporal.activities.postgres.Postgres.__init__')
@patch('orchestrator.activities.activities.MetricsController')
def test___init__(
mock_metrics_controller,
mock_postgres_init,
mock_email_init,
mock_formatters_init,
@@ -79,6 +81,7 @@ def test___init__(
password='password',
logger=logger,
notification_handler=notification_handler,
metrics_controller=mock_metrics_controller.return_value,
)
mock_mongodb_init.assert_called_once_with(
@@ -88,6 +91,7 @@ def test___init__(
ttl_index_seconds=3600,
logger=logger,
notification_handler=notification_handler,
metrics_controller=mock_metrics_controller.return_value,
)
mock_temporal_manager_init.assert_called_once_with(
@@ -97,6 +101,7 @@ def test___init__(
laborious_namespace='laborious',
logger=logger,
notification_handler=notification_handler,
metrics_controller=mock_metrics_controller.return_value,
)
mock_formatters_init.assert_called_once_with(
@@ -105,28 +110,23 @@ def test___init__(
laborious_namespace='laborious',
logger=logger,
notification_handler=notification_handler,
metrics_controller=mock_metrics_controller.return_value,
)
@patch('orchestrator.activities.mongo_db.MongoDB.__init__')
@patch('orchestrator.activities.temporal_manager.TemporalManager.__init__')
@patch('orchestrator.activities.slot_manager.SlotManager.__init__')
@patch('orchestrator.activities.formatters.Formatters.__init__')
@patch('orchestrator.activities.email.Email.__init__')
@patch('sientia_do.temporal.activities.postgres.Postgres.__init__')
@patch('orchestrator.activities.email.Email.shutdown')
@patch('sientia_do.temporal.activities.postgres.Postgres.close')
@patch('orchestrator.activities.mongo_db.MongoDB.shutdown')
@patch('orchestrator.activities.activities.MongoDB')
@patch('orchestrator.activities.activities.TemporalManager')
@patch('orchestrator.activities.activities.SlotManager')
@patch('orchestrator.activities.activities.Formatters')
@patch('orchestrator.activities.activities.Email')
@patch('orchestrator.activities.activities.Postgres')
def test_shutdown(
mock_mongodb_close,
mock_postgres_shutdown,
mock_email_close,
mock_postgres_init,
mock_email_init,
mock_formatters_init,
mock_slot_manager_init,
mock_temporal_manager_init,
mock_mongodb_init,
mock_mongodb,
mock_temporal_manager,
mock_slot_manager,
mock_formatters,
mock_email,
mock_postgres,
):
activities = Activities(
temporal_config=MagicMock(),
@@ -140,6 +140,9 @@ def test_shutdown(
activities.shutdown()
mock_mongodb_close.assert_called()
mock_postgres_shutdown.assert_called()
mock_email_close.assert_called()
mock_mongodb.close.assert_called()
mock_temporal_manager.close.assert_called()
mock_slot_manager.close.assert_called()
mock_formatters.close.assert_called()
mock_email.close.assert_called()
mock_postgres.close.assert_called()

75
tests/orchestrator/activities/test_couchbase.py
View File

@@ -1,75 +0,0 @@
from unittest.mock import ANY, MagicMock, patch
from pytest import fixture, mark, raises
from sientia_do.notifications.models import NotificationLevel
from orchestrator.activities.couchbase import Couchbase
@fixture
@patch('orchestrator.activities.couchbase.Cluster')
def couchbase(_cluster_mock):
return Couchbase(
connection_string='couchbase://localhost',
username='admin',
password='password',
logger=MagicMock(),
notification_handler=MagicMock(),
)
def test_shutdown_success(couchbase):
couchbase.shutdown()
couchbase.cluster.close.assert_called_once()
def test_shutdown_failure(couchbase):
couchbase.cluster.close.side_effect = Exception('Test error')
couchbase.shutdown()
couchbase.logger.error.assert_called_once_with(
f'Failed to close Couchbase connection: {couchbase.cluster.close.side_effect}'
)
@mark.asyncio
async def test_load_query_from_couchbase_success(couchbase):
couchbase.cluster.query.return_value.rows.return_value = [
{'id': '1', 'name': 'test'},
{'id': '2', 'name': 'test2'},
]
query = 'SELECT * FROM bucket'
result = await couchbase.load_query_from_couchbase(
{
'query': query,
}
)
assert result == [
{'id': '1', 'name': 'test'},
{'id': '2', 'name': 'test2'},
]
couchbase.cluster.query.assert_called_once_with(query)
couchbase.notification_handler.build_and_send_notification.assert_not_called()
@mark.asyncio
async def test_load_query_from_couchbase_failure(couchbase):
couchbase.cluster.query.side_effect = ValueError('Test error')
query = 'SELECT * FROM bucket'
with raises(ValueError):
await couchbase.load_query_from_couchbase(
{
'query': query,
}
)
couchbase.cluster.query.assert_called_once_with(query)
couchbase.notification_handler.build_and_send_notification.assert_called_once_with(
notification_id='COUCHBASE_LOAD_QUERY_ERROR',
message='Failed to execute couchbase query: Test error',
block='load_query_from_couchbase',
level=NotificationLevel.ERROR,
attachment_content=ANY,
)

15
tests/orchestrator/activities/test_email.py
View File

@@ -1,5 +1,5 @@
from smtplib import SMTPServerDisconnected
from unittest.mock import MagicMock, call, patch
from unittest.mock import AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
@@ -17,8 +17,13 @@ def email(smtplib, email_builder):
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
email_builder.send_notification_async = AsyncMock()
email_builder.send_notification = MagicMock()
email.send_notification_async = AsyncMock()
email.send_notification = MagicMock()
email.emit_metric = AsyncMock()
return email
@@ -33,6 +38,7 @@ def test___init___with_password(smtplib, email_builder):
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
assert email.sender_email == 'test@test.com'
@@ -56,6 +62,7 @@ def test___init___without_password(smtplib, email_builder):
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
assert email.sender_email == 'test@test.com'
@@ -76,10 +83,12 @@ metadata = {
}
def test_shutdown(email):
email.shutdown()
@patch('orchestrator.activities.email.SientiaMonitoring')
def test_close(sientia_monitoring_mock, email):
email.close()
email.server.quit.assert_called_once()
sientia_monitoring_mock.shutdown.assert_called_once()
@mark.asyncio

29
tests/orchestrator/activities/test_formatters.py
View File

@@ -1,5 +1,5 @@
import json
from unittest.mock import MagicMock, call, patch
from unittest.mock import AsyncMock, MagicMock, call, patch
from pandas import DataFrame
from pytest import fixture, mark
@@ -15,9 +15,12 @@ def formatters():
laborious_namespace='laborious',
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
formatters.send_notification = MagicMock()
formatters.send_notification_async = AsyncMock()
formatters.emit_metric = AsyncMock()
return formatters
@@ -173,7 +176,7 @@ async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, forma
]
)
formatters.send_notification.assert_has_calls(
formatters.send_notification_async.assert_has_calls(
[
call(
metadata=metadata['metadata'],
@@ -288,14 +291,15 @@ async def test_create_slot_config(formatters):
}
def test_send_success_report(formatters):
formatters.send_success_report(
@mark.asyncio
async def test_send_success_report(formatters):
await formatters.send_success_report(
metadata=metadata,
message='test_message',
notification_id='test_notification_id',
attachment={'test': 'test'},
)
formatters.send_notification.assert_called_once_with(
formatters.send_notification_async.assert_called_once_with(
metadata=metadata,
notification_id='test_notification_id',
message='test_message',
@@ -305,14 +309,15 @@ def test_send_success_report(formatters):
)
def test_send_error_report(formatters):
formatters.send_error_report(
@mark.asyncio
async def test_send_error_report(formatters):
await formatters.send_error_report(
metadata=metadata,
message='test_message',
notification_id='test_notification_id',
attachment='test_attachment',
)
formatters.send_notification.assert_called_once_with(
formatters.send_notification_async.assert_called_once_with(
metadata=metadata,
notification_id='test_notification_id',
message='test_message',
@@ -364,8 +369,8 @@ def test_parse_report_schedule(formatters):
@mark.asyncio
async def test_report_schedule_orchestration(formatters):
formatters.parse_report_schedule = MagicMock(side_effect=formatters.parse_report_schedule)
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.send_success_report = AsyncMock()
formatters.send_error_report = AsyncMock()
input_data = {
**metadata,
@@ -496,8 +501,8 @@ async def test_report_schedule_orchestration(formatters):
@mark.asyncio
async def test_report_slot_orchestration(formatters):
formatters.parse_report = MagicMock(side_effect=formatters.parse_report)
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.send_success_report = AsyncMock()
formatters.send_error_report = AsyncMock()
input_data = {
**metadata,

317
tests/orchestrator/activities/test_mongo_db.py
View File

@@ -1,47 +1,15 @@
from datetime import datetime
from unittest.mock import ANY, MagicMock, patch
from unittest.mock import ANY, AsyncMock, MagicMock, patch
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ
from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ, DATETIME_FORMAT_WITH_TZ
from orchestrator.activities.mongo_db import MongoDB, clear_mongo_id
def test_clear_mongo_id():
input_data = [
[
{
'name': 'test',
'_id': '12345',
}
],
{
'name': 'test',
'_id': '12345',
'nested': {
'_id': '67890',
'value': [1, 2, 3],
'list': [{'_id': 'abcde', 'item': 'value'}],
},
'nested_list': [{'_id': 'fghij', 'item': 'value1'}, {'_id': 'klmno', 'item': 'value2'}],
},
]
output = clear_mongo_id(input_data)
assert output == [
[{'name': 'test'}],
{
'name': 'test',
'nested': {'value': [1, 2, 3], 'list': [{'item': 'value'}]},
'nested_list': [{'item': 'value1'}, {'item': 'value2'}],
},
]
from orchestrator.activities.mongo_db import MongoDB
@fixture
@patch('orchestrator.activities.mongo_db.MongoClient')
@patch('orchestrator.activities.mongo_db.MongoDBRepository')
def mongo_db(mongo_mock):
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
@@ -49,63 +17,72 @@ def mongo_db(mongo_mock):
ttl_index_seconds=3600,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
mongo.send_notification = MagicMock()
mongo.send_notification_async = AsyncMock()
mongo.emit_metric = AsyncMock()
return mongo
@patch('orchestrator.activities.mongo_db.MongoClient')
@patch('orchestrator.activities.mongo_db.MongoDBRepository')
def test___init__(mongo_mock):
mongo_db = MongoDB(
logger = MagicMock()
notification_handler = MagicMock()
metrics_controller = AsyncMock()
MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
ttl_index_seconds=3600,
logger=MagicMock(),
notification_handler=MagicMock(),
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
assert mongo_db.connection_string == 'mongodb://localhost:27017'
assert mongo_db.database_name == 'test_db'
mongo_mock.assert_called_once_with('mongodb://localhost:27017', serverSelectionTimeoutMS=5000)
mongo_db.client.server_info.assert_called_once()
mongo_db.client.__getitem__.assert_called_once_with('test_db')
def test_shutdown_success(mongo_db):
mongo_db.shutdown()
mongo_db.client.close.assert_called_once()
mongo_db.logger.info.assert_any_call('Closing MongoDB connection...')
mongo_db.logger.info.assert_any_call('MongoDB connection closed successfully')
def test_shutdown_failure(mongo_db):
mongo_db.client.close.side_effect = Exception('Close failed')
mongo_db.shutdown()
mongo_db.logger.error.assert_called_once_with(
'Failed to close MongoDB connection: Close failed'
mongo_mock.assert_called_once_with(
connection_string='mongodb://localhost:27017',
database_name='test_db',
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
@patch('orchestrator.activities.mongo_db.SientiaMonitoring')
def test_close(sientia_monitoring_mock, mongo_db):
mongo_db.close()
mongo_db.mongo_db_repository.close.assert_called_once()
sientia_monitoring_mock.shutdown.assert_called_once()
def test___del__(mongo_db):
mongo_db.close = MagicMock()
mongo_db.__del__()
mongo_db.close.assert_called_once()
@mark.asyncio
async def test_find_documents_in_mongodb_success(mongo_db):
input_data = {'collection': 'test_collection', 'filters': {'name': {'$exists': True}}}
mock_collection = MagicMock()
mock_collection.find.return_value = [
{
'_id': '12345',
'name': 'test1',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
{
'_id': '67890',
'name': 'test2',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
]
mongo_db.database.__getitem__.return_value = mock_collection
mongo_db.mongo_db_repository.find = AsyncMock(
return_value=[
{
'name': 'test1',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
{
'name': 'test2',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
]
)
result = await mongo_db.find_documents_in_mongodb(
{'query': input_data, 'timestamp_fields': ['timestamp']}
@@ -114,7 +91,9 @@ async def test_find_documents_in_mongodb_success(mongo_db):
assert len(result) == 2
assert result[0] == {'name': 'test1', 'timestamp': '2023-01-01 12:00:00.000000+0000'}
assert result[1] == {'name': 'test2', 'timestamp': '2023-01-01 12:00:00.000000+0000'}
mock_collection.find.assert_called_once_with({'name': {'$exists': True}}, {'_id': 0})
mongo_db.mongo_db_repository.find.assert_called_once_with(
'test_collection', {'name': {'$exists': True}}, {}
)
metadata = {
@@ -130,15 +109,13 @@ metadata = {
@mark.asyncio
async def test_find_documents_in_mongodb_failure(mongo_db):
input_data = {'collection': 'test_collection', 'filters': {'name': {'$exists': True}}}
mongo_db.database.__getitem__.return_value = MagicMock(
find=MagicMock(side_effect=Exception('Error'))
)
mongo_db.mongo_db_repository.find = AsyncMock(side_effect=Exception('Error'))
try:
await mongo_db.find_documents_in_mongodb({'query': input_data, **metadata})
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_QUERY_ERROR',
message='Failed to execute MongoDB query: Error',
@@ -171,24 +148,22 @@ async def test_aggregate_documents_in_mongodb_success(mongo_db):
'collection': 'test_collection',
'aggregation': [{'$match': {'name': {'$exists': True}}}, {'$project': {'name': 1}}],
}
mock_collection = MagicMock()
mock_collection.aggregate.return_value = [
{
'_id': 'asdad',
'name': 'test1',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
{
'_id': 'adzx',
'name': 'test2',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
]
mongo_db.database.__getitem__.return_value = mock_collection
mongo_db.mongo_db_repository.aggregate = AsyncMock(
return_value=[
{
'name': 'test1',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
{
'name': 'test2',
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
},
]
)
result = await mongo_db.aggregate_documents_in_mongodb(
{'query': input_data, 'timestamp_fields': ['timestamp']}
@@ -200,7 +175,9 @@ async def test_aggregate_documents_in_mongodb_success(mongo_db):
expected_pipeline = input_data['aggregation']
expected_pipeline.append({'$project': {'_id': 0}})
mock_collection.aggregate.assert_called_once_with(expected_pipeline)
mongo_db.mongo_db_repository.aggregate.assert_called_once_with(
'test_collection', expected_pipeline, {}
)
@mark.asyncio
@@ -209,15 +186,13 @@ async def test_aggregate_documents_in_mongodb_failure(mongo_db):
'collection': 'test_collection',
'aggregation': [{'$match': {'name': {'$exists': True}}}, {'$project': {'name': 1}}],
}
mongo_db.database.__getitem__.return_value = MagicMock(
aggregate=MagicMock(side_effect=Exception('Error'))
)
mongo_db.mongo_db_repository.aggregate = AsyncMock(side_effect=Exception('Error'))
try:
await mongo_db.aggregate_documents_in_mongodb({'query': input_data, **metadata})
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_AGGREGATION_ERROR',
message='Failed to execute MongoDB aggregation: Error',
@@ -267,9 +242,10 @@ async def test_update_pipelines_timestamps_success(now_mock, mongo_db):
{'schedule_name': 'test2', 'namespace': 'test2', 'success': True},
]
}
mongo_db.database['pipelines'].update_many.return_value = MagicMock()
mongo_db.mongo_db_repository.update_many = AsyncMock(return_value=MagicMock())
await mongo_db.update_pipelines_timestamps(input_data)
mongo_db.database['pipelines'].update_many.assert_called_once_with(
mongo_db.mongo_db_repository.update_many.assert_called_once_with(
'orchestrated_schedules',
{
'$or': [
{'schedule_name': 'test1', 'namespace': 'test1'},
@@ -277,6 +253,7 @@ async def test_update_pipelines_timestamps_success(now_mock, mongo_db):
]
},
{'$set': {'updated_at': now_mock.return_value}},
{},
)
@@ -291,12 +268,12 @@ async def test_update_pipelines_timestamps_failure(now_mock, mongo_db):
**metadata,
}
mongo_db.database['pipelines'].update_many.side_effect = Exception('Error')
mongo_db.mongo_db_repository.update_many.side_effect = Exception('Error')
try:
await mongo_db.update_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_UPDATE_PIPELINES_ERROR',
message='Failed to update pipelines timestamps: Error',
@@ -318,13 +295,15 @@ async def test_create_pipelines_timestamps_success(now_mock, mongo_db):
{'schedule_name': 'test2', 'namespace': 'test2', 'success': True},
]
}
mongo_db.database['pipelines'].insert_many.return_value = MagicMock()
mongo_db.mongo_db_repository.insert_many = AsyncMock(return_value=MagicMock())
await mongo_db.create_pipelines_timestamps(input_data)
mongo_db.database['pipelines'].insert_many.assert_called_once_with(
mongo_db.mongo_db_repository.insert_many.assert_called_once_with(
'orchestrated_schedules',
[
{'schedule_name': 'test1', 'namespace': 'test1', 'updated_at': now_mock.return_value},
{'schedule_name': 'test2', 'namespace': 'test2', 'updated_at': now_mock.return_value},
]
],
{},
)
@@ -338,12 +317,12 @@ async def test_create_pipelines_timestamps_failure(now_mock, mongo_db):
],
**metadata,
}
mongo_db.database['pipelines'].insert_many.side_effect = Exception('Error')
mongo_db.mongo_db_repository.insert_many.side_effect = Exception('Error')
try:
await mongo_db.create_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_CREATE_PIPELINES_ERROR',
message='Failed to create pipelines timestamps: Error',
@@ -365,15 +344,17 @@ async def test_delete_pipelines_timestamps_success(now_mock, mongo_db):
{'schedule_name': 'test2', 'namespace': 'test2', 'success': True},
]
}
mongo_db.database['pipelines'].delete_many.return_value = MagicMock()
mongo_db.mongo_db_repository.delete_many = AsyncMock(return_value=MagicMock())
await mongo_db.delete_pipelines_timestamps(input_data)
mongo_db.database['pipelines'].delete_many.assert_called_once_with(
mongo_db.mongo_db_repository.delete_many.assert_called_once_with(
'orchestrated_schedules',
{
'$or': [
{'schedule_name': 'test1', 'namespace': 'test1'},
{'schedule_name': 'test2', 'namespace': 'test2'},
]
}
},
{},
)
@@ -387,12 +368,12 @@ async def test_delete_pipelines_timestamps_failure(now_mock, mongo_db):
],
**metadata,
}
mongo_db.database['pipelines'].delete_many.side_effect = Exception('Error')
mongo_db.mongo_db_repository.delete_many.side_effect = Exception('Error')
try:
await mongo_db.delete_pipelines_timestamps(input_data)
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_DELETE_PIPELINES_ERROR',
message='Failed to delete pipelines timestamps: Error',
@@ -416,10 +397,12 @@ async def test_create_collection_with_ttl_index_success(mongo_db):
},
}
mongo_db.database.list_collection_names.return_value = [
'raw_scouter_pipeline_2',
'raw_scouter_pipeline_3',
]
mongo_db.mongo_db_repository.database.list_collection_names = MagicMock(
return_value=[
'raw_scouter_pipeline_2',
'raw_scouter_pipeline_3',
]
)
collection_1 = MagicMock(
list_indexes=MagicMock(
@@ -438,15 +421,17 @@ async def test_create_collection_with_ttl_index_success(mongo_db):
list_indexes=MagicMock(return_value=[{'key': 'inserted_at', 'expireAfterSeconds': 3600}])
)
mongo_db.database.__getitem__ = MagicMock(
mongo_db.mongo_db_repository.database.__getitem__ = MagicMock(
side_effect=[collection_1, collection_2, collection_3]
)
await mongo_db.create_collection_with_ttl_index(input_data)
mongo_db.database.list_collection_names.assert_called_once_with()
mongo_db.mongo_db_repository.database.list_collection_names.assert_called_once_with()
mongo_db.database.create_collection.assert_called_once_with('raw_scouter_pipeline')
mongo_db.mongo_db_repository.database.create_collection.assert_called_once_with(
'raw_scouter_pipeline'
)
collection_1.list_indexes.assert_called_once()
collection_1.create_index.assert_called_once_with(
@@ -466,15 +451,15 @@ async def test_create_collection_with_ttl_index_success(mongo_db):
async def test_create_collection_with_ttl_index_failure(mongo_db):
input_data = {**metadata, 'pipelines': {'scouter-pipeline': {'topic': 'raw_scouter_pipeline'}}}
mongo_db.database.list_collection_names.return_value = []
mongo_db.mongo_db_repository.database.list_collection_names.return_value = []
mongo_db.database.create_collection.side_effect = Exception('Error')
mongo_db.mongo_db_repository.database.create_collection.side_effect = Exception('Error')
try:
await mongo_db.create_collection_with_ttl_index(input_data)
except Exception as e:
assert str(e) == 'Error'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='MONGODB_CREATE_COLLECTION_ERROR',
message='Failed to create collection raw_scouter_pipeline with TTL index: Error',
@@ -489,18 +474,15 @@ async def test_create_collection_with_ttl_index_failure(mongo_db):
@mark.asyncio
async def test_load_latest_data_none_last_data_timestamp(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
}
]
mongo_db.mongo_db_repository.find = AsyncMock(
return_value=[
{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00+0000',
}
]
)
result = await mongo_db.load_latest_data(
{
@@ -511,77 +493,70 @@ async def test_load_latest_data_none_last_data_timestamp(mongo_db):
}
)
mongo_db.database.__getitem__.assert_called_once_with('test_collection')
mongo_db.mongo_db_repository.find.assert_called_once_with(
'test_collection',
{'level': 'ERROR'},
{'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
)
collection.find.assert_called_once_with({'level': 'ERROR'}, {'_id': 0})
assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00.000000+0000'}]
assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
@mark.asyncio
async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.return_value = [
{
'name': 'test1',
'value': 1,
'timestamp': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
),
}
]
mongo_db.mongo_db_repository.find = AsyncMock(
return_value=[
{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00+0000',
}
]
)
result = await mongo_db.load_latest_data(
{
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
'last_data_timestamp': '2023-01-01 12:00:00+0000',
'base_data_filter': {'level': 'ERROR'},
}
)
mongo_db.database.__getitem__.assert_called_once_with('test_collection')
collection.find.assert_called_once_with(
mongo_db.mongo_db_repository.find.assert_called_once_with(
'test_collection',
{
'level': 'ERROR',
'timestamp': {
'$gt': datetime.strptime(
'2023-01-01 12:00:00.000000+0000', DATETIME_FORMAT_MS_WITH_TZ
)
'$gt': datetime.strptime('2023-01-01 12:00:00+0000', DATETIME_FORMAT_WITH_TZ)
},
},
{'_id': 0},
{'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
)
assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00.000000+0000'}]
assert result == [{'name': 'test1', 'value': 1, 'timestamp': '2023-01-01 12:00:00+0000'}]
@mark.asyncio
async def test_load_latest_data_error(mongo_db):
"""Test load_latest_data"""
collection = MagicMock()
mongo_db.send_notification = MagicMock()
mongo_db.database.__getitem__.return_value = collection
collection.find.side_effect = Exception('test')
mongo_db.mongo_db_repository.find.side_effect = Exception('test')
try:
await mongo_db.load_latest_data(
{
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': '2023-01-01 12:00:00.000000+0000',
'last_data_timestamp': '2023-01-01 12:00:00+0000',
'base_data_filter': {'level': 'ERROR'},
}
)
except Exception as e:
assert str(e) == 'test'
mongo_db.send_notification.assert_called_once_with(
mongo_db.send_notification_async.assert_called_once_with(
metadata={'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
notification_id='MONGO_LOAD_ERROR',
message='Error loading data from MongoDB: test',

125
tests/orchestrator/activities/test_slot_manager.py
View File

@@ -1,5 +1,5 @@
from datetime import timedelta
from unittest.mock import ANY, MagicMock, call, patch
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
from pandas import DataFrame
from pytest import fixture, mark
@@ -19,7 +19,7 @@ metadata = {
@fixture
@patch('orchestrator.activities.slot_manager.Redis.__init__')
@patch('orchestrator.activities.slot_manager.RedisRepository')
def slot_manager(_redis_mock):
slot_manager = SlotManager(
host='localhost',
@@ -28,31 +28,36 @@ def slot_manager(_redis_mock):
password='password',
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
slot_manager.redis_client = MagicMock()
slot_manager.redis_repository = MagicMock()
slot_manager.logger = MagicMock()
slot_manager.notification_handler = MagicMock()
slot_manager.send_notification = MagicMock()
slot_manager.send_notification_async = AsyncMock()
slot_manager.emit_metric = AsyncMock()
return slot_manager
@mark.asyncio
async def test_load_opc_slots_no_slot_keys(slot_manager):
slot_manager.redis_client.keys.return_value = []
slot_manager.redis_repository.keys = AsyncMock(return_value=[])
assert await slot_manager.load_opc_slots(metadata) == {}
@mark.asyncio
async def test_load_opc_slots(slot_manager):
slot_manager.redis_client.keys.return_value = [
b'slot:opc_tags:1',
b'slot:opc_tags:2',
b'slot:opc_tags:3',
]
slot_manager.redis_repository.keys = AsyncMock(
return_value=[
b'slot:opc_tags:1',
b'slot:opc_tags:2',
b'slot:opc_tags:3',
]
)
slot_manager.get = MagicMock(side_effect=['value1', 'value2', None])
slot_manager.redis_repository.get = AsyncMock(side_effect=['value1', 'value2', None])
response = await slot_manager.load_opc_slots(metadata)
@@ -65,13 +70,15 @@ async def test_load_opc_slots(slot_manager):
@mark.asyncio
async def test_load_opc_slots_no_decode(slot_manager):
slot_manager.redis_client.keys.return_value = [
'slot:opc_tags:1',
'slot:opc_tags:2',
'slot:opc_tags:3',
]
slot_manager.redis_repository.keys = AsyncMock(
return_value=[
'slot:opc_tags:1',
'slot:opc_tags:2',
'slot:opc_tags:3',
]
)
slot_manager.get = MagicMock(side_effect=['value1', 'value2', None])
slot_manager.redis_repository.get = AsyncMock(side_effect=['value1', 'value2', None])
response = await slot_manager.load_opc_slots(metadata)
@@ -84,19 +91,21 @@ async def test_load_opc_slots_no_decode(slot_manager):
@mark.asyncio
async def test_load_opc_slots_error(slot_manager):
slot_manager.redis_client.keys.return_value = [
'slot:opc_tags:1',
'slot:opc_tags:2',
'slot:opc_tags:3',
]
slot_manager.redis_repository.keys = AsyncMock(
return_value=[
'slot:opc_tags:1',
'slot:opc_tags:2',
'slot:opc_tags:3',
]
)
slot_manager.get = MagicMock(side_effect=Exception('Test exception'))
slot_manager.redis_repository.get = AsyncMock(side_effect=Exception('Test exception'))
try:
await slot_manager.load_opc_slots(metadata)
except Exception as e:
assert str(e) == 'Test exception'
slot_manager.send_notification.assert_called_once_with(
slot_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='REDIS_GET_ERROR',
message='Failed to load OPC slots: Test exception',
@@ -111,11 +120,13 @@ async def test_load_opc_slots_error(slot_manager):
@mark.asyncio
async def test_load_active_ingestors(slot_manager):
slot_manager.redis_client.keys.return_value = [
b'heartbeat:ingestor:1',
b'heartbeat:ingestor:2',
'heartbeat:ingestor:3',
]
slot_manager.redis_repository.keys = AsyncMock(
return_value=[
b'heartbeat:ingestor:1',
b'heartbeat:ingestor:2',
'heartbeat:ingestor:3',
]
)
response = await slot_manager.load_active_ingestors(metadata)
@@ -124,19 +135,21 @@ async def test_load_active_ingestors(slot_manager):
@mark.asyncio
async def test_load_active_ingestors_error(slot_manager):
slot_manager.redis_client.keys.return_value = [
'heartbeat:ingestor:1',
'heartbeat:ingestor:2',
'heartbeat:ingestor:3',
]
slot_manager.redis_repository.keys = AsyncMock(
return_value=[
'heartbeat:ingestor:1',
'heartbeat:ingestor:2',
'heartbeat:ingestor:3',
]
)
slot_manager.redis_client.keys.side_effect = Exception('Test exception')
slot_manager.redis_repository.keys = AsyncMock(side_effect=Exception('Test exception'))
try:
await slot_manager.load_active_ingestors(metadata)
except Exception as e:
assert str(e) == 'Test exception'
slot_manager.send_notification.assert_called_once_with(
slot_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='REDIS_GET_ERROR',
message='Failed to load active ingestors: Test exception',
@@ -151,11 +164,11 @@ async def test_load_active_ingestors_error(slot_manager):
@mark.asyncio
async def test_update_slots(slot_manager):
slot_manager.set = MagicMock(side_effect=[None, Exception('Test exception')])
slot_manager.redis_repository.set = AsyncMock(side_effect=[None, Exception('Test exception')])
response = await slot_manager.update_slots({'to_insert': {'1': 'value1', '2': 'value2'}})
slot_manager.set.assert_has_calls(
slot_manager.redis_repository.set.assert_has_calls(
[call('slot:opc_tags:1', 'value1', ttl=None), call('slot:opc_tags:2', 'value2', ttl=None)]
)
@@ -167,11 +180,13 @@ async def test_update_slots(slot_manager):
@mark.asyncio
async def test_delete_slots(slot_manager):
slot_manager.redis_client.delete = MagicMock(side_effect=[None, Exception('Test exception')])
slot_manager.redis_repository.delete = AsyncMock(
side_effect=[None, Exception('Test exception')]
)
response = await slot_manager.delete_slots({'to_delete': ['1', '2']})
slot_manager.redis_client.delete.assert_has_calls(
slot_manager.redis_repository.delete.assert_has_calls(
[call('slot:opc_tags:1'), call('slot:opc_tags:2')]
)
@@ -191,7 +206,7 @@ async def test_get_last_data_timestamp_none(slot_manager):
'mail_type': 'test_mail_type',
}
slot_manager.get = MagicMock(return_value=None)
slot_manager.redis_repository.get = AsyncMock(return_value=None)
result = await slot_manager.get_last_data_timestamp(test_data)
@@ -208,11 +223,13 @@ async def test_get_last_data_timestamp_not_none(slot_manager):
'mail_type': 'test_mail_type',
}
slot_manager.get = MagicMock(return_value='2023-01-01 12:00:00')
slot_manager.redis_repository.get = AsyncMock(return_value='2023-01-01 12:00:00')
result = await slot_manager.get_last_data_timestamp(test_data)
slot_manager.get.assert_called_once_with('notification_last_timestamp:test_mail_type')
slot_manager.redis_repository.get.assert_called_once_with(
'notification_last_timestamp:test_mail_type'
)
assert result == '2023-01-01 12:00:00'
@@ -227,8 +244,8 @@ async def test_get_last_data_timestamp_error(slot_manager):
'mail_type': 'test_mail_type',
}
slot_manager.send_notification = MagicMock()
slot_manager.get = MagicMock(side_effect=Exception('test'))
slot_manager.send_notification_async = AsyncMock()
slot_manager.redis_repository.get = AsyncMock(side_effect=Exception('test'))
try:
await slot_manager.get_last_data_timestamp(test_data)
@@ -236,7 +253,7 @@ async def test_get_last_data_timestamp_error(slot_manager):
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
slot_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='REDIS_GET_ERROR',
message='Error getting last data timestamp: test',
@@ -288,13 +305,13 @@ async def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
'mail_type': 'test_mail_type',
}
slot_manager.set = MagicMock()
slot_manager.redis_repository.set = AsyncMock()
result = await slot_manager.put_last_data_timestamp(test_data)
assert result == '2023-01-01 12:00:01'
slot_manager.set.assert_called_once_with(
slot_manager.redis_repository.set.assert_called_once_with(
'notification_last_timestamp:test_mail_type', '2023-01-01 12:00:01', ttl=18000
)
@@ -316,8 +333,8 @@ async def test_put_last_data_timestamp_error(slot_manager):
'mail_type': 'test_mail_type',
}
slot_manager.send_notification = MagicMock()
slot_manager.set = MagicMock(side_effect=Exception('test'))
slot_manager.send_notification_async = AsyncMock()
slot_manager.redis_repository.set = AsyncMock(side_effect=Exception('test'))
try:
await slot_manager.put_last_data_timestamp(test_data)
@@ -325,7 +342,7 @@ async def test_put_last_data_timestamp_error(slot_manager):
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
slot_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='REDIS_SET_ERROR',
message='Error setting last data timestamp: test',
@@ -340,7 +357,7 @@ async def test_put_last_data_timestamp_error(slot_manager):
@mark.asyncio
async def test_filter_notification_alerts(slot_manager):
slot_manager.get = MagicMock(
slot_manager.redis_repository.get = AsyncMock(
side_effect=[
None,
(now() - timedelta(seconds=600)).strftime(DATETIME_FORMAT_MS_WITH_TZ),
@@ -402,8 +419,10 @@ async def test_store_notification_cache(slot_manager):
'sent_ttl': 600,
}
slot_manager.set = MagicMock()
slot_manager.redis_repository.set = AsyncMock()
await slot_manager.store_notification_cache(test_data)
slot_manager.set.assert_called_once_with('test_schedule_1:test_notification_id_1', ANY, ttl=600)
slot_manager.redis_repository.set.assert_called_once_with(
'test_schedule_1:test_notification_id_1', ANY, ttl=600
)

5
tests/orchestrator/activities/test_temporal_manager.py
View File

@@ -26,10 +26,13 @@ def temporal_manager(connect_mock):
laborious_namespace='laborious',
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
temporal_manager.temporal_clients['scouter'] = MagicMock()
temporal_manager.temporal_clients['laborious'] = MagicMock()
temporal_manager.send_notification_async = AsyncMock()
temporal_manager.emit_metric = AsyncMock()
temporal_manager.send_notification = MagicMock()
return temporal_manager
@@ -100,7 +103,7 @@ async def test_normalize_schedules_error(temporal_manager):
await temporal_manager.normalize_schedules(metadata)
except Exception as e:
assert str(e) == 'Test exception'
temporal_manager.send_notification.assert_called_once_with(
temporal_manager.send_notification_async.assert_called_once_with(
metadata=metadata['metadata'],
notification_id='SCHEDULER_NORMALIZE_SCHEDULES_ERROR',
message='Failed to normalize schedules: Test exception',

11
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images.
pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion.
tag: "0.4.9"
tag: "0.5.0"
# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
imagePullSecrets:
@@ -151,7 +151,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "fix/SIENTIAPDE-1314-ajustes-nas-camadas-de-monitoramento-do-sientia"
value: "feature/SIENTIAPDE-1325-adicionar-metricas-especificas-de-operacoes-externas"
- name: PYTHON_APP
value: "orchestrator.worker.worker"
@@ -171,13 +171,6 @@ env:
name: redis
key: redis-password
- name: COUCHBASE_CONNECTION_STRING
value: "couchbase://sientia.couchbase.svc.cluster.local"
- name: COUCHBASE_USERNAME
value: "sientia"
- name: COUCHBASE_PASSWORD
value: "sientia"
- name: MONGODB_USERNAME
value: "root"
- name: MONGODB_PASSWORD