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

SIENTIAPDE-1172

Browse Source 
refactor: streamline email generation and enhance notification handling

- Updated email HTML structure to improve clarity and presentation of notifications.
- Enhanced notification generator to better manage execution counts and output formatting.
- Refactored Email class to improve error handling and logging during email sending.
- Adjusted EmailBuilder to efficiently process error, warning, and info models.
- Improved test coverage for email and notification functionalities to ensure reliability.
This commit is contained in:
vitor-aignosi
2025-07-29 10:25:31 -03:00
parent c3ba2e4b5a
commit bc456cc8be
14 changed files with 503 additions and 401 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
email.html
View File

@@ -15,231 +15,11 @@
<body>
<h1>SIENTIA™ Alerts</h1>
<h3>Errors detected:</h3>
<h3></h3>
<h4>Model: <span>do elit</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<h3></h3>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>lorem_adipiscing</td>
<td>2025-07-28 20:05:38.650661+00:00</td>
<td>incididunt dolor lorem incididunt elit incididunt dolor consectetur ipsum ut</td>
</tr>
</tbody>
</table>
<h4>Model: <span>dolor magna</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>tempor_dolore_eiusmod</td>
<td>2025-07-28 20:05:38.650910+00:00</td>
<td>ipsum amet ipsum aliqua et eiusmod</td>
</tr>
</tbody>
</table>
<h4>Model: <span>labore sit lorem</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_CONNECTION_RETRY__server_6</td>
<td>dolor_labore_lorem</td>
<td>2025-07-28 20:05:38.650949+00:00</td>
<td>et adipiscing ipsum ut aliqua ut adipiscing aliqua amet</td>
</tr>
</tbody>
</table>
<h3>Warnings detected:</h3>
<h4>Model: <span>incididunt adipiscing elit sit</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>ADIPISCING_AMET_IPSUM</td>
<td>eiusmod_aliqua_incididunt</td>
<td>2025-07-28 20:05:38.650621+00:00</td>
<td>magna incididunt dolore sed eiusmod sit incididunt ut</td>
</tr>
</tbody>
</table>
<h4>Model: <span>do adipiscing et</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node46:tag_73_LISTENNING_STOPPED</td>
<td>amet_dolor</td>
<td>2025-07-28 20:05:38.650706+00:00</td>
<td>dolore do et consectetur adipiscing do magna eiusmod eiusmod</td>
</tr>
</tbody>
</table>
<h4>Model: <span>lorem dolore amet</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node13:tag_59_LISTENNING_STOPPED</td>
<td>ipsum_tempor_ipsum</td>
<td>2025-07-28 20:05:38.650785+00:00</td>
<td>et sed adipiscing tempor dolor magna do lorem</td>
</tr>
</tbody>
</table>
<h4>Model: <span>eiusmod sed</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_SUBSCRIPTION__server_2</td>
<td>do_ipsum</td>
<td>2025-07-28 20:05:38.650826+00:00</td>
<td>eiusmod et consectetur do dolore dolor sed amet</td>
</tr>
</tbody>
</table>
<h3>Infos detected:</h3>
<h4>Model: <span>et sed</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>OPC_LISTENNING_STOPPED__server_6</td>
<td>labore_do_ipsum_ipsum</td>
<td>2025-07-28 20:05:38.650570+00:00</td>
<td>tempor dolore eiusmod sit sit sed elit</td>
</tr>
</tbody>
</table>
<h4>Model: <span>dolore sit</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>REPORT_PARTITION_MANAGER</td>
<td>et_lorem_incididunt_ipsum</td>
<td>2025-07-28 20:05:38.650742+00:00</td>
<td>adipiscing magna adipiscing adipiscing sit tempor</td>
</tr>
</tbody>
</table>
<h4>Model: <span>adipiscing do</span></h4>
<table>
<thead>
<tr>
<th>Notification ID</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>TAG_node76:tag_73_LISTENNING_STOPPED</td>
<td>tempor_elit</td>
<td>2025-07-28 20:05:38.650875+00:00</td>
<td>et do aliqua lorem incididunt</td>
</tr>
</tbody>
</table>
<h3></h3>

343
notification_generator.ipynb
View File

File diff suppressed because one or more lines are too long

48
orchestrator/activities/email.py
View File

@@ -95,6 +95,8 @@ class Email(BaseActivity):
self.logger.error(
f"Failed to attach content of {att_name}: {e}")
raise e
return msg
@activity.defn(name="send_email")
@@ -113,29 +115,29 @@ class Email(BaseActivity):
metadata=metadata)
for group_name, group_config in receiver_groups.items():
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",
"content": notification['attachment_content']
}
for notification in group_config['notifications']
if notification['attachment_content']],
msg)
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.server.sendmail(
self.sender_email, receivers, msg.as_string())
except Exception as e:
@@ -146,7 +148,7 @@ class Email(BaseActivity):
else:
group_config['status'] = 'sent'
self.info(f"Email sent to {group_name}.",
self.info(f"Email sent to {group_name}: {receivers}",
metadata=metadata)
self.info(f"Email sent for {mail_type} mail type.",

2
orchestrator/activities/slot_manager.py
View File

@@ -335,3 +335,5 @@ class SlotManager(Redis):
if status == 'sent':
key = f"{row['schedule']}:{row['notification_id']}"
self.set(key, now, ttl=sent_ttl)
self.info("Notification cache stored...", metadata=metadata)

9
orchestrator/utils/email_builder.py
View File

@@ -31,11 +31,14 @@ class EmailBuilder:
return {
'mail_type': mail_type,
'error_events': self.replace_parameters(self.general_template,
error_models) if error_models else '',
general_events.get(
'ERROR')) if error_models else '',
'warning_events': self.replace_parameters(self.general_template,
warning_models) if warning_models else '',
general_events.get(
'WARNING')) if warning_models else '',
'info_events': self.replace_parameters(self.general_template,
info_models) if info_models else '',
general_events.get(
'INFO')) if info_models else '',
}
def build_email(self, report_data: list[dict], mail_type: str) -> str:

2
orchestrator/utils/templates/general_template.html
View File

@@ -5,6 +5,7 @@
<thead>
<tr>
<th>Notification ID</th>
<th>Schedule</th>
<th>Block</th>
<th>Timestamp</th>
<th>Message</th>
@@ -14,6 +15,7 @@
{% 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>

10
orchestrator/workflows/subworkflows/load_notification_package.py
View File

@@ -67,14 +67,6 @@ class LoadNotificationPackage:
start_to_close_timeout=timedelta(seconds=60),
retry_policy=retry_policy
)
if not notification_package:
return {
'last_timestamp': last_timestamp,
'notification_package': [],
'sending_configs': await sending_configs_handler
}
sending_configs = await sending_configs_handler
if not sending_configs or not notification_package:
@@ -103,5 +95,5 @@ class LoadNotificationPackage:
return {
'last_timestamp': last_timestamp,
'notification_package': notification_package,
'sending_configs': await sending_configs_handler
'sending_configs': sending_configs
}

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
)

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

@@ -11,8 +11,8 @@ def email(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password="test",
smpt_server="test",
port=587,
smtp_server="test",
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock()
)
@@ -27,21 +27,22 @@ def test___init___with_password(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password="test",
smpt_server="test",
port=587,
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.port == 587
assert email.smtp_port == 587
smtplib.SMTP_SSL.assert_called_once_with("test", 587)
smtplib.SMTP_SSL.return_value.login.assert_called_once_with(
smtplib.SMTP.assert_called_once_with("test", 587)
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_SSL.return_value
assert email.server == smtplib.SMTP.return_value
@patch('orchestrator.activities.email.EmailBuilder')
@@ -50,15 +51,15 @@ def test___init___without_password(smtplib, email_builder):
email = Email(
sender_email="test@test.com",
sender_password=None,
smpt_server="test",
port=587,
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.port == 587
assert email.smtp_port == 587
smtplib.SMTP.assert_called_once_with("test", 587)
assert email.server == smtplib.SMTP.return_value
@@ -189,7 +190,10 @@ def test_handle_attachments_failure(mime_base, email):
}
]
email.handle_attachments(attachments, message)
try:
email.handle_attachments(attachments, message)
except Exception as e:
assert str(e) == "test"
assert message.attach.call_count == 0

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

@@ -555,17 +555,11 @@ async def test_load_latest_data_none_last_data_timestamp(mongo_db):
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'timestamp': {
0: '2023-01-01 12:00:00.000000'
}
}
assert result == [{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00.000000'
}]
@mark.asyncio
@@ -606,17 +600,11 @@ async def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
{"_id": 0}
)
assert result == {
'name': {
0: 'test1'
},
'value': {
0: 1
},
'timestamp': {
0: '2023-01-01 12:00:00.000000'
}
}
assert result == [{
'name': 'test1',
'value': 1,
'timestamp': '2023-01-01 12:00:00.000000'
}]
@mark.asyncio

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
}

59
tests/orchestrator/workflows/subworkflows/test_load_notification_package.py
View File

@@ -23,7 +23,10 @@ metadata = {
@patch("orchestrator.workflows.subworkflows.load_notification_package.workflow", new_callable=AsyncMock)
async def test_run(workflow_mock, load_notification_package):
input_data = {
'metadata': metadata
'metadata': metadata,
'base_data_filter': {
'level': 'ERROR'
}
}
workflow_mock.start_local_activity_method.side_effect = [
@@ -113,3 +116,57 @@ async def test_run(workflow_mock, load_notification_package):
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,
'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'
}
}
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()

9
tests/orchestrator/workflows/subworkflows/test_process_notifications.py
View File

@@ -39,11 +39,14 @@ async def test_run(workflow_mock, process_notifications):
Activities.build_email_html,
{
**metadata,
'receiver_groups': input_data['notification_package']
'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,
{
@@ -51,6 +54,8 @@ async def test_run(workflow_mock, process_notifications):
'receiver_groups': workflow_mock.execute_activity_method.return_value,
'mail_type': input_data['mail_type']
},
schedule_to_close_timeout=ANY,
retry_policy=ANY
)
])

39
tests/orchestrator/workflows/test_alerts.py
View File

@@ -64,7 +64,9 @@ async def test_run_full_flow(workflow_mock, alerts):
'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
)
])
@@ -75,6 +77,41 @@ async def test_run_full_flow(workflow_mock, alerts):
**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()