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

Merge pull request #12 from Aignosi/SIENTIAPDE-1172-criar-pipeline-de-alertas-orquestrador

Browse Source 
Sientiapde 1172 criar pipeline de alertas orquestrador
This commit is contained in:
Bruno Domingues
2025-08-04 10:00:42 -03:00
committed by GitHub
parent b0171614aa 3a8eb04462
commit 1b7603ae70
30 changed files with 3408 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

267
email.html Normal file
View File

@@ -0,0 +1,267 @@
<!DOCTYPE html>
<html lang="pt-BR">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIENTIA™ Report</title>
<style>
body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
h1, h2 { color: #333; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #f4f4f4; }
</style>
</head>
<body>
<h1>SIENTIA™ Alerts</h1>
<h3>Errors detected:</h3>
<h4>Model: <span>ipsum et</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node34:tag_8_LISTENNING_STOPPED</td>
<td>adipiscing_eiusmod_do</td>
<td>dolor_labore_consectetur_ipsum</td>
<td>2025-07-29 14:39:52.952316+00:00</td>
<td>sed elit dolore incididunt incididunt aliqua</td>
</tr>
</tbody>
</table>
<h4>Model: <span>incididunt et do</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_LISTENNING_STOPPED__server_5</td>
<td>eiusmod_adipiscing_dolore_ipsum_incididunt_incididunt</td>
<td>aliqua_dolore</td>
<td>2025-07-29 14:39:52.952579+00:00</td>
<td>dolor elit do ipsum consectetur amet ut do amet et</td>
</tr>
</tbody>
</table>
<h4>Model: <span>magna adipiscing aliqua</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>do_dolore_ut_amet</td>
<td>et_eiusmod</td>
<td>2025-07-29 14:39:52.952688+00:00</td>
<td>incididunt lorem eiusmod do et ipsum et ut</td>
</tr>
</tbody>
</table>
<h3>Warnings detected:</h3>
<h4>Model: <span>adipiscing magna lorem</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>ipsum_et_incididunt_do</td>
<td>ut_sed_incididunt</td>
<td>2025-07-29 14:39:52.952121+00:00</td>
<td>aliqua tempor aliqua sit ipsum amet elit ipsum</td>
</tr>
</tbody>
</table>
<h4>Model: <span>sed amet adipiscing incididunt</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>ALIQUA_SIT_INCIDIDUNT_EIUSMOD</td>
<td>consectetur_dolore_sit_sed_sed</td>
<td>dolore_ipsum</td>
<td>2025-07-29 14:39:52.952259+00:00</td>
<td>do aliqua ut incididunt consectetur consectetur sed et labore</td>
</tr>
</tbody>
</table>
<h4>Model: <span>ipsum ipsum amet tempor</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_LISTENNING_STOPPED__server_6</td>
<td>sit_labore_sed_dolor_et_elit</td>
<td>dolor_do_magna_et</td>
<td>2025-07-29 14:39:52.952376+00:00</td>
<td>magna consectetur do et dolor aliqua</td>
</tr>
</tbody>
</table>
<h4>Model: <span>sed ut</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>elit_et_ipsum_dolore</td>
<td>sed_ipsum</td>
<td>2025-07-29 14:39:52.952426+00:00</td>
<td>elit dolor dolore dolor eiusmod</td>
</tr>
</tbody>
</table>
<h4>Model: <span>eiusmod labore</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_CONNECTION_RETRY__server_3</td>
<td>dolor_ut_dolor_sit_adipiscing_incididunt</td>
<td>sit_tempor_dolore</td>
<td>2025-07-29 14:39:52.952483+00:00</td>
<td>ipsum consectetur magna elit dolore</td>
</tr>
</tbody>
</table>
<h4>Model: <span>do dolor tempor amet</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>sed_sit_do_tempor_ut</td>
<td>eiusmod_adipiscing</td>
<td>2025-07-29 14:39:52.952637+00:00</td>
<td>incididunt do do adipiscing dolore ut ut elit sit labore</td>
</tr>
</tbody>
</table>
<h3>Infos detected:</h3>
<h4>Model: <span>sed consectetur ut</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TEMPOR_DO_DOLOR_TEMPOR_DOLOR</td>
<td>elit_tempor_dolore_consectetur_dolore_adipiscing</td>
<td>magna_labore_lorem</td>
<td>2025-07-29 14:39:52.952199+00:00</td>
<td>et et adipiscing lorem magna eiusmod labore do</td>
</tr>
</tbody>
</table>
</body>
</html>

416
notification_generator.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

32
orchestrator/activities/activities.py
View File

@@ -1,26 +1,29 @@
from temporalio import activity, workflow
from temporalio.client import Client
from temporalio import workflow
from orchestrator.activities.mongo_db import MongoDB
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.couchbase import Couchbase
from orchestrator.activities.temporal_manager import TemporalManager
from orchestrator.activities.slot_manager import SlotManager
from orchestrator.activities.formatters import Formatters
from orchestrator.activities.email import Email
from orchestrator.activities.mongo_db import MongoDB
from typing import Any
from logging import Logger
from sientia_do.temporal.activities.postgres import Postgres
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
class Activities( # Couchbase,
TemporalManager, SlotManager, Formatters, MongoDB):
TemporalManager, SlotManager, Formatters, MongoDB, Email,
Postgres):
def __init__(self,
temporal_config: dict[str, Any],
# couchbase_config: dict[str, Any],
redis_config: dict[str, Any],
mongodb_config: dict[str, Any],
email_config: dict[str, Any],
postgres_config: dict[str, Any],
logger: Logger,
notification_handler: NotificationHandler):
@@ -59,5 +62,24 @@ class Activities( # Couchbase,
logger=logger,
notification_handler=notification_handler)
Email.__init__(self,
sender_email=email_config['sender_email'],
sender_password=email_config['sender_password'],
smtp_server=email_config['smtp_server'],
smtp_port=email_config['smtp_port'],
logger=logger,
notification_handler=notification_handler)
Postgres.__init__(self,
host=postgres_config['host'],
port=postgres_config['port'],
user=postgres_config['user'],
password=postgres_config['password'],
dbname=postgres_config['dbname'],
min_connections=postgres_config['min_connections'],
max_connections=postgres_config['max_connections'],
logger=logger,
notification_handler=notification_handler)
def shutdown(self):
MongoDB.shutdown(self)

187
orchestrator/activities/email.py Normal file
View File

@@ -0,0 +1,187 @@
from smtplib import SMTPServerDisconnected
from temporalio import workflow, activity
with workflow.unsafe.imports_passed_through():
import traceback
import smtplib
from typing import Any
from sientia_do.temporal.activities.base import BaseActivity
from sientia_do.temporal.utils.logger import Logger
from sientia_do.notifications.handlers import NotificationHandler
from orchestrator.utils.email_builder import EmailBuilder
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders
class Email(BaseActivity):
def __init__(self, sender_email: str, sender_password: str,
smtp_server: str, smtp_port: int,
logger: Logger, notification_handler: NotificationHandler):
self.email_builder = EmailBuilder(logger=logger)
self.sender_email = sender_email
self.sender_password = sender_password
self.smtp_port = smtp_port
self.smtp_server = smtp_server
logger.info(f"Initializing Email with {smtp_server}:{smtp_port}")
self.server = smtplib.SMTP(smtp_server, smtp_port, timeout=20)
if self.sender_password:
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
BaseActivity.__init__(self,
logger=logger,
notification_handler=notification_handler)
@activity.defn(name="build_email_html")
async def build_email_html(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Builds the email html for each receiver group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']
mail_type = input_data['mail_type']
self.info(f"Building email html for {mail_type} mail type.",
metadata=metadata)
for group_name, group_config in receiver_groups.items():
html = self.email_builder.build_email(
group_config['notifications'], mail_type)
group_config['html'] = html
self.info(f"Email html built for {mail_type} mail type.",
metadata=metadata)
return receiver_groups
def handle_attachments(self, attachments: list[dict], msg: MIMEMultipart) -> MIMEMultipart:
"""
Attaches a list of attachments to an email message.
Args:
attachments (List[Dict]): A list of dictionaries where each dictionary contains
the keys 'filename' and 'content' representing the attachment details.
msg (MIMEMultipart): The email message object to which the attachments will be added.
Returns:
MIMEMultipart: The email message object with the attachments added.
Raises:
Exception: If an attachment cannot be added, an error is logged.
"""
for attachment in attachments:
att_name = attachment['filename']
try:
# Create the attachment as a MIMEBase object
part = MIMEBase('application', 'octet-stream')
part.set_payload(
attachment['attachment_content'].encode('utf-8'))
encoders.encode_base64(part)
part.add_header(
'Content-Disposition',
f'attachment; filename="{att_name}"'
)
msg.attach(part)
except Exception as e:
self.logger.error(
f"Failed to attach content of {att_name}: {e}")
raise e
return msg
def try_send_email(self, msg: MIMEMultipart, receivers: str):
"""
Sends an email to the receivers.
"""
try:
self.server.sendmail(
self.sender_email, receivers, msg.as_string())
except SMTPServerDisconnected as e:
self.logger.error(f"SMTP server disconnected: {e}")
self.logger.info(
f"Reconnecting to {self.smtp_server}:{self.smtp_port}")
if self.server:
try:
self.server.quit()
except SMTPServerDisconnected as e:
self.logger.info(f"Server already disconnected: {e}")
except Exception as e:
self.logger.error(f"Failed to quit server: {e}")
raise e
self.server = smtplib.SMTP(
self.smtp_server, self.smtp_port, timeout=20)
if self.sender_password:
self.server.starttls()
self.server.login(self.sender_email, self.sender_password)
self.server.sendmail(self.sender_email, receivers, msg.as_string())
@activity.defn(name="send_email")
async def send_email(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Sends an email to the receivers of each group.
input_data:
- receiver_groups (dict): The receiver groups.
- mail_type (str): The mail type.
"""
metadata = input_data['metadata']
receiver_groups = input_data['receiver_groups']
mail_type = input_data['mail_type']
self.info(f"Sending email for {mail_type} mail type.",
metadata=metadata)
for group_name, group_config in receiver_groups.items():
try:
receivers = ", ".join(group_config['members'])
self.info(f"Sending email to {group_name}: {receivers}",
metadata=metadata)
msg = MIMEMultipart()
msg.attach(MIMEText(group_config['html'], 'html'))
msg['From'] = self.sender_email
msg['To'] = receivers
msg['Subject'] = f"SIENTIA™ {mail_type}"
msg = self.handle_attachments(
[
{
"filename": f"{notification['trigger']}_{notification['notification_id']}.txt",
"attachment_content": notification['attachment_content']
}
for notification in group_config['notifications']
if notification.get('attachment_content') is not None],
msg)
self.try_send_email(msg, receivers)
except Exception as e:
self.error(f"Failed to send email to {group_name}: {e}",
metadata=metadata)
traceback.print_exc()
group_config['status'] = 'failed'
else:
group_config['status'] = 'sent'
self.info(f"Email sent to {group_name}: {receivers}",
metadata=metadata)
self.info(f"Email sent for {mail_type} mail type.",
metadata=metadata)
return receiver_groups

87
orchestrator/activities/formatters.py
View File

@@ -1,4 +1,5 @@
from pandas import DataFrame
from temporalio import activity, workflow
from orchestrator.utils.orchestrator_functions import minimal_retrain
@@ -488,3 +489,89 @@ class Formatters(BaseActivity):
notification_id="REPORT_ORCHESTRATION_DELETED_SLOTS",
attachment=deleted_slots
)
@activity.defn(name="format_log_report")
async def format_log_report(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Formats the receiver_groups status to a dataframe to be stored in the database.
input_data:
- receiver_groups (dict): The receiver groups.
"""
metadata = input_data["metadata"]
mail_type = input_data["mail_type"]
self.info("Formatting log report...", metadata=metadata)
receiver_groups = input_data['receiver_groups']
data = {}
for group_name, group_config in receiver_groups.items():
for notification in group_config['notifications']:
notification_id = notification['notification_id']
trigger = notification['trigger']
key = f"{notification_id}:{trigger}"
if key not in data:
data[key] = {
'status': group_config['status'],
'timestamp': notification['timestamp'],
'groups': [group_name],
'message': notification['message'],
'level': notification['level'],
'notification_id': notification_id,
'block': notification['block'],
'schedule': trigger,
'pipeline': notification['pipeline'],
'project': notification['project'],
'model_name': notification['model_name'],
'model_id': notification['model_id'],
'mail_type': mail_type
}
else:
if group_name not in data[key]['groups']:
data[key]['groups'].append(group_name)
return DataFrame(list(data.values())).to_dict()
@activity.defn(name="filter_notification_reports")
async def filter_notification_reports(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Filter notification reports.
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']
sending_configs = input_data['sending_configs']
self.info("Filtering notification reports...", metadata=metadata)
receiver_groups = {}
for receiver_group in sending_configs:
group_name = receiver_group['group_name']
receiver_groups[group_name] = {
**receiver_group,
"notifications": []
}
receiver_groups[group_name]['notifications'] = []
already_added_keys = []
ignore_list = receiver_group.get('ignore', [])
for notification in notification_package:
alert_type = "reports"
notification_id = notification['notification_id']
key = f"{notification['trigger']}:{notification_id}"
# Check if this group must be notified
if alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys:
receiver_groups[group_name]["notifications"].append(
notification)
already_added_keys.append(key)
return receiver_groups

89
orchestrator/activities/mongo_db.py
View File

@@ -1,3 +1,4 @@
from pandas import DataFrame
from temporalio import workflow, activity
@@ -80,6 +81,15 @@ class MongoDB(BaseActivity):
"""
self.shutdown()
def find(self, collection_name: str, filters: dict[str, Any]) -> list[dict[str, Any]]:
collection = self.database[collection_name]
documents = list(collection.find(filters, {"_id": 0}))
documents = clear_mongo_id(documents)
return documents
@activity.defn(name="find_documents_in_mongodb",)
async def find_documents_in_mongodb(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
@@ -106,11 +116,7 @@ class MongoDB(BaseActivity):
f"Loading documents from collection '{collection_name}' with filters: {filters}", metadata=metadata)
try:
collection = self.database[collection_name]
documents = list(collection.find(filters, {"_id": 0}))
documents = clear_mongo_id(documents)
documents = self.find(collection_name, filters)
self.info(
f"Loaded {len(documents)} documents from collection '{collection_name}'", metadata=metadata)
@@ -352,3 +358,76 @@ class MongoDB(BaseActivity):
)
self.error(trace, metadata=metadata)
raise e
@activity.defn(name="load_latest_data")
async def load_latest_data(self, input_data: dict[str, Any]) -> list[dict[str, Any]]:
"""
Loads the latest data from MongoDB.
input_data:
- metadata (dict): The metadata of the workflow.
- collection_name (str): The name of the collection to load data from.
- last_data_timestamp (str): The timestamp of the last data to load.
- base_data_filter (dict): The base data filter to apply to the query.
returns:
- data (list[dict]): The data loaded from MongoDB.
"""
metadata = input_data['metadata']
collection_name = input_data['collection_name']
last_data_timestamp = input_data['last_data_timestamp']
base_data_filter = input_data['base_data_filter']
self.debug(
f"Loading data from MongoDB: {input_data}",
metadata=metadata
)
try:
if last_data_timestamp is None:
data_filter = base_data_filter
else:
data_filter = {
**base_data_filter,
"timestamp": {
"$gt": datetime.strptime(last_data_timestamp, DEFAULT_DATE_FORMAT)
}
}
self.debug(
f"Data filter: {data_filter}",
metadata=metadata
)
data = self.find(collection_name, data_filter)
self.debug(
f"Collected: {data}",
metadata=metadata
)
for item in data:
item['timestamp'] = item['timestamp'].strftime(
DEFAULT_DATE_FORMAT)
self.info(
f"Loaded {len(data)} documents from MongoDB",
metadata=metadata
)
self.debug(
f"Loaded data: {data}",
metadata=metadata
)
return data
except Exception as e:
trace = traceback.format_exc()
self.send_notification(
metadata=metadata,
notification_id="MONGO_LOAD_ERROR",
message=f"Error loading data from MongoDB: {e}",
block="load_latest_data",
level=NotificationLevel.ERROR,
attachment_content=trace
)
raise e

147
orchestrator/activities/slot_manager.py
View File

@@ -1,3 +1,4 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
@@ -8,6 +9,9 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.temporal.activities.redis_base import Redis
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
from pandas import DataFrame
from datetime import datetime, timedelta
class SlotManager(Redis):
@@ -193,3 +197,146 @@ class SlotManager(Redis):
f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}", metadata=metadata)
return report
@activity.defn(name="get_last_data_timestamp")
async def get_last_data_timestamp(self, input_data: dict[str, Any]) -> str | None:
"""
Gets the last data timestamp from redis.
"""
metadata = input_data['metadata']
key = f"notification_last_timestamp:{input_data['mail_type']}"
try:
data_hold = self.get(key)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_GET_ERROR",
message=f"Error getting last data timestamp: {e}",
block="get_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
self.debug(
f"Last collected timestamp: {data_hold}",
metadata=metadata
)
if not data_hold:
return None
return data_hold
@activity.defn(name="put_last_data_timestamp")
async def put_last_data_timestamp(self, input_data: dict[str, Any]):
"""
Puts the last data timestamp into redis.
"""
metadata = input_data['metadata']
key = f"notification_last_timestamp:{input_data['mail_type']}"
data = DataFrame(input_data['data'])
if data.empty:
self.warning("No data to insert",
metadata=metadata
)
return None
last_data_timestamp = data['timestamp'].max()
self.debug(
f"Last collected timestamp to insert: {last_data_timestamp}",
metadata=metadata
)
try:
self.set(key, last_data_timestamp, ttl=60*60*5)
except Exception as e:
self.send_notification(
metadata=metadata,
notification_id="REDIS_SET_ERROR",
message=f"Error setting last data timestamp: {e}",
block="put_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=traceback.format_exc()
)
raise e
return last_data_timestamp
@activity.defn(name="filter_notification_alerts")
async def filter_notification_alerts(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Filter notification alerts
"""
metadata = input_data['metadata']
notification_package = input_data['notification_package']
sending_configs = input_data['sending_configs']
notification_ttl = input_data['notification_ttl']
self.info("Filtering notification alerts...", metadata=metadata)
receiver_groups = {}
for receiver_group in sending_configs:
group_name = receiver_group['group_name']
receiver_groups[group_name] = {
**receiver_group,
"notifications": []
}
receiver_groups[group_name]['notifications'] = []
already_added_keys = []
ignore_list = receiver_group.get('ignore', [])
for notification in notification_package:
alert_type = "do_nothing"
notification_id = notification['notification_id']
# Check if notification was recently sent
key = f"{notification['trigger']}:{notification_id}"
last_sent = self.get(key)
if last_sent is None:
alert_type = "core_alerts"
else:
last_sent = datetime.strptime(
last_sent, DEFAULT_DATE_FORMAT)
# Check if "notification_ttl" seconds has passed since last sent
if (datetime.now() - last_sent) > timedelta(seconds=notification_ttl):
alert_type = "persistent_alerts"
# Check if this group must be notified
if alert_type in receiver_group['contents'] and notification_id not in ignore_list and key not in already_added_keys:
receiver_groups[group_name]["notifications"].append(
notification)
already_added_keys.append(key)
return receiver_groups
@activity.defn(name="store_notification_cache")
async def store_notification_cache(self, input_data: dict[str, Any]) -> None:
"""
Store notification cache
"""
metadata = input_data['metadata']
log_report = DataFrame(input_data['log_report'])
sent_ttl = input_data['sent_ttl']
self.info("Storing notification cache...", metadata=metadata)
now = datetime.now().strftime(DEFAULT_DATE_FORMAT)
for index, row in log_report.iterrows():
status = row['status']
if status == 'sent':
key = f"{row['schedule']}:{row['notification_id']}"
self.set(key, now, ttl=sent_ttl)
self.info("Notification cache stored...", metadata=metadata)

21
orchestrator/utils/connectors_config.py
View File

@@ -38,3 +38,24 @@ def build_temporal_config():
'temporal_scouter_namespace': getenv('TEMPORAL_SCOUTER_NAMESPACE', 'scouter'),
'temporal_laborious_namespace': getenv('TEMPORAL_LABORIOUS_NAMESPACE', 'laborious')
}
def build_postgres_config():
return {
'host': getenv('POSTGRES_HOST', 'localhost'),
'port': int(getenv('POSTGRES_PORT', '5432')),
'user': getenv('POSTGRES_USER', 'sientia'),
'password': getenv('POSTGRES_PASSWORD', 'sientia'),
'dbname': getenv('POSTGRES_DBNAME', 'sientia'),
'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')),
'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20'))
}
def build_email_config():
return {
'sender_email': getenv('EMAIL_SENDER', 'sientia-alerts@aignosi.com'),
'sender_password': getenv('EMAIL_SENDER_PASSWORD', 'sientia'),
'smtp_server': getenv('EMAIL_SMTP_SERVER', 'smtp.gmail.com'),
'smtp_port': int(getenv('EMAIL_SMTP_PORT', '587'))
}

76
orchestrator/utils/email_builder.py Normal file
View File

@@ -0,0 +1,76 @@
import json
from sientia_do.temporal.utils.logger import Logger
from sientia_do.notifications.models import NotificationLevel
from jinja2 import Template
import re
class EmailBuilder:
def __init__(self, logger: Logger):
self.logger = logger
self.report_template_file = './orchestrator/utils/templates/email_template.html'
self.general_template_file = './orchestrator/utils/templates/general_template.html'
with open(self.report_template_file, 'r') as file:
self.report_template = file.read()
with open(self.general_template_file, 'r') as file:
self.general_template = file.read()
def replace_parameters(self, template: str, parameters: dict) -> str:
# Criar um template Jinja2
template = Template(template)
return template.render(parameters)
def parameters(self, general_events: dict, mail_type: str) -> dict:
error_models = general_events.get('ERROR', {}).get('models', [])
warning_models = general_events.get('WARNING', {}).get('models', [])
info_models = general_events.get('INFO', {}).get('models', [])
return {
'mail_type': mail_type,
'error_events': self.replace_parameters(self.general_template,
general_events.get(
'ERROR')) if error_models else '',
'warning_events': self.replace_parameters(self.general_template,
general_events.get(
'WARNING')) if warning_models else '',
'info_events': self.replace_parameters(self.general_template,
general_events.get(
'INFO')) if info_models else '',
}
def build_email(self, report_data: list[dict], mail_type: str) -> str:
"""
Builds the email html.
"""
general_events = {}
for report in report_data:
level = report['level']
model_name = report['model_name']
if level not in general_events:
general_events[level] = {
'section_name': f'{level.capitalize()}s detected:',
'models': {}
}
if model_name not in general_events[level]['models']:
general_events[level]['models'][model_name] = {
'model_name': model_name,
'events': []
}
general_events[level]['models'][model_name]['events'].append(
report)
for _type, content in general_events.items():
content['models'] = list(content['models'].values())
return self.replace_parameters(
self.report_template, self.parameters(
general_events, mail_type
))

24
orchestrator/utils/templates/email_template.html Normal file
View File

@@ -0,0 +1,24 @@
<!DOCTYPE html>
<html lang="pt-BR">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>SIENTIA™ Report</title>
<style>
body { font-family: Arial, sans-serif; margin: 0; padding: 20px; }
h1, h2 { color: #333; }
table { width: 100%; border-collapse: collapse; margin-top: 20px; }
th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
th { background-color: #f4f4f4; }
</style>
</head>
<body>
<h1>SIENTIA™ {{ mail_type }}</h1>
{{ error_events }}
{{ warning_events }}
{{ info_events }}
{{ special_events }}
</body>
</html>

26
orchestrator/utils/templates/general_template.html Normal file
View File

@@ -0,0 +1,26 @@
<h3>{{ section_name }}</h3>
{% for model in models %}
<h4>Model: <span>{{ model.model_name }}</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
{% for event in model.events %}
<tr>
<td>{{ event.notification_id }}</td>
<td>{{ event.trigger }}</td>
<td>{{ event.block }}</td>
<td>{{ event.timestamp }}</td>
<td>{{ event.message }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endfor %}

57
orchestrator/worker/worker.py
View File

@@ -1,18 +1,24 @@
from temporalio import workflow, client
from temporalio.worker import Worker
from orchestrator.utils.connectors_config import build_temporal_config
with workflow.unsafe.imports_passed_through():
import os
import sys
import asyncio
from orchestrator.workflows.alerts import Alerts
from orchestrator.workflows.reports import Reports
from orchestrator.workflows.subworkflows.load_notification_package import LoadNotificationPackage
from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications
from orchestrator.workflows.orchestrator import Orchestrator
from orchestrator.activities.activities import Activities
from orchestrator.utils.connectors_config import (
# build_couchbase_config,
build_redis_config,
build_mongodb_config
build_mongodb_config,
build_temporal_config,
build_email_config,
build_postgres_config
)
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.temporal.utils.logger import get_logger
@@ -48,6 +54,8 @@ async def main():
temporal_config=build_temporal_config(),
redis_config=build_redis_config(),
mongodb_config=build_mongodb_config(),
email_config=build_email_config(),
postgres_config=build_postgres_config(),
logger=logger,
notification_handler=notification_handler
)
@@ -90,6 +98,51 @@ async def main():
activities.report_slot_orchestration,
activities.format_schedule_config,
]
),
Worker(
temporal_client,
task_queue='alerts-queue',
workflows=[Alerts, LoadNotificationPackage, ProcessNotifications],
activities=[
# Load notifications
activities.get_last_data_timestamp,
activities.find_documents_in_mongodb,
activities.load_latest_data,
activities.put_last_data_timestamp,
# Format and filter notifications
activities.filter_notification_alerts,
# Send email and export data to postgres
activities.build_email_html,
activities.send_email,
activities.format_log_report,
activities.export_data_to_postgres,
# Store notification cache
activities.store_notification_cache
]
),
Worker(
temporal_client,
task_queue='reports-queue',
workflows=[Reports, LoadNotificationPackage, ProcessNotifications],
activities=[
# Load notifications
activities.get_last_data_timestamp,
activities.find_documents_in_mongodb,
activities.load_latest_data,
activities.put_last_data_timestamp,
# Format and filter notifications
activities.filter_notification_reports,
# Send email and export data to postgres
activities.build_email_html,
activities.send_email,
activities.format_log_report,
activities.export_data_to_postgres
]
)
]

95
orchestrator/workflows/alerts.py Normal file
View File

@@ -0,0 +1,95 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from datetime import timedelta
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="alerts")
class Alerts:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Workflow to send alerts to the users
Args:
input_data (dict[str, Any]): Input data. It contains the following keys:
- schedule_name: str - Name of the schedule
- notification_ttl: int - Period before consider some notification persistent
- sent_ttl: int - Time to live for the sent notification
Returns:
None
Raises:
Exception: If the workflow fails
"""
metadata = {
'metadata': {
'schedule_name': input_data['schedule_name'],
'workflow_name': 'alerts',
'model_name': '-',
'model_id': '-'
}
}
mail_type = "Alerts"
input_data['metadata'] = metadata
input_data['mail_type'] = mail_type
input_data['base_data_filter'] = {
'level': 'ERROR'
}
# Call subworkflow "load_notification_package" passing the static filters
# (level = "ERROR" and timestamp > last timestamp)
package = await workflow.execute_child_workflow(
'load_notification_package',
input_data
)
if not package['notification_package'] or not package['sending_configs']:
return
# Filter notification package by groups custom configs, levels and
# timestamp cached
receiver_groups = await workflow.execute_local_activity_method(
Activities.filter_notification_alerts,
{
**metadata,
'notification_package': package['notification_package'],
'sending_configs': package['sending_configs'],
'notification_ttl': input_data['notification_ttl']
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Call subworkflow "process_notifications" passing the notification package
log_report = await workflow.execute_child_workflow(
'process_notifications',
{
'metadata': metadata,
'mail_type': mail_type,
'notification_package': receiver_groups,
'schema': 'sientia_data',
'table_name': 'log_report'
}
)
# Store the notification_id sendings to avoid sending them again
await workflow.execute_activity_method(
Activities.store_notification_cache,
{
**metadata,
'log_report': log_report,
'sent_ttl': input_data['sent_ttl']
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)

9
orchestrator/workflows/orchestrator.py
View File

@@ -11,6 +11,15 @@ with workflow.unsafe.imports_passed_through():
class Orchestrator:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Orchestrates the pipeline and slot management. Gets configuration from MongoDB and Redis,
creates the configuration and deploys the schedules and slots in the Temporal server and
Redis server.
input_data:
- schedule_name (str): The name of the schedule.
- pipelines_query (dict): The query to get the pipelines.
- opc_servers_query (dict): The query to get the OPC servers.
"""
input_data['workflow_name'] = 'orchestrator'

78
orchestrator/workflows/reports.py Normal file
View File

@@ -0,0 +1,78 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from datetime import timedelta
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="reports")
class Reports:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Workflow to send reports to the users
Args:
input_data (dict[str, Any]): Input data. It contains the following keys:
- schedule_name: str - Name of the schedule
Returns:
None
Raises:
Exception: If the workflow fails
"""
metadata = {
'metadata': {
'schedule_name': input_data['schedule_name'],
'workflow_name': 'reports',
'model_name': '-',
'model_id': '-'
}
}
mail_type = "Reports"
input_data['metadata'] = metadata
input_data['mail_type'] = mail_type
input_data['base_data_filter'] = {}
# Call subworkflow "load_notification_package" passing the static filters
# (timestamp > last timestamp)
package = await workflow.execute_child_workflow(
'load_notification_package',
input_data
)
if not package['notification_package'] or not package['sending_configs']:
return
# Filter notification package by groups custom configs, levels and
# timestamp cached
receiver_groups = await workflow.execute_local_activity_method(
Activities.filter_notification_reports,
{
**metadata,
'notification_package': package['notification_package'],
'sending_configs': package['sending_configs']
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Call subworkflow "process_notifications" passing the notification package
await workflow.execute_child_workflow(
'process_notifications',
{
'metadata': metadata,
'mail_type': mail_type,
'notification_package': receiver_groups,
'schema': 'sientia_data',
'table_name': 'log_report'
}
)

102
orchestrator/workflows/subworkflows/load_notification_package.py Normal file
View File

@@ -0,0 +1,102 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from sientia_do.temporal.utils.policies import retry_policy
from datetime import timedelta
@workflow.defn(name="load_notification_package")
class LoadNotificationPackage:
@workflow.run
async def run(self, input_data: dict[str, Any]):
"""
Loads the notification package from the MongoDB collection "notification_queue"
and the sending configs from the MongoDB collection "receiver_groups".
input_data:
- metadata (dict): The metadata of the workflow.
returns:
- last_timestamp (str): The last timestamp of the notification package.
- notification_package (list[dict]): The notification package.
- sending_configs (list[dict]): The sending configs.
- mail_type (str): The mail type.
"""
metadata = input_data['metadata']
# Load last timestamp from redis "notification_last_timestamp"
last_timestamp_handler = workflow.start_local_activity_method(
Activities.get_last_data_timestamp,
{
**metadata,
'mail_type': input_data['mail_type']
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# In parallel, load sending configs from collection "receiver_groups"
sending_configs_handler = workflow.start_local_activity_method(
Activities.find_documents_in_mongodb,
{
**metadata,
'query': {
'collection': 'receiver_groups',
'filters': {
'active': True
}
}
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
last_timestamp = await last_timestamp_handler
# Load notification package from collection "notification_queue", using a
# static filter
notification_package = await workflow.start_local_activity_method(
Activities.load_latest_data,
{
**metadata,
'collection_name': 'notification_queue',
'last_data_timestamp': last_timestamp,
'base_data_filter': input_data['base_data_filter']
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
sending_configs = await sending_configs_handler
if not sending_configs or not notification_package:
return {
'last_timestamp': last_timestamp,
'notification_package': notification_package,
'sending_configs': sending_configs
}
# Put last collected timestamp in redis "notification_last_timestamp"
await workflow.start_activity_method(
Activities.put_last_data_timestamp,
{
**metadata,
'data': notification_package,
'mail_type': input_data['mail_type']
},
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Return a dict with the following keys:
# - last_timestamp
# - notification_package
# - sending_configs
return {
'last_timestamp': last_timestamp,
'notification_package': notification_package,
'sending_configs': sending_configs
}

90
orchestrator/workflows/subworkflows/process_notifications.py Normal file
View File

@@ -0,0 +1,90 @@
from temporalio import workflow
with workflow.unsafe.imports_passed_through():
from orchestrator.activities.activities import Activities
from typing import Any
from datetime import timedelta
from sientia_do.temporal.utils.policies import retry_policy
@workflow.defn(name="process_notifications")
class ProcessNotifications:
@workflow.run
async def run(self, input_data: dict[str, Any]) -> dict[str, Any]:
"""
Processes the notifications. Builds the report html for each group and each model,
sends the report html to the receivers of each group, stores the sending log in the
postgres database "log_report", and returns the log report to the caller.
input_data:
- metadata (dict): The metadata of the workflow.
- mail_type (str): The mail type.
- schema (str): The schema of the table.
- table_name (str): The name of the table.
- notification_package (list[dict]): The notification package. the format of each
notification package is:
{
'group_name' (str)
'group_members' (list[str])
'notifications' (dict)
{
'model_name' (dict[str, list[dict]])
}
}
returns:
- log_report (dict)
"""
metadata = input_data["metadata"]
# Use notification package to create the report html for each group and each model
data_to_sent = await workflow.execute_local_activity_method(
Activities.build_email_html,
{
**metadata,
"receiver_groups": input_data["notification_package"],
"mail_type": input_data["mail_type"]
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Send the report html to the receivers of each group
log_report = await workflow.execute_activity_method(
Activities.send_email,
{
**metadata,
"receiver_groups": data_to_sent,
"mail_type": input_data["mail_type"]
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Format the log report to a dataframe to be stored in the database
log_report = await workflow.execute_local_activity_method(
Activities.format_log_report,
{
**metadata,
"receiver_groups": log_report,
"mail_type": input_data["mail_type"]
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Store sending log in postgres database "log_report"
await workflow.execute_activity_method(
Activities.export_data_to_postgres,
{
**metadata,
"schema": input_data["schema"],
"table_name": input_data["table_name"],
"data": log_report
},
schedule_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
# Return the log report to the caller
return log_report

3
requirements.txt
View File

@@ -4,4 +4,5 @@ sqlalchemy
redis
couchbase
pymongo
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.4
jinja2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.3.5

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

@@ -12,7 +12,12 @@ from orchestrator.activities.formatters import Formatters
@patch('orchestrator.activities.temporal_manager.TemporalManager.__init__')
@patch('orchestrator.activities.slot_manager.SlotManager.__init__')
@patch('orchestrator.activities.formatters.Formatters.__init__')
def test___init__(mock_formatters_init, mock_slot_manager_init,
@patch('orchestrator.activities.email.Email.__init__')
@patch('sientia_do.temporal.activities.postgres.Postgres.__init__')
def test___init__(mock_postgres_init,
mock_email_init,
mock_formatters_init,
mock_slot_manager_init,
mock_temporal_manager_init,
mock_mongodb_init):
@@ -35,6 +40,23 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
'temporal_laborious_namespace': 'laborious'
}
email_config = {
'sender_email': 'test@test.com',
'sender_password': 'test',
'smtp_server': 'test',
'smtp_port': 587
}
postgres_config = {
'host': 'localhost',
'port': 5432,
'user': 'admin',
'password': 'password',
'dbname': 'test_db',
'min_connections': 1,
'max_connections': 10,
}
logger = MagicMock()
notification_handler = MagicMock()
@@ -42,6 +64,8 @@ def test___init__(mock_formatters_init, mock_slot_manager_init,
temporal_config=temporal_config,
redis_config=redis_config,
mongodb_config=mongo_db_config,
email_config=email_config,
postgres_config=postgres_config,
logger=logger,
notification_handler=notification_handler
)

359
tests/orchestrator/activities/test_email.py Normal file
View File

@@ -0,0 +1,359 @@
from smtplib import SMTPServerDisconnected
from unittest.mock import MagicMock, call, patch
from pytest import mark, fixture
from orchestrator.activities.email import Email
@fixture
@patch('orchestrator.activities.email.EmailBuilder')
@patch('orchestrator.activities.email.smtplib')
def email(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password="test",
smtp_server="test",
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock()
)
email_builder.send_notification = MagicMock()
return email
@patch('orchestrator.activities.email.EmailBuilder')
@patch('orchestrator.activities.email.smtplib')
def test___init___with_password(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password="test",
smtp_server="test",
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock()
)
assert email.sender_email == "test@test.com"
assert email.sender_password == "test"
assert email.smtp_port == 587
smtplib.SMTP.assert_called_once_with("test", 587, timeout=20)
smtplib.SMTP.return_value.starttls.assert_called_once()
smtplib.SMTP.return_value.login.assert_called_once_with(
"test@test.com", "test")
assert email.server == smtplib.SMTP.return_value
@patch('orchestrator.activities.email.EmailBuilder')
@patch('orchestrator.activities.email.smtplib')
def test___init___without_password(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password=None,
smtp_server="test",
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock()
)
assert email.sender_email == "test@test.com"
assert email.sender_password is None
assert email.smtp_port == 587
smtplib.SMTP.assert_called_once_with("test", 587, timeout=20)
assert email.server == smtplib.SMTP.return_value
metadata = {
"metadata": {
"schedule_name": "test",
"model_name": "test",
"model_id": "test",
"workflow_name": "test",
}
}
@mark.asyncio
async def test_build_email_html(email):
email.email_builder.build_email = MagicMock(
return_value="test"
)
input_data = {
**metadata,
"receiver_groups": {
"group_1": {
"notifications": [
{
"type": "test",
"subject": "test",
"body": "test"
},
{
"type": "test",
"subject": "test",
"body": "test"
}
]
}
},
"mail_type": "test"
}
response = await email.build_email_html(input_data)
assert response == {
"group_1": {
"notifications": [
{
"type": "test",
"subject": "test",
"body": "test",
},
{
"type": "test",
"subject": "test",
"body": "test",
}
],
"html": "test"
}
}
email.email_builder.build_email.assert_called_once_with(
input_data['receiver_groups']['group_1']['notifications'],
input_data['mail_type']
)
@patch('orchestrator.activities.email.MIMEBase')
@patch('orchestrator.activities.email.encoders')
def test_handle_attachments_success(encoders, mime_base, email):
message = MagicMock()
attachments = [
{
"filename": "file_1",
"attachment_content": "test_content_1"
},
{
"filename": "file_2",
"attachment_content": "test_content_2"
},
{
"filename": "file_3",
"attachment_content": "test_content_3"
}
]
response = email.handle_attachments(attachments, message)
assert response == message
mime_base.assert_called_with('application', 'octet-stream')
assert mime_base.return_value.set_payload.call_count == 3
mime_base.return_value.set_payload.assert_has_calls(
[
call("test_content_1".encode('utf-8')),
call("test_content_2".encode('utf-8')),
call("test_content_3".encode('utf-8'))
]
)
encoders.encode_base64.assert_called_with(mime_base.return_value)
assert encoders.encode_base64.call_count == 3
mime_base.return_value.add_header.assert_has_calls(
[
call('Content-Disposition', 'attachment; filename="file_1"'),
call('Content-Disposition', 'attachment; filename="file_2"'),
call('Content-Disposition', 'attachment; filename="file_3"')
]
)
message.attach.assert_called_with(mime_base.return_value)
assert message.attach.call_count == 3
@patch('orchestrator.activities.email.MIMEBase')
def test_handle_attachments_failure(mime_base, email):
message = MagicMock()
mime_base.side_effect = Exception("test")
attachments = [
{
"filename": "file_1",
"attachment_content": "test_content_1"
}
]
try:
email.handle_attachments(attachments, message)
except Exception as e:
assert str(e) == "test"
assert message.attach.call_count == 0
def test_try_send_email_success(email):
email.server.sendmail = MagicMock()
msg = MagicMock()
email.try_send_email(msg, "test")
email.server.sendmail.assert_called_once_with(
"test@test.com", "test", msg.as_string.return_value)
@patch('orchestrator.activities.email.smtplib.SMTP')
def test_try_send_email_reconnect_quit_success(smtp, email):
email.server.sendmail = MagicMock(
side_effect=SMTPServerDisconnected("test")
)
email.server.quit = MagicMock()
msg = MagicMock()
email.try_send_email(msg, "test")
smtp.assert_has_calls([
call("test", 587, timeout=20)
])
smtp.return_value.starttls.assert_called_once()
smtp.return_value.login.assert_called_once_with(
"test@test.com", "test")
smtp.return_value.sendmail.assert_called_once_with(
"test@test.com", "test", msg.as_string.return_value)
@patch('orchestrator.activities.email.smtplib.SMTP')
def test_try_send_email_reconnect_quit_failure_disconnect(smtp, email):
email.server.sendmail = MagicMock(
side_effect=SMTPServerDisconnected("test")
)
email.server.quit = MagicMock(
side_effect=SMTPServerDisconnected("test")
)
msg = MagicMock()
email.try_send_email(msg, "test")
smtp.assert_has_calls([
call("test", 587, timeout=20)
])
smtp.return_value.starttls.assert_called_once()
smtp.return_value.login.assert_called_once_with(
"test@test.com", "test")
smtp.return_value.sendmail.assert_called_once_with(
"test@test.com", "test", msg.as_string.return_value)
@patch('orchestrator.activities.email.smtplib.SMTP')
def test_try_send_email_reconnect_quit_failure(smtp, email):
email.server.sendmail = MagicMock(
side_effect=SMTPServerDisconnected("test")
)
email.server.quit = MagicMock(
side_effect=Exception("test")
)
msg = MagicMock()
try:
email.try_send_email(msg, "test")
except Exception as e:
assert str(e) == "test"
else:
assert False, "Expected exception"
@mark.asyncio
@patch('orchestrator.activities.email.MIMEText')
@patch('orchestrator.activities.email.MIMEMultipart')
async def test_send_email(mimemultipart, mimetext, email):
side_effect_1 = MagicMock()
side_effect_2 = MagicMock()
mimemultipart.side_effect = [
side_effect_1,
side_effect_2
]
email.email_builder.send_notification = MagicMock()
email.try_send_email = MagicMock(
side_effect=[
None,
Exception("test")
]
)
input_data = {
**metadata,
"receiver_groups": {
"group_1": {
"members": ["test@test.com", "test2@test.com"],
"notifications": [
{
"attachment_content": "test_content_1",
"trigger": "test_trigger",
"notification_id": "test_notification_id"
}
],
"html": "test_html1"
},
"group_2": {
"members": ["test3@test.com", "test4@test.com"],
"notifications": [],
"html": "test_html2"
}
},
"mail_type": "test_TYPE"
}
response = await email.send_email(input_data)
assert response['group_1']['status'] == 'sent'
assert response['group_2']['status'] == 'failed'
assert mimemultipart.call_count == 2
mimetext.assert_has_calls(
[
call("test_html1", "html"),
call("test_html2", "html")
]
)
side_effect_1.__setitem__.assert_has_calls(
[
call('From', 'test@test.com'),
call('To', 'test@test.com, test2@test.com'),
call('Subject', 'SIENTIA™ test_TYPE')
]
)
side_effect_2.__setitem__.assert_has_calls(
[
call('From', 'test@test.com'),
call('To', 'test3@test.com, test4@test.com'),
call('Subject', 'SIENTIA™ test_TYPE')
]
)
email.try_send_email.assert_has_calls(
[
call(side_effect_1, 'test@test.com, test2@test.com'),
call(side_effect_2, 'test3@test.com, test4@test.com')
]
)
assert email.try_send_email.call_count == 2

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

@@ -1,5 +1,6 @@
from unittest.mock import MagicMock, patch, call, ANY
import json
from pandas import DataFrame
from pytest import fixture, mark
from sientia_do.notifications.models import NotificationLevel
from orchestrator.activities.formatters import Formatters
@@ -696,3 +697,129 @@ async def test_report_slot_orchestration(formatters):
attachment=input_data['deleted_slots']
)
])
@mark.asyncio
async def test_format_log_report(formatters):
input_data = {
**metadata,
"receiver_groups": {
"test_receiver_group": {
"notifications": [
{
"notification_id": "test_notification_id",
"trigger": "test_trigger",
"timestamp": "2021-01-01",
"message": "test_message",
"level": "test_level",
"block": "test_block",
"pipeline": "test_pipeline",
"project": "test_project",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
],
"status": "sent",
},
"test_receiver_group2": {
"notifications": [
{
"notification_id": "test_notification_id",
"trigger": "test_trigger",
"timestamp": "2021-01-01",
"message": "test_message",
"level": "test_level",
"block": "test_block",
"pipeline": "test_pipeline",
"project": "test_project",
"model_name": "test_model_name",
"model_id": "test_model_id"
}
],
"status": "sent",
}
},
"mail_type": "test_mail_type"
}
result = await formatters.format_log_report(input_data)
expected_result = DataFrame(
[
{
"status": "sent",
"timestamp": "2021-01-01",
"groups": ["test_receiver_group", "test_receiver_group2"],
"message": "test_message",
"level": "test_level",
"notification_id": "test_notification_id",
"block": "test_block",
"schedule": "test_trigger",
"pipeline": "test_pipeline",
"project": "test_project",
"model_name": "test_model_name",
"model_id": "test_model_id",
"mail_type": "test_mail_type"
}
]
)
assert DataFrame(result).equals(expected_result)
@mark.asyncio
async def test_filter_notification_reports(formatters):
input_data = {
**metadata,
'notification_package': [
{
'trigger': 'test_trigger_1',
'notification_id': 'test_notification_id_1'
},
{
'trigger': 'test_trigger_2',
'notification_id': 'test_notification_id_2'
},
{
'trigger': 'test_trigger_3',
'notification_id': 'test_notification_id_3'
}
],
'sending_configs': [
{
'group_name': 'test_group_1',
'contents': ['reports'],
'ignore': ['test_notification_id_1']
},
{
'group_name': 'test_group_2',
'contents': ['core_alerts']
}
]
}
response = await formatters.filter_notification_reports(input_data)
assert response == {
'test_group_1': {
'group_name': 'test_group_1',
'contents': ['reports'],
'ignore': ['test_notification_id_1'],
'notifications': [
{
'trigger': 'test_trigger_2',
'notification_id': 'test_notification_id_2'
},
{
'trigger': 'test_trigger_3',
'notification_id': 'test_notification_id_3'
}
]
},
'test_group_2': {
'group_name': 'test_group_2',
'contents': ['core_alerts'],
'notifications': []
}
}

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

@@ -1,4 +1,5 @@
from curses import meta
from datetime import datetime
from unittest.mock import MagicMock, patch, ANY
from pytest import fixture, mark
from orchestrator.activities.mongo_db import clear_mongo_id
@@ -518,3 +519,121 @@ async def test_create_collection_with_ttl_index_failure(mongo_db):
)
else:
assert False, "Expected an exception to be raised"
@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', '%Y-%m-%d %H:%M:%S.%f')
}
]
result = await mongo_db.load_latest_data({
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': None,
'base_data_filter': {
'level': 'ERROR'
}
})
mongo_db.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{
'level': 'ERROR'
},
{"_id": 0}
)
assert result == [{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00.000000'
}]
@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', '%Y-%m-%d %H:%M:%S.%f')
}
]
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',
'base_data_filter': {
'level': 'ERROR'
}
})
mongo_db.database.__getitem__.assert_called_once_with(
'test_collection')
collection.find.assert_called_once_with(
{
'level': 'ERROR',
'timestamp': {
'$gt': datetime.strptime(
'2023-01-01 12:00:00.000000', '%Y-%m-%d %H:%M:%S.%f')
}
},
{"_id": 0}
)
assert result == [{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00.000000'
}]
@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')
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',
'base_data_filter': {
'level': 'ERROR'
}
})
except Exception as e:
assert str(e) == 'test'
mongo_db.send_notification.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',
block='load_latest_data',
level=NotificationLevel.ERROR,
attachment_content=ANY
)

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

@@ -1,7 +1,10 @@
from unittest.mock import MagicMock, patch, call, ANY
from datetime import datetime, timedelta
from pandas import DataFrame
from pytest import mark, fixture
from orchestrator.activities.slot_manager import SlotManager
from sientia_do.notifications.models import NotificationLevel
from orchestrator.utils.patterns import DEFAULT_DATE_FORMAT
metadata = {
"metadata": {
@@ -206,3 +209,257 @@ async def test_delete_slots(slot_manager):
"message": "Test exception"
}
}
@mark.asyncio
async def test_get_last_data_timestamp_none(slot_manager):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'mail_type': 'test_mail_type'
}
slot_manager.get = MagicMock(return_value=None)
result = await slot_manager.get_last_data_timestamp(test_data)
assert result is None
@mark.asyncio
async def test_get_last_data_timestamp_not_none(slot_manager):
"""Test get_last_data_timestamp"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'mail_type': 'test_mail_type'
}
slot_manager.get = MagicMock(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'
)
assert result == '2023-01-01 12:00:00'
@mark.asyncio
async def test_get_last_data_timestamp_error(slot_manager):
"""Test get_last_data_timestamp error"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'mail_type': 'test_mail_type'
}
slot_manager.send_notification = MagicMock()
slot_manager.get = MagicMock(side_effect=Exception('test'))
try:
await slot_manager.get_last_data_timestamp(test_data)
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_GET_ERROR",
message="Error getting last data timestamp: test",
block="get_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected exception"
@mark.asyncio
async def test_put_last_data_timestamp_empty_dataframe(slot_manager):
"""Test put_last_data_timestamp with empty dataframe"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': DataFrame(columns=['name', 'value', 'timestamp']).to_dict('records'),
'mail_type': 'test_mail_type'
}
slot_manager.set = MagicMock()
result = await slot_manager.put_last_data_timestamp(test_data)
assert result is None
slot_manager.set.assert_not_called()
@mark.asyncio
async def test_put_last_data_timestamp_not_empty_dataframe(slot_manager):
"""Test put_last_data_timestamp with not empty dataframe"""
data = DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'timestamp': ['2023-01-01 12:00:00', '2023-01-01 12:00:01']
})
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': data.to_dict('records'),
'mail_type': 'test_mail_type'
}
slot_manager.set = MagicMock()
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(
'notification_last_timestamp:test_mail_type',
'2023-01-01 12:00:01',
ttl=18000
)
@mark.asyncio
async def test_put_last_data_timestamp_error(slot_manager):
"""Test put_last_data_timestamp error"""
test_data = {
**metadata,
'workflow_name': 'test_pipeline',
'schedule_name': 'test_schedule',
'data': DataFrame({
'name': ['sensor1', 'sensor2'],
'value': [25.5, 30.0],
'timestamp': ['2023-01-01 12:00:00'] * 2
}).to_dict('records'),
'mail_type': 'test_mail_type'
}
slot_manager.send_notification = MagicMock()
slot_manager.set = MagicMock(side_effect=Exception('test'))
try:
await slot_manager.put_last_data_timestamp(test_data)
except Exception as e:
assert str(e) == 'test'
slot_manager.send_notification.assert_called_once_with(
metadata=metadata['metadata'],
notification_id="REDIS_SET_ERROR",
message="Error setting last data timestamp: test",
block="put_last_data_timestamp",
level=NotificationLevel.ERROR,
attachment_content=ANY
)
else:
assert False, "Expected exception"
@mark.asyncio
async def test_filter_notification_alerts(slot_manager):
slot_manager.get = MagicMock(side_effect=[
None,
(datetime.now() - timedelta(seconds=600)
).strftime(DEFAULT_DATE_FORMAT),
datetime.now().strftime(DEFAULT_DATE_FORMAT),
None,
(datetime.now() - timedelta(seconds=600)
).strftime(DEFAULT_DATE_FORMAT),
datetime.now().strftime(DEFAULT_DATE_FORMAT)])
input_data = {
**metadata,
'notification_package': [
{
'trigger': 'test_trigger_1',
'notification_id': 'test_notification_id_1'
},
{
'trigger': 'test_trigger_2',
'notification_id': 'test_notification_id_2'
},
{
'trigger': 'test_trigger_3',
'notification_id': 'test_notification_id_3'
}
],
'sending_configs': [
{
'group_name': 'test_group_1',
'contents': ['core_alerts', 'persistent_alerts']
},
{
'group_name': 'test_group_2',
'contents': ['core_alerts']
}
],
'notification_ttl': 300,
'mail_type': 'test_mail_type'
}
response = await slot_manager.filter_notification_alerts(input_data)
assert response == {
'test_group_1': {
'group_name': 'test_group_1',
'contents': ['core_alerts', 'persistent_alerts'],
'notifications': [
{
'trigger': 'test_trigger_1',
'notification_id': 'test_notification_id_1'
},
{
'trigger': 'test_trigger_2',
'notification_id': 'test_notification_id_2'
}
]
},
'test_group_2': {
'group_name': 'test_group_2',
'contents': ['core_alerts'],
'notifications': [
{
'trigger': 'test_trigger_1',
'notification_id': 'test_notification_id_1'
}
]
}
}
@mark.asyncio
async def test_store_notification_cache(slot_manager):
"""Test store_notification_cache"""
test_data = {
**metadata,
'log_report': DataFrame({
'status': ['sent', 'error'],
'schedule': ['test_schedule_1', 'test_schedule_2'],
'notification_id': ['test_notification_id_1', 'test_notification_id_2']
}).to_dict(),
'sent_ttl': 600
}
slot_manager.set = MagicMock()
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
)

71
tests/orchestrator/utils/test_connectors_config.py
View File

@@ -1,7 +1,10 @@
from os import environ
from orchestrator.utils.connectors_config import (build_redis_config,
build_couchbase_config,
build_mongodb_config, build_temporal_config)
build_mongodb_config,
build_temporal_config,
build_email_config,
build_postgres_config)
def test_build_redis_config_with_env_vars():
@@ -100,3 +103,69 @@ def test_build_temporal_config_with_defaults():
'temporal_scouter_namespace': 'scouter',
'temporal_laborious_namespace': 'laborious'
}
def test_build_email_config_with_env_vars():
environ['EMAIL_SENDER'] = 'test@test.com'
environ['EMAIL_SENDER_PASSWORD'] = 'test'
environ['EMAIL_SMTP_SERVER'] = 'test'
environ['EMAIL_SMTP_PORT'] = '587'
assert build_email_config() == {
'sender_email': 'test@test.com',
'sender_password': 'test',
'smtp_server': 'test',
'smtp_port': 587
}
def test_build_email_config_with_defaults():
environ.pop('EMAIL_SENDER', None)
environ.pop('EMAIL_SENDER_PASSWORD', None)
environ.pop('EMAIL_SMTP_SERVER', None)
environ.pop('EMAIL_SMTP_PORT', None)
assert build_email_config() == {
'sender_email': 'sientia-alerts@aignosi.com',
'sender_password': 'sientia',
'smtp_server': 'smtp.gmail.com',
'smtp_port': 587
}
def test_build_postgres_config_with_env_vars():
environ['POSTGRES_HOST'] = 'localhost'
environ['POSTGRES_PORT'] = '5432'
environ['POSTGRES_USER'] = 'sientia'
environ['POSTGRES_PASSWORD'] = 'sientia'
environ['POSTGRES_DBNAME'] = 'sientia'
environ['POSTGRES_MIN_CONNECTIONS'] = '5'
environ['POSTGRES_MAX_CONNECTIONS'] = '20'
assert build_postgres_config() == {
'host': 'localhost',
'port': 5432,
'user': 'sientia',
'password': 'sientia',
'dbname': 'sientia',
'min_connections': 5,
'max_connections': 20
}
def test_build_postgres_config_with_defaults():
environ.pop('POSTGRES_HOST', None)
environ.pop('POSTGRES_PORT', None)
environ.pop('POSTGRES_USER', None)
environ.pop('POSTGRES_PASSWORD', None)
environ.pop('POSTGRES_DBNAME', None)
environ.pop('POSTGRES_MIN_CONNECTIONS', None)
environ.pop('POSTGRES_MAX_CONNECTIONS', None)
assert build_postgres_config() == {
'host': 'localhost',
'port': 5432,
'user': 'sientia',
'password': 'sientia',
'dbname': 'sientia',
'min_connections': 5,
'max_connections': 20
}

145
tests/orchestrator/utils/test_email_builder.py Normal file
View File

@@ -0,0 +1,145 @@
import json
from unittest.mock import MagicMock, patch
from pytest import fixture
from orchestrator.utils.email_builder import EmailBuilder
@fixture
@patch('orchestrator.utils.email_builder.open')
def report_builder(open):
return EmailBuilder(MagicMock())
@patch('orchestrator.utils.email_builder.Template')
def test_replace_parameters(template, report_builder):
output = report_builder.replace_parameters('template', {'key': 'value'})
assert output == template.return_value.render.return_value
template.assert_called_once_with('template')
template.return_value.render.assert_called_once_with({'key': 'value'})
def test_parameters(report_builder):
report_builder.replace_parameters = MagicMock()
general_events = {
'ERROR': {
'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_1', 'level': 'ERROR', 'project': 'project'}]},
]
},
'WARNING': {
'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_2', 'level': 'WARNING', 'project': 'project'}]},
]
},
'INFO': {
'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_3', 'level': 'INFO', 'project': 'project'}]},
]
}
}
output = report_builder.parameters(general_events, 'model_name')
assert output == {
'mail_type': 'model_name',
'error_events': report_builder.replace_parameters.return_value,
'warning_events': report_builder.replace_parameters.return_value,
'info_events': report_builder.replace_parameters.return_value,
}
report_builder.replace_parameters.assert_any_call(
report_builder.general_template,
{'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_1', 'level': 'ERROR', 'project': 'project'}]},
]}
)
report_builder.replace_parameters.assert_any_call(
report_builder.general_template,
{'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_2', 'level': 'WARNING', 'project': 'project'}]},
]}
)
report_builder.replace_parameters.assert_any_call(
report_builder.general_template,
{'models': [
{'model_name': 'model_name', 'events': [
{'notification_id': 'ID_3', 'level': 'INFO', 'project': 'project'}]},
]}
)
def test_build_email(report_builder):
report_builder.parameters = MagicMock()
report_builder.replace_parameters = MagicMock()
report_data = [
{'notification_id': 'ID_1', 'level': 'ERROR',
'project': 'project', 'model_name': 'model_name'},
{'notification_id': 'ID_2', 'level': 'WARNING',
'project': 'project', 'model_name': 'model_name'},
{'notification_id': 'ID_2', 'level': 'INFO',
'project': 'project', 'model_name': 'model_name'},
{'notification_id': 'ID_3', 'level': 'ERROR',
'project': 'project', 'model_name': 'model_name'}
]
html = report_builder.build_email(report_data, 'type_1')
report_builder.replace_parameters.assert_called_once_with(
report_builder.report_template,
report_builder.parameters.return_value
)
assert html == report_builder.replace_parameters.return_value
report_builder.parameters.assert_called_once_with(
{
'ERROR': {
'section_name': 'Errors detected:',
'models': [
{
'model_name': 'model_name',
'events': [
{'notification_id': 'ID_1', 'level': 'ERROR',
'project': 'project', 'model_name': 'model_name'},
{'notification_id': 'ID_3', 'level': 'ERROR',
'project': 'project', 'model_name': 'model_name'}
]
}
]
},
'WARNING': {
'section_name': 'Warnings detected:',
'models': [
{
'model_name': 'model_name',
'events': [
{'notification_id': 'ID_2', 'level': 'WARNING',
'project': 'project', 'model_name': 'model_name'}
]
}
]
},
'INFO': {
'section_name': 'Infos detected:',
'models': [
{
'model_name': 'model_name',
'events': [
{'notification_id': 'ID_2', 'level': 'INFO',
'project': 'project', 'model_name': 'model_name'}
]
}
]
}
},
'type_1'
)

179
tests/orchestrator/workflows/subworkflows/test_load_notification_package.py Normal file
View File

@@ -0,0 +1,179 @@
from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark
from orchestrator.workflows.subworkflows.load_notification_package import LoadNotificationPackage
from orchestrator.activities.activities import Activities
@fixture
def load_notification_package():
return LoadNotificationPackage()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'test-workflow',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, load_notification_package):
input_data = {
'metadata': metadata,
'mail_type': 'test_mail_type',
'base_data_filter': {
'level': 'ERROR'
}
}
workflow_mock.start_local_activity_method.side_effect = [
'2023-01-01 12:00:00',
[
{
'id': '1',
}
],
[
{
'id_r': '1',
}
]
]
output = await load_notification_package.run(input_data)
assert output == {
'last_timestamp': '2023-01-01 12:00:00',
'notification_package': [
{
'id': '1',
}
],
'sending_configs': [
{
'id_r': '1',
}
]
}
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.get_last_data_timestamp,
{
**input_data['metadata'],
'mail_type': 'test_mail_type'
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.find_documents_in_mongodb,
{
**input_data['metadata'],
'query': {
'collection': 'receiver_groups',
'filters': {
'active': True
}
}
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_local_activity_method.assert_has_calls([
call(
Activities.load_latest_data,
{
**input_data['metadata'],
'collection_name': 'notification_queue',
'last_data_timestamp': '2023-01-01 12:00:00',
'base_data_filter': {
'level': 'ERROR'
}
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.start_activity_method.assert_has_calls([
call(
Activities.put_last_data_timestamp,
{
**input_data['metadata'],
'data': [
{
'id': '1',
}
],
'mail_type': 'test_mail_type'
},
start_to_close_timeout=ANY,
retry_policy=ANY
)
])
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock)
async def test_run_no_data(workflow_mock, load_notification_package):
input_data = {
'metadata': metadata,
'mail_type': 'test_mail_type',
'base_data_filter': {
'level': 'ERROR'
}
}
workflow_mock.start_local_activity_method.side_effect = [
'2023-01-01 12:00:00',
[],
[]
]
output = await load_notification_package.run(input_data)
assert output == {
'last_timestamp': '2023-01-01 12:00:00',
'notification_package': [],
'sending_configs': []
}
workflow_mock.start_activity_method.assert_not_called()
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock)
async def test_run_no_data(workflow_mock, load_notification_package):
input_data = {
'metadata': metadata,
'base_data_filter': {
'level': 'ERROR'
},
'mail_type': 'test_mail_type'
}
workflow_mock.start_local_activity_method.side_effect = [
'2023-01-01 12:00:00',
["data"],
[]
]
output = await load_notification_package.run(input_data)
assert output == {
'last_timestamp': '2023-01-01 12:00:00',
'notification_package': ["data"],
'sending_configs': []
}
workflow_mock.start_activity_method.assert_not_called()

84
tests/orchestrator/workflows/subworkflows/test_process_notifications.py Normal file
View File

@@ -0,0 +1,84 @@
from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark
from orchestrator.workflows.subworkflows.process_notifications import ProcessNotifications
from orchestrator.activities.activities import Activities
@fixture
def process_notifications():
return ProcessNotifications()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'test-workflow',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio
@patch("orchestrator.workflows.subworkflows.process_notifications.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, process_notifications):
input_data = {
'metadata': metadata,
'notification_package': ["content"],
'mail_type': 'test_mail_type',
'schema': 'test_schema',
'table_name': 'test_table_name',
}
response = await process_notifications.run(input_data)
assert response == workflow_mock.execute_local_activity_method.return_value
workflow_mock.execute_local_activity_method.assert_has_calls([
call(
Activities.build_email_html,
{
**metadata,
'receiver_groups': input_data['notification_package'],
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
call(
Activities.format_log_report,
{
**metadata,
'receiver_groups': workflow_mock.execute_activity_method.return_value,
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
call(
Activities.send_email,
{
**metadata,
'receiver_groups': workflow_mock.execute_local_activity_method.return_value,
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
),
call(
Activities.export_data_to_postgres,
{
**metadata,
'schema': input_data['schema'],
'table_name': input_data['table_name'],
'data': workflow_mock.execute_local_activity_method.return_value
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])

117
tests/orchestrator/workflows/test_alerts.py Normal file
View File

@@ -0,0 +1,117 @@
from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark
from orchestrator.workflows.alerts import Alerts
from orchestrator.activities.activities import Activities
@fixture
def alerts():
return Alerts()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'alerts',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio
@patch("orchestrator.workflows.alerts.workflow", new_callable=AsyncMock)
async def test_run_full_flow(workflow_mock, alerts):
input_data = {
'schedule_name': 'test-schedule-name',
'notification_ttl': 300,
'sent_ttl': 600
}
await alerts.run(input_data)
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'load_notification_package',
{
**input_data,
'metadata': metadata,
'base_data_filter': {
'level': 'ERROR'
}
}
)
])
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'process_notifications',
{
'metadata': metadata,
'mail_type': 'Alerts',
'notification_package': workflow_mock.execute_local_activity_method.return_value,
'schema': 'sientia_data',
'table_name': 'log_report'
}
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
call(
Activities.filter_notification_alerts,
{
**metadata,
'notification_package': workflow_mock.execute_child_workflow.return_value['notification_package'],
'sending_configs': workflow_mock.execute_child_workflow.return_value['sending_configs'],
'notification_ttl': input_data['notification_ttl']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
workflow_mock.execute_activity_method.assert_has_calls([
call(
Activities.store_notification_cache,
{
**metadata,
'log_report': workflow_mock.execute_child_workflow.return_value,
'sent_ttl': input_data['sent_ttl']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
@mark.asyncio
@patch("orchestrator.workflows.alerts.workflow", new_callable=AsyncMock)
async def test_run_no_data(workflow_mock, alerts):
workflow_mock.execute_child_workflow.return_value = {
'last_timestamp': '2023-01-01 12:00:00.000000',
'notification_package': [],
'sending_configs': []
}
input_data = {
'schedule_name': 'test-schedule-name',
'notification_ttl': 300,
'sent_ttl': 600
}
await alerts.run(input_data)
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'load_notification_package',
{
**input_data,
'metadata': metadata,
'base_data_filter': {
'level': 'ERROR'
}
}
)
])
workflow_mock.execute_local_activity_method.assert_not_called()

99
tests/orchestrator/workflows/test_reports.py Normal file
View File

@@ -0,0 +1,99 @@
from unittest.mock import AsyncMock, patch, ANY, call
from pytest import fixture, mark
from orchestrator.workflows.reports import Reports
from orchestrator.activities.activities import Activities
@fixture
def reports():
return Reports()
metadata = {
'metadata': {
'schedule_name': 'test-schedule-name',
'workflow_name': 'reports',
'model_name': '-',
'model_id': '-',
}
}
@mark.asyncio
@patch("orchestrator.workflows.reports.workflow", new_callable=AsyncMock)
async def test_run_full_flow(workflow_mock, reports):
input_data = {
'schedule_name': 'test-schedule-name',
'notification_ttl': 300,
'sent_ttl': 600
}
await reports.run(input_data)
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'load_notification_package',
{
**input_data,
'metadata': metadata,
'base_data_filter': {}
}
)
])
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'process_notifications',
{
'metadata': metadata,
'mail_type': 'Reports',
'notification_package': workflow_mock.execute_local_activity_method.return_value,
'schema': 'sientia_data',
'table_name': 'log_report'
}
)
])
workflow_mock.execute_local_activity_method.assert_has_calls([
call(
Activities.filter_notification_reports,
{
**metadata,
'notification_package': workflow_mock.execute_child_workflow.return_value['notification_package'],
'sending_configs': workflow_mock.execute_child_workflow.return_value['sending_configs']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])
@mark.asyncio
@patch("orchestrator.workflows.reports.workflow", new_callable=AsyncMock)
async def test_run_no_data(workflow_mock, reports):
workflow_mock.execute_child_workflow.return_value = {
'last_timestamp': '2023-01-01 12:00:00.000000',
'notification_package': [],
'sending_configs': []
}
input_data = {
'schedule_name': 'test-schedule-name',
'notification_ttl': 300,
'sent_ttl': 600
}
await reports.run(input_data)
workflow_mock.execute_child_workflow.assert_has_calls([
call(
'load_notification_package',
{
**input_data,
'metadata': metadata,
'base_data_filter': {}
}
)
])
workflow_mock.execute_local_activity_method.assert_not_called()

36
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.2.5"
tag: "0.2.7"
# 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:
@@ -132,7 +132,7 @@ env:
- name: GITHUB_REPO_URL
value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git"
- name: GITHUB_BRANCH
value: "SIENTIAPDE-1171-criar-pipeline-de-retreino-laborious"
value: "SIENTIAPDE-1172-criar-pipeline-de-alertas-orquestrador"
- name: PYTHON_APP
value: "orchestrator.worker.worker"
@@ -170,6 +170,34 @@ env:
- name: MONGODB_TTL_INDEX_HOURS
value: "1"
- name: EMAIL_SENDER
value: "vitor.santos@aignosi.com.br"
- name: EMAIL_SENDER_PASSWORD
valueFrom:
secretKeyRef:
name: smtp-credentials
key: app_password
- name: EMAIL_SMTP_SERVER
value: "smtp.gmail.com"
- name: EMAIL_SMTP_PORT
value: "587"
# Application variables
- name: POSTGRES_HOST
value: "paradedb-rw.paradedb.svc.cluster.local"
- name: POSTGRES_PORT
value: "5432"
- name: POSTGRES_USER
value: "sientia"
- name: POSTGRES_PASSWORD
value: "sientia"
- name: POSTGRES_DBNAME
value: "sientia"
- name: POSTGRES_MIN_CONNECTIONS
value: "10"
- name: POSTGRES_MAX_CONNECTIONS
value: "40"
- name: KAFKA_BOOTSTRAP_SERVERS
value: "kafka.kafka.svc.cluster.local:9092"
@@ -201,3 +229,7 @@ ssh:
# --namespace sientia \
# --from-file=ssh-privatekey=git_key \
# --type=kubernetes.io/ssh-auth
# kubectl create secret generic smtp-credentials \
# --namespace sientia \
# --from-literal=app_password='sua-senha-de-app-de-16-digitos'