diff --git a/orchestrator/utils/opc_slots.py b/orchestrator/utils/opc_slots.py index 5980b2c..f6c8113 100644 --- a/orchestrator/utils/opc_slots.py +++ b/orchestrator/utils/opc_slots.py @@ -149,6 +149,27 @@ def _add_read_tag( entry['names'].append(name) +def _add_pipeline_read_tags( + entries: dict[tuple[str, str], dict[str, Any]], + servers: dict[str, dict[str, Any]], + issues: dict[str, set[str]], + pipeline: dict[str, Any], + connectors: dict[str, dict[str, Any]], + tags_by_id: dict[int, dict[str, Any]], +) -> None: + """Add the read tags of one slot pipeline to entries, reporting the ones left out in issues.""" + for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id): + tag_id = tag_id_of(tag) + if tag_id is None or _invalid_read_tag(tag): + issues['invalid'].add(f'{pipeline["schedule_name"]}/{_read_tag_ref(tag)}') + continue + + located = _locate_read_tag(tag_id, connectors, tags_by_id, issues) + if located is None: + continue + _add_read_tag(entries, servers, pipeline, tag, *located) + + def gather_read_tags( pipelines: list[dict[str, Any]], connectors: dict[str, dict[str, Any]], @@ -201,17 +222,7 @@ def gather_read_tags( if not _is_slot_pipeline(pipeline): issues['invalid'].add(str(pipeline.get('schedule_name'))) continue - - for tag in resolve_tag_names(pipeline.get('read_tags') or [], tags_by_id): - tag_id = tag_id_of(tag) - if tag_id is None or _invalid_read_tag(tag): - issues['invalid'].add(f'{pipeline["schedule_name"]}/{_read_tag_ref(tag)}') - continue - - located = _locate_read_tag(tag_id, connectors, tags_by_id, issues) - if located is None: - continue - _add_read_tag(entries, servers, pipeline, tag, *located) + _add_pipeline_read_tags(entries, servers, issues, pipeline, connectors, tags_by_id) _assign_server_keys(servers) diff --git a/tests/orchestrator/activities/test_mongo_db.py b/tests/orchestrator/activities/test_mongo_db.py index 0e1902a..205ff2a 100644 --- a/tests/orchestrator/activities/test_mongo_db.py +++ b/tests/orchestrator/activities/test_mongo_db.py @@ -1,14 +1,14 @@ from datetime import UTC, datetime from unittest.mock import ANY, AsyncMock, MagicMock, call, patch -from pytest import fixture, mark, raises +import pytest from sientia_do.notifications.models import NotificationLevel from sientia_do.temporal.constants import DATETIME_FORMAT_MS_WITH_TZ from orchestrator.activities.mongo_db import MongoDB -@fixture +@pytest.fixture @patch('orchestrator.activities.mongo_db.MongoDBRepository') def mongo_db(mongo_mock): mongo = MongoDB( @@ -294,7 +294,7 @@ def test_update_pipelines_timestamps_success(now_mock, mongo_db): @patch('orchestrator.activities.mongo_db.now') -@mark.parametrize( +@pytest.mark.parametrize( 'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]] ) def test_update_pipelines_timestamps_all_failed_stamps_nothing(now_mock, mongo_db, pipelines): @@ -373,7 +373,7 @@ def test_update_pipelines_timestamps_failures_notified_together(now_mock, mongo_ Exception('timeout'), ] - with raises(RuntimeError, match='Failed to update 2 of 3 pipelines timestamps'): + with pytest.raises(RuntimeError, match='Failed to update 2 of 3 pipelines timestamps'): mongo_db.update_pipelines_timestamps(input_data) assert mongo_db.mongo_db_repository.update_one.call_count == 3 @@ -462,7 +462,7 @@ def test_delete_pipelines_timestamps_success(now_mock, mongo_db): ) -@mark.parametrize( +@pytest.mark.parametrize( 'pipelines', [[], [{'schedule_name': 'test1', 'namespace': 'test1', 'success': False}]] ) def test_delete_pipelines_timestamps_all_failed_deletes_nothing(mongo_db, pipelines):