diff --git a/e2e/helpers.py b/e2e/helpers.py index 451148b..0f3b0e1 100644 --- a/e2e/helpers.py +++ b/e2e/helpers.py @@ -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, diff --git a/e2e/scenario_inputs/orchestrator_pi_web_api.json b/e2e/scenario_inputs/orchestrator_pi_web_api.json new file mode 100644 index 0000000..e3835fe --- /dev/null +++ b/e2e/scenario_inputs/orchestrator_pi_web_api.json @@ -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"] +} diff --git a/e2e/scenarios.md b/e2e/scenarios.md index aa3e9af..58cf99f 100644 --- a/e2e/scenarios.md +++ b/e2e/scenarios.md @@ -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 diff --git a/e2e/test_orchestrator_main_workflow.py b/e2e/test_orchestrator_main_workflow.py index 86d0091..04cf76e 100644 --- a/e2e/test_orchestrator_main_workflow.py +++ b/e2e/test_orchestrator_main_workflow.py @@ -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 diff --git a/orchestrator/activities/email.py b/orchestrator/activities/email.py index 6fa0b08..f40e969 100644 --- a/orchestrator/activities/email.py +++ b/orchestrator/activities/email.py @@ -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: diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 3c9d9ba..2d7254c 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -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) diff --git a/tests/orchestrator/activities/test_email.py b/tests/orchestrator/activities/test_email.py index 99e0fd8..e74f5a9 100644 --- a/tests/orchestrator/activities/test_email.py +++ b/tests/orchestrator/activities/test_email.py @@ -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 + ) diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index d7896af..3ed8f3e 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -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']] diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 6feda41..e18cdcb 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -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() diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index f3c44ed..51dd084 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -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() diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index a34c80a..3b77555 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -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()) diff --git a/tests/orchestrator/worker/test_worker.py b/tests/orchestrator/worker/test_worker.py index d23cbb7..ced676c 100644 --- a/tests/orchestrator/worker/test_worker.py +++ b/tests/orchestrator/worker/test_worker.py @@ -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) diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index 6422d5e..7fc3278 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -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, + ]