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

SIENTIAPDE-1974

Browse Source 
Add e2e tag-database scenarios and cover worker/email/formatter gaps

- Add e2e scenarios 1.9.1-1.9.4 for PI Web API through the tag
  database: scouter and output tag resolution, mixed-connector schedule
  hold, and OPC slot unchanged by PI connectors. Seed connectors and
  connector_tags from a new scenario input.
- Tighten the happy-path e2e to assert both schedules and that no PI
  output config is produced when the tag database is unused.
- Drop dead `if self.server` / `if notification_handler` guards in the
  email reconnect and worker shutdown paths, since both are always set.
- Log the traceback inside the message on worker failure, as the
  sientia_do Logger.error does not accept exc_info.
- Add unit tests for the worker failure exit, Prometheus startup
  failure, email reconnect without a password, and formatter, mongo,
  temporal manager and orchestrator edge cases.
This commit is contained in:
Vitor Pimentel dos Santos
2026-10-28 16:07:10 -03:00
parent 16e76f11ea
commit 11e350a6ce
13 changed files with 761 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
e2e/helpers.py
View File

@@ -27,6 +27,8 @@ MONGO_COLLECTIONS = (
'pipelines',
'opc_servers',
'opc-servers',
'connectors',
'connector_tags',
)
DATETIME_FORMAT_MS_WITH_TZ = '%Y-%m-%d %H:%M:%S.%f%z'
@@ -160,6 +162,23 @@ def seed_opc_servers(
client.close()
def seed_tag_database(
mongo_uri: str,
database: str,
connectors: list[dict[str, Any]],
connector_tags: list[dict[str, Any]],
) -> None:
"""Insert tag database documents (connectors, connector_tags) into the test Mongo database."""
client = MongoClient(mongo_uri)
try:
if connectors:
client[database]['connectors'].insert_many(connectors)
if connector_tags:
client[database]['connector_tags'].insert_many(connector_tags)
finally:
client.close()
def seed_receiver_groups(
mongo_uri: str,
database: str,

118
e2e/scenario_inputs/orchestrator_pi_web_api.json Normal file
View File

@@ -0,0 +1,118 @@
{
"workflow_input": {
"schedule_name": "orchestrator-e2e-pi-web-api",
"pipelines_query": {
"collection": "pipelines",
"aggregation": [{"$match": {"active": true}}]
},
"opc_servers_query": {
"collection": "opc_servers",
"filters": {"active": true}
}
},
"connectors": [
{
"connector_id": "pi-usina",
"type": "pi_web_api",
"name": "PI Usina",
"url": "https://pi-usina.example/piwebapi",
"status": "Active",
"pi_web_api": {
"auth": {"type": "bearer", "secret": {"name": "pi-web-api-auth-token", "key": "token"}},
"period": "*-2h",
"max_count": 10
}
},
{
"connector_id": "pi-other",
"type": "pi_web_api",
"name": "PI Other",
"url": "https://pi-other.example/piwebapi",
"status": "Active",
"pi_web_api": {
"auth": {"type": "basic", "secret": {"name": "pi-other-auth", "key": "credentials"}}
}
},
{
"connector_id": "plant",
"type": "opc_ua",
"name": "plant",
"url": "opc.tcp://plant:4840",
"server_uri": "urn:plant",
"use_security": false,
"subscription_period_ms": 500,
"status": "Active"
}
],
"connector_tags": [
{"tag_id": 42, "connector_id": "pi-usina", "address": "F1DP-A", "tag_name": "TI-101-db", "category": "Read", "status": "Active"},
{"tag_id": 43, "connector_id": "pi-other", "address": "F1DP-B", "tag_name": "TI-102-db", "category": "Read", "status": "Active"},
{"tag_id": 77, "connector_id": "pi-usina", "address": "F1DP-P", "tag_name": "PRED-101", "category": "Write", "status": "Active"},
{"tag_id": 78, "connector_id": "pi-usina", "address": "F1DP-C", "tag_name": "CONF-101", "category": "Prediction Confidence", "status": "Active"},
{"tag_id": 7, "connector_id": "plant", "address": "ns=3;i=1001", "tag_name": "Pressure", "category": "Read", "status": "Active"}
],
"pipelines": [
{
"schedule_name": "pi-tagdb",
"workflow_type": "pi_web_api_scouter",
"model_id": "model-pi",
"model": {"name": "Model model-pi"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": {
"TI-101": {"tag_id": 42, "aggr_func": "avg"},
"TI-404": {"tag_id": 404}
},
"pi_web_api_config": {"endpoint": "/streamsets/recorded", "max_count": 5}
},
{
"schedule_name": "pi-legacy",
"workflow_type": "pi_web_api_scouter",
"model_id": "model-pi-legacy",
"model": {"name": "Model model-pi-legacy"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": {"TI-900": {"webid": "F1DP-X"}},
"pi_web_api_config": {"endpoint": "/streamsets/recorded", "api_timeout": 10}
},
{
"schedule_name": "pred-pi",
"workflow_type": "predictions_batch",
"model_id": "model-pi",
"model": {"name": "Model model-pi"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"query": "SELECT 1",
"pi_web_api_output_config": {
"prediction_tags": [{"tag_id": 77}],
"confidence_tags": [{"tag_id": 78}, {"tag_id": 42}]
}
},
{
"schedule_name": "pred-pi-legacy",
"workflow_type": "predictions_batch",
"model_id": "model-pi-legacy",
"model": {"name": "Model model-pi-legacy"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"query": "SELECT 1",
"pi_web_api_output_config": {"prediction_tags": {"PRED-900": "F1DP-L"}}
},
{
"schedule_name": "opc-tagdb",
"workflow_type": "scouter",
"model_id": "model-opc",
"model": {"name": "Model model-opc"},
"active": true,
"updated_at": "@now",
"frequency": "1m",
"read_tags": [{"tag_id": 7, "frequency": 1000, "aggr_func": "avg"}]
}
],
"opc_servers": [],
"active_ingestors": ["heartbeat:ingestor:1"]
}

36
e2e/scenarios.md
View File

@@ -26,7 +26,9 @@ Source: `e2e/test_orchestrator_main_workflow.py`
### 1.1.1 Happy path
Pipelines in Mongo → Temporal schedule created in the `laborious` namespace, `orchestrated_schedules` entries written.
Pipelines in Mongo → both Temporal schedules created in the `laborious` namespace, one
`orchestrated_schedules` entry per pipeline, and no PI Web API output config when no pipeline uses
the tag database.
### 1.3.1 Create-only
@@ -64,6 +66,38 @@ Two `drift` pipelines on `frequency='1m'`: one without `interval_minutes` and wi
`interval_minutes=60` and `chunk_period='s'` → both forwarded unchanged, no warning
(SIENTIAPDE-2212).
### 1.9 PI Web API through the tag database (connectors-pi-web-api)
Scenario input: `orchestrator_pi_web_api.json` (two `pi_web_api` connectors, one `opc_ua`
connector, their `connector_tags`, PI Web API pipelines by `tag_id` and legacy ones, and an OPC
scouter by `tag_id`).
#### 1.9.1 PI scouter through the tag database
`pi_web_api_scouter` with `tag_id` read tags → `model_tags` WebID from `connector_tags.address`,
`pi_web_api_query` with the connector `period` under the pipeline `max_count` (`api_timeout` capped
at the frequency), and `connection` with base URL and the secret reference only. An unknown
`tag_id` is left out and notified (`ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION`).
A legacy pipeline (inline `webid`) gets the pre-change input, without `connection`.
#### 1.9.2 PI output tags through the tag database
`predictions_batch` `pi_web_api_output_config` given as `tag_id` lists → `{tag_name: WebID}` from
`Write` / `Prediction Confidence` tags; a `Read` tag given as a confidence tag is left out and
notified (`ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION`). A legacy
`{tag_name: webid}` mapping passes through unchanged.
#### 1.9.3 Mixed PI connectors hold the schedule
A pipeline updated to read tags of two PI connectors → its existing scouter schedule is neither
updated nor deleted, and the conflict is notified
(`ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION`).
#### 1.9.4 OPC slot unchanged by PI connectors
The OPC slot v2 written to Redis for an OPC scouter by `tag_id` is the expected one with the PI
connectors and tags present, and identical after removing them.
---
## 2. Alerts workflow

230
e2e/test_orchestrator_main_workflow.py
View File

@@ -1,5 +1,6 @@
"""E2E tests for the Orchestrator main workflow."""
import json
from datetime import UTC, datetime, timedelta
import pytest
@@ -17,6 +18,7 @@ from e2e.helpers import (
seed_opc_servers,
seed_orchestrated_schedules,
seed_pipelines,
seed_tag_database,
start_and_await_workflow,
)
from orchestrator.workflows.orchestrator import Orchestrator
@@ -41,7 +43,7 @@ def _apply_orchestrator_seeds(
redis_client: Redis,
) -> None:
"""
Seed Mongo and Redis with the pipelines/opc_servers/ingestors declared in a scenario.
Seed Mongo and Redis with the pipelines/opc_servers/tag database/ingestors of a scenario.
Args:
scenario: Scenario payload returned by ``load_scenario_input``.
@@ -50,6 +52,12 @@ def _apply_orchestrator_seeds(
"""
seed_pipelines(mongo_uri, E2E_DATABASE, scenario.get('pipelines', []))
seed_opc_servers(mongo_uri, E2E_DATABASE, scenario.get('opc_servers', []))
seed_tag_database(
mongo_uri,
E2E_DATABASE,
scenario.get('connectors', []),
scenario.get('connector_tags', []),
)
seed_active_ingestors(redis_client, scenario.get('active_ingestors', []))
if scenario.get('orchestrated_schedules'):
seed_orchestrated_schedules(
@@ -84,14 +92,17 @@ async def test_scenario_1_1_1_happy_path(
try:
tracked = list(mongo[E2E_DATABASE]['orchestrated_schedules'].find())
names = {doc['schedule_name'] for doc in tracked}
assert names, f'Expected orchestrated_schedules rows, got {tracked}'
assert names == {'pred-legacy', 'drift-gpu'}, f'Unexpected orchestrated_schedules {tracked}'
finally:
mongo.close()
laborious_schedules = await _schedule_ids(temporal_host, 'laborious')
assert 'pred-legacy' in laborious_schedules or 'drift-gpu' in laborious_schedules, (
assert {'pred-legacy', 'drift-gpu'} <= set(laborious_schedules), (
f'Expected Temporal schedules in laborious, got {laborious_schedules}'
)
# No pipeline uses the tag database: the PI Web API loads stay empty
pred = await _schedule_input(temporal_host, 'pred-legacy')
assert pred['pi_web_api_output_config'] == {}
@pytest.mark.e2e
@@ -315,9 +326,9 @@ async def test_scenario_1_7_2_simple_metrics_adaptive_frequency_round_trip(
assert await _schedule_every(temporal_host, short) == timedelta(hours=1)
async def _schedule_input(host: str, schedule_id: str) -> dict:
"""Return the decoded workflow input of a laborious-namespace schedule."""
client = await Client.connect(host, namespace='laborious')
async def _schedule_input(host: str, schedule_id: str, namespace: str = 'laborious') -> dict:
"""Return the decoded workflow input of a schedule (laborious namespace by default)."""
client = await Client.connect(host, namespace=namespace)
description = await client.get_schedule_handle(schedule_id).describe()
action = description.schedule.action
assert isinstance(action, ScheduleActionStartWorkflow)
@@ -364,3 +375,210 @@ async def test_scenario_1_8_1_drift_window_and_chunk_period(
)
assert len(warnings) == 1
assert warnings[0]['model_id'] == 'model-dr'
# 1.9 PI Web API through the tag database (connectors-pi-web-api)
PI_SECRET = {'name': 'pi-web-api-auth-token', 'key': 'token'}
async def _run_pi_scenario(temporal_env: WorkflowEnvironment, scenario: dict, tag: str) -> None:
await start_and_await_workflow(
temporal_env.client,
Orchestrator.run,
scenario['workflow_input'],
make_workflow_id(f'orchestrator-pi-{tag}'),
)
def _notifications(notification_handler, notification_id: str) -> list[dict]:
return list(notification_handler.mongo_collection.find({'notification_id': notification_id}))
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_1_pi_scouter_through_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.1: tag_id read tags get WebID + connection from the connector; legacy unchanged."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'scouter')
resolved = await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter')
# The unknown tag_id is left out; the rest of the pipeline is still scheduled
assert resolved['model_tags'] == {
'TI-101': {'webid': 'F1DP-A', 'aggr_func': 'avg', 'data_range': [-100, 100]}
}
assert resolved['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-2h', # connector default
'max_count': 5, # pipeline wins over the connector
'api_timeout': 60, # capped at the frequency
}
assert resolved['connection'] == {
'connector_id': 'pi-usina',
'base_url': 'https://pi-usina.example/piwebapi',
'auth': {'type': 'bearer', 'secret': PI_SECRET},
}
legacy = await _schedule_input(temporal_host, 'pi-legacy', namespace='scouter')
assert 'connection' not in legacy
assert legacy['model_tags'] == {
'TI-900': {'webid': 'F1DP-X', 'aggr_func': 'lts', 'data_range': [-100, 100]}
}
assert legacy['pi_web_api_query'] == {
'endpoint': '/streamsets/recorded',
'period': '*-1d',
'max_count': 1,
'api_timeout': 10,
}
[not_found] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_TAG_NOT_FOUND_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pi-tagdb/404' in not_found['message']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_2_pi_output_tags_through_tag_database(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.2: predictions_batch output tags by tag_id become {tag_name: WebID}; legacy unchanged."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'output')
resolved = await _schedule_input(temporal_host, 'pred-pi')
# The Read tag given as a confidence tag is left out
assert resolved['pi_web_api_output_config'] == {
'prediction_tags': {'PRED-101': 'F1DP-P'},
'confidence_tags': {'CONF-101': 'F1DP-C'},
}
legacy = await _schedule_input(temporal_host, 'pred-pi-legacy')
assert legacy['pi_web_api_output_config'] == {'prediction_tags': {'PRED-900': 'F1DP-L'}}
[wrong_category] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_TAG_WRONG_CATEGORY_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pred-pi/42' in wrong_category['message']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_3_pi_mixed_connectors_hold_schedule(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
temporal_host: str,
redis_client: Redis,
notification_handler,
):
"""1.9.3: tags of two PI connectors in one pipeline leave its schedule untouched."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'hold-create')
before = await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter')
mongo = MongoClient(mongo_uri)
try:
mongo[E2E_DATABASE]['pipelines'].update_one(
{'schedule_name': 'pi-tagdb'},
{
'$set': {
'read_tags.TI-102': {'tag_id': 43},
'updated_at': datetime.now(UTC) + timedelta(minutes=1),
}
},
)
finally:
mongo.close()
await _run_pi_scenario(temporal_env, scenario, 'hold-mixed')
assert 'pi-tagdb' in await _schedule_ids(temporal_host, 'scouter')
assert await _schedule_input(temporal_host, 'pi-tagdb', namespace='scouter') == before
[mixed] = _notifications(
notification_handler, 'ORCHESTRATOR_PI_MIXED_CONNECTORS_DURING_SCHEDULE_CONFIGURATION'
)
assert 'pi-tagdb/pi-other, pi-usina' in mixed['message']
def _opc_slots(redis_client: Redis) -> dict[str, dict]:
return {
key if isinstance(key, str) else key.decode(): json.loads(redis_client.get(key))
for key in redis_client.keys('slot:opc_tags:*')
}
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_scenario_1_9_4_opc_slot_unchanged_by_pi_connectors(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
redis_client: Redis,
):
"""1.9.4: the OPC slot built from the tag database ignores the PI connectors and tags."""
scenario = load_scenario_input('orchestrator_pi_web_api')
_apply_orchestrator_seeds(scenario, mongo_uri, redis_client)
await _run_pi_scenario(temporal_env, scenario, 'slot-with-pi')
with_pi = _opc_slots(redis_client)
assert with_pi == {
'slot:opc_tags:1': {
'plant': {
'server_id': 'plant',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': False,
'subscription_period_ms': 500,
'tags': {
'ns=3;i=1001': {
'frequency': 1000,
'active_read': {'enabled': False},
'names': [
{
'collection': 'raw_opc-tagdb',
'tag_name': 'Pressure',
'model_id': 'model-opc',
'model_name': 'Model model-opc',
}
],
}
},
}
}
}
mongo = MongoClient(mongo_uri)
try:
mongo[E2E_DATABASE]['connectors'].delete_many({'type': 'pi_web_api'})
mongo[E2E_DATABASE]['connector_tags'].delete_many({'connector_id': {'$ne': 'plant'}})
finally:
mongo.close()
redis_client.delete(*redis_client.keys('slot:opc_tags:*'))
await _run_pi_scenario(temporal_env, scenario, 'slot-without-pi')
assert _opc_slots(redis_client) == with_pi

15
orchestrator/activities/email.py
View File

@@ -177,14 +177,13 @@ class Email(SientiaMonitoring):
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
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:

10
orchestrator/worker/worker.py
View File

@@ -5,6 +5,7 @@ with workflow.unsafe.imports_passed_through():
import asyncio
import os
import sys
import traceback
from prometheus_client import start_http_server
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -192,13 +193,12 @@ async def main():
try:
await asyncio.gather(*handlers)
except BaseException as e:
logger.error(f'An unhandled exception occurred: {e}', exc_info=True)
# sientia_do's Logger.error takes only a message: the traceback goes in it
logger.error(f'An unhandled exception occurred: {e}\n{traceback.format_exc()}')
exit_code = 1
finally:
if notification_handler:
notification_handler.shutdown()
if activities:
activities.shutdown()
notification_handler.shutdown()
activities.shutdown()
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
sys.exit(exit_code)

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

@@ -326,3 +326,37 @@ def test_send_email(mimemultipart, mimetext, email):
)
assert email.try_send_email.call_count == 2
@patch('orchestrator.activities.email.EmailBuilder')
@patch('orchestrator.activities.email.smtplib')
def test___init___without_smtp_server(smtplib, email_builder):
email = Email(
sender_email='test@test.com',
sender_password='test',
smtp_server=None,
smtp_port=587,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
smtplib.SMTP.assert_not_called()
assert not hasattr(email, 'server')
@patch('orchestrator.activities.email.smtplib.SMTP')
def test_try_send_email_reconnect_without_password(smtp, email):
email.sender_password = None
email.server.sendmail = MagicMock(side_effect=SMTPServerDisconnected('test'))
email.server.quit = MagicMock()
msg = MagicMock()
email.try_send_email(msg, 'test')
smtp.return_value.starttls.assert_not_called()
smtp.return_value.login.assert_not_called()
smtp.return_value.sendmail.assert_called_once_with(
'test@test.com', 'test', msg.as_string.return_value
)

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

@@ -1649,3 +1649,148 @@ def test_create_schedule_config_leaves_held_schedule_untouched(formatters):
'to_create': {'scouter': {}, 'laborious': {}},
'to_delete': {'scouter': [], 'laborious': []},
}
def test_format_schedule_config_groups_schedules_of_one_namespace(formatters):
result = formatters.format_schedule_config(
{
'schedule_config': [
{'namespace': 'laborious', 'schedule_name': 'a', 'updated_at': '2021-01-01'},
{'namespace': 'laborious', 'schedule_name': 'b', 'updated_at': '2021-01-02'},
],
**metadata,
}
)
assert result == {'laborious': {'a': '2021-01-01', 'b': '2021-01-02'}}
def test_report_schedule_orchestration_success_only_and_empty_types(formatters):
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.report_schedule_orchestration(
{
**metadata,
'created_schedules': [
{'namespace': 'laborious', 'schedule_name': 'a', 'success': True}
],
'updated_schedules': [],
'deleted_schedules': [],
}
)
formatters.send_success_report.assert_called_once()
assert formatters.send_success_report.call_args.kwargs['notification_id'] == (
'REPORT_ORCHESTRATION_CREATED_SCHEDULES'
)
formatters.send_error_report.assert_not_called()
def test_report_schedule_orchestration_errors_only(formatters):
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.report_schedule_orchestration(
{
**metadata,
'created_schedules': [],
'updated_schedules': [],
'deleted_schedules': [
{
'namespace': 'scouter',
'schedule_name': 'b',
'success': False,
'message': 'boom',
'attachment': None,
}
],
}
)
formatters.send_success_report.assert_not_called()
formatters.send_error_report.assert_called_once()
assert formatters.send_error_report.call_args.kwargs['notification_id'] == (
'REPORT_ORCHESTRATION_DELETED_SCHEDULES_ERROR'
)
def test_report_slot_orchestration_success_only_inserts(formatters):
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.report_slot_orchestration(
{**metadata, 'inserted_slots': {'1': {'success': True}}, 'deleted_slots': {}}
)
formatters.send_success_report.assert_called_once_with(
metadata=metadata['metadata'],
message='Inserted slots: \n 1',
notification_id='REPORT_ORCHESTRATION_INSERTED_SLOTS',
)
formatters.send_error_report.assert_not_called()
def test_report_slot_orchestration_errors_only_deletes(formatters):
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
deleted = {'2': {'success': False, 'error': 'boom'}}
formatters.report_slot_orchestration(
{**metadata, 'inserted_slots': {}, 'deleted_slots': deleted}
)
formatters.send_success_report.assert_not_called()
formatters.send_error_report.assert_called_once_with(
metadata=metadata['metadata'],
message='Failed to delete slots: \n 2',
notification_id='REPORT_ORCHESTRATION_DELETED_SLOTS',
attachment=deleted,
)
def test_report_slot_orchestration_errors_only_inserts_success_only_deletes(formatters):
formatters.send_success_report = MagicMock()
formatters.send_error_report = MagicMock()
formatters.report_slot_orchestration(
{
**metadata,
'inserted_slots': {'1': {'success': False, 'error': 'boom'}},
'deleted_slots': {'2': {'success': True}},
}
)
assert [c.kwargs['notification_id'] for c in formatters.send_error_report.call_args_list] == [
'REPORT_ORCHESTRATION_INSERTED_SLOTS'
]
assert [c.kwargs['notification_id'] for c in formatters.send_success_report.call_args_list] == [
'REPORT_ORCHESTRATION_DELETED_SLOTS'
]
def test_format_log_report_same_notification_twice_in_one_group(formatters):
notification = {
'notification_id': 'n1',
'trigger': 't1',
'timestamp': '2021-01-01',
'message': 'm',
'level': 'ERROR',
'block': 'b',
'pipeline': 'p',
'project': 'pr',
'model_name': 'mn',
'model_id': 'mi',
}
result = formatters.format_log_report(
{
**metadata,
'receiver_groups': {
'g1': {'notifications': [notification, dict(notification)], 'status': 'sent'}
},
'mail_type': 'alerts',
}
)
assert [row['groups'] for row in DataFrame(result).to_dict('records')] == [['g1']]

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

@@ -704,3 +704,48 @@ def test_load_latest_data_error(mongo_db):
level=NotificationLevel.ERROR,
attachment_content=ANY,
)
def test_find_documents_in_mongodb_leaves_documents_without_timestamp_field(mongo_db):
mongo_db.mongo_db_repository.find = MagicMock(return_value=[{'name': 'no-timestamp'}])
result = mongo_db.find_documents_in_mongodb(
{'query': {'collection': 'test_collection'}, 'timestamp_fields': ['timestamp']}
)
assert result == [{'name': 'no-timestamp'}]
def test_aggregate_documents_in_mongodb_leaves_documents_without_timestamp_field(mongo_db):
mongo_db.mongo_db_repository.aggregate = MagicMock(return_value=[{'name': 'no-timestamp'}])
result = mongo_db.aggregate_documents_in_mongodb(
{
'query': {'collection': 'test_collection', 'aggregation': [{'$match': {}}]},
'timestamp_fields': ['timestamp'],
}
)
assert result == [{'name': 'no-timestamp'}]
@patch('orchestrator.activities.mongo_db.now')
def test_create_pipelines_timestamps_all_failed_inserts_nothing(_now_mock, mongo_db):
mongo_db.mongo_db_repository.insert_many = MagicMock()
mongo_db.create_pipelines_timestamps(
{'created_pipelines': [{'schedule_name': 'a', 'namespace': 'n', 'success': False}]}
)
mongo_db.mongo_db_repository.insert_many.assert_not_called()
def test_create_collection_with_ttl_index_skips_pipelines_without_topic(mongo_db):
database = mongo_db.mongo_db_repository.database
database.list_collection_names = MagicMock(return_value=[])
mongo_db.create_collection_with_ttl_index(
{**metadata, 'pipelines': {'pi-scouter': {'model_tags': {}}, 'held': {'held': True}}}
)
database.create_collection.assert_not_called()

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

@@ -723,3 +723,23 @@ async def test_create_schedules_drift_keeps_default_skip_overlap(
mock_schedule.assert_called_once()
policy = mock_schedule.call_args.kwargs.get('policy')
assert policy is None or policy.overlap == ScheduleOverlapPolicy.SKIP
@mark.asyncio
async def test_normalize_schedules_keeps_schedules_known_in_mongo(temporal_manager):
"""An orchestrated schedule still tracked in orchestrated_schedules is not deleted"""
input_data = {
'orchestrated_schedules': {
'scouter': {'test-schedule-id': '2021-01-01'},
'laborious': {'test-schedule-id': '2021-01-01'},
},
**metadata,
}
for client in temporal_manager.temporal_clients.values():
client.list_schedules = AsyncMock(return_value=async_iter())
client.get_schedule_handle = MagicMock()
await temporal_manager.normalize_schedules(input_data)
for client in temporal_manager.temporal_clients.values():
client.get_schedule_handle.assert_not_called()

41
tests/orchestrator/utils/test_orchestrator_functions.py
View File

@@ -1515,3 +1515,44 @@ def test_pi_web_api_scouter_connection_carries_only_the_secret_reference():
assert result['connection']['auth'] == {'type': 'bearer', 'secret': PI_SECRET}
assert 'leaked-credential' not in str(result)
def test_predictions_batch_groups_write_tags_by_server_and_type():
config = {
'workflow_type': 'predictions_batch',
'schedule_name': 'batch',
'model_id': 'model_id',
'model': {'name': 'model_name'},
'query': 'SELECT 1',
'write_tags': [
{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P1'},
{'server_id': '1', 'type': 'prediction', 'addr': 'ns=2;s=P2', 'data_type': 'int'},
# Neither prediction nor confidence: the server is known, the tag is ignored
{'server_id': '2', 'type': 'other', 'addr': 'ns=2;s=O1'},
],
}
result = predictions_batch(config)
assert result['opc_output_config'] == {
'1': {
'prediction_tags': {
'ns=2;s=P1': {'data_type': 'float'},
'ns=2;s=P2': {'data_type': 'int'},
}
},
'2': {},
}
def test_referenced_pi_tag_ids_ignores_malformed_read_tags():
assert referenced_pi_tag_ids([_pi_scouter_pipeline([{'tag_id': 42}])]) == []
def test_resolve_pi_output_tags_ignores_non_dict_entries():
resolved, issues = resolve_pi_output_tags(
{'prediction_tags': ['F1DP-P', None, {'PRED-101': 'F1DP-P'}]}, {}, {}
)
assert resolved == {'prediction_tags': {'PRED-101': 'F1DP-P'}}
assert not any(issues.values())

42
tests/orchestrator/worker/test_worker.py
View File

@@ -3,7 +3,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
from pytest import mark
from orchestrator.worker.worker import main
from orchestrator.worker.worker import main, start_prometheus_server
from orchestrator.workflows.alerts import Alerts
from orchestrator.workflows.orchestrator import Orchestrator
from orchestrator.workflows.reports import Reports
@@ -47,3 +47,43 @@ async def test_main_starts_three_workers_for_expected_workflows(
assert main_workflows == [Orchestrator, Alerts, Reports]
assert worker_mock.run.call_count == 3
@mark.asyncio
@patch('orchestrator.worker.worker.Runtime')
@patch('orchestrator.worker.worker.sys')
@patch('orchestrator.worker.worker.start_http_server')
@patch('orchestrator.worker.worker.NotificationHandler')
@patch('orchestrator.worker.worker.Activities')
@patch('orchestrator.worker.worker.prepare_worker')
@patch('orchestrator.worker.worker.client.Client.connect', new_callable=AsyncMock)
async def test_main_exits_with_error_when_a_worker_fails(
_connect_mock,
prepare_worker_mock,
activities_mock,
notification_handler_mock,
_start_http_server,
sys_mock,
_runtime_mock,
):
"""A worker failure shuts everything down and exits with code 1."""
activities_instance = activities_mock.return_value
activities_instance.connect_to_temporal = AsyncMock()
worker_mock = MagicMock()
worker_mock.run = AsyncMock(side_effect=RuntimeError('worker crashed'))
prepare_worker_mock.return_value = worker_mock
await main()
notification_handler_mock.return_value.shutdown.assert_called_once()
activities_instance.shutdown.assert_called_once()
sys_mock.exit.assert_called_once_with(1)
@patch('orchestrator.worker.worker.os._exit')
@patch('orchestrator.worker.worker.start_http_server', side_effect=OSError('port in use'))
def test_start_prometheus_server_exits_on_failure(_start_http_server, exit_mock):
start_prometheus_server()
exit_mock.assert_called_once_with(1)

27
tests/orchestrator/workflows/test_orchestrator.py
View File

@@ -632,3 +632,30 @@ async def test_run_replay_before_pi_web_api_patch(workflow_mock, _ids, _pi_ids,
schedules_input = _process_schedules_input(workflow_mock)
assert schedules_input['pi_connectors'] == []
assert schedules_input['pi_connector_tags'] == []
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.referenced_tag_ids', return_value=[])
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_without_changes_skips_reports_and_timestamps(workflow_mock, _ids, orchestrator):
"""Nothing created, updated or deleted: no report and no pipeline timestamp activity"""
workflow_mock.patched = MagicMock(return_value=True)
workflow_mock.start_activity_method = AsyncMock(return_value=[])
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
'schedule_name': 'test-schedule-name',
}
await orchestrator.run(input_data)
started = [c.args[0] for c in workflow_mock.start_activity_method.call_args_list]
assert started == [
Activities.normalize_schedules,
Activities.create_collection_with_ttl_index,
Activities.delete_slots,
Activities.update_slots,
Activities.delete_schedules,
Activities.create_schedules,
Activities.update_schedules,
]