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

SIENTIAPDE-1974: feat(slots): write the OPC slot v2 from pipelines and the tag database

Browse Source 
The Orchestrator now writes the slot v2 (contract C1) the OPC Gateway reads:
one entry per (server, address) with a names[] list, so an address read by N
pipelines is read once and written to N collections (fixes "one name per
address", where the second model's scouter discarded everything).

- O1: load connectors (opc_ua, Active) and connector_tags (Read, Active) in
  parallel with the current loads. Queries are optional in the input with
  defaults in code (D9); empty or missing collections load nothing. The loads
  sit behind workflow.patched('SIENTIAPDE-1974-tag-database'), so a run in
  flight during a deploy replays without them.
- O2: gather_read_tags groups by (server, address). A read tag with tag_id
  resolves address/connector from connector_tags and the server from connectors
  (RF3); without tag_id it keeps server_id + tag_address + opc-servers (RF4),
  so every environment without a tag database keeps collecting (P4).
- O3/SC1: resolve_tag_names is the single source of tag names for both the
  scouter's model_tags and the slot's names[].
- O4: build_tag_config emits server_id/name/url/server_uri/use_security/
  subscription_period_ms + tags{frequency, active_read, names}; no certificate
  paths (the gateway takes the certificate from MinIO). frequency = lowest
  among pipelines; subscription period = connector value, else min/2 as before;
  active_read off unless a pipeline enables it (lowest cycle_multiple, D3).
- O5: process_slots splits by address (names of an address never split) and
  notifies, once per run, unknown tag_id, missing/inactive connector and
  unknown opc-servers server_id.
- O6: create_slot_config writes only new or changed slots; the slot is built
  in a stable order, so unchanged pipelines do not make the gateway
  re-subscribe (and the "Inserted slots" report stops firing every run).
- O7: unit tests for multi-name, repeated address, unknown tag_id, inactive
  connector, mixed tags, server-name clash, active_read, slot division,
  stability, SC1 and the replay path (184 tests).
- O8: README documents the slot v2 contract and the new optional inputs.
- test_email.test_close: count only the instance under test; an Email left by
  another test can be garbage-collected (Email.__del__ -> close) while the
  patch is active, which the new tests' allocation pattern exposed.

Dry run with the dev data (scouter-sin-approx): the v2 slot is equivalent to
the current v1 slot as seen by the gateway's parse_slot (same server, addresses,
collections, names, frequency and subscription period).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-10-06 15:44:07 -03:00
parent 6334be2edf
commit efa0d9411b
9 changed files with 1307 additions and 414 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -85,7 +85,9 @@ def test_close(sientia_monitoring_mock, email):
email.close()
email.server.quit.assert_called_once()
sientia_monitoring_mock.shutdown.assert_called_once()
# Count only this instance: an Email left by another test may be garbage-collected
# (Email.__del__ -> close) while the patch is active
assert sientia_monitoring_mock.shutdown.call_args_list.count(call(email)) == 1
def test_build_email_html(email):

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

@@ -218,109 +218,348 @@ def test_process_schedules_with_invalid_workflow_type(formatters):
mock_scouter.assert_any_call(pipelines[2])
@patch(
'orchestrator.activities.formatters.gather_read_tags',
return_value={
'1:test_tag_address': {
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
'topics': ['raw_test_schedule'],
'frequency': 1000,
},
'2:test_tag_address2': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address2',
'topics': ['raw_test_schedule2'],
'frequency': 1000,
},
'2:test_tag_address3': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address3',
'topics': ['raw_test_schedule2'],
'frequency': 1000,
},
},
)
@patch('orchestrator.activities.formatters.build_tag_config')
def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters):
def _scouter_pipeline(schedule_name, model_id, read_tags):
return {
'workflow_type': 'scouter',
'schedule_name': schedule_name,
'model_id': model_id,
'model': {'name': f'model {model_id}'},
'read_tags': read_tags,
}
slot_opc_servers = [
{'id': '1', 'server_name': 'kepserver', 'url': 'opc.tcp://kep:49320', 'uri': 'urn:kep'},
]
def test_process_slots(formatters):
"""Slot v2 split by address: the names of an address never land in different slots"""
pipelines = [
_scouter_pipeline(
'sin',
'm1',
[
{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000},
{'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
],
),
_scouter_pipeline(
'cos',
'm2',
[
{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 500},
{'server_id': '1', 'tag_address': 'ns=2;s=C', 'tag_name': 'c', 'frequency': 4000},
],
),
]
input_data = {
'opc_servers': [
{'id': '1', 'server_name': 'test_server_name', 'url': 'test_url', 'uri': 'test_uri'},
{'id': '2', 'server_name': 'test_server_name2', 'url': 'test_url2', 'uri': 'test_uri2'},
],
'active_ingestors': ['test_active_ingestor1', 'test_active_ingestor2'],
'pipelines': 'test_gather_read_tags',
'opc_servers': slot_opc_servers,
'active_ingestors': ['heartbeat:ingestor:1', 'heartbeat:ingestor:2'],
'pipelines': pipelines,
**metadata,
}
expected_opc_servers = {
result = formatters.process_slots(input_data)
server = {
'server_id': '1',
'name': 'kepserver',
'url': 'opc.tcp://kep:49320',
'server_uri': 'urn:kep',
'use_security': False,
}
assert result == {
'1': {
'id': '1',
'server_name': 'test_server_name',
'url': 'test_url',
'uri': 'test_uri',
'kepserver': {
**server,
'subscription_period_ms': 250,
'tags': {
'ns=2;s=A': {
'frequency': 500,
'active_read': {'enabled': False},
'names': [
{
'collection': 'raw_cos',
'tag_name': 'x',
'model_id': 'm2',
'model_name': 'model m2',
},
{
'collection': 'raw_sin',
'tag_name': 'a',
'model_id': 'm1',
'model_name': 'model m1',
},
],
},
'ns=2;s=B': {
'frequency': 1000,
'active_read': {'enabled': False},
'names': [
{
'collection': 'raw_sin',
'tag_name': 'b',
'model_id': 'm1',
'model_name': 'model m1',
},
],
},
},
}
},
'2': {
'id': '2',
'server_name': 'test_server_name2',
'url': 'test_url2',
'uri': 'test_uri2',
'kepserver': {
**server,
'subscription_period_ms': 2000,
'tags': {
'ns=2;s=C': {
'frequency': 4000,
'active_read': {'enabled': False},
'names': [
{
'collection': 'raw_cos',
'tag_name': 'c',
'model_id': 'm2',
'model_name': 'model m2',
},
],
},
},
}
},
}
formatters.send_notification.assert_not_called()
slot_mock = {
'test_server_name': {
'server_id': '1',
'name': 'test_server_name',
'url': 'test_url',
'server_uri': 'test_uri',
'cert_path': None,
'private_key_path': None,
'server_cert_path': None,
}
def test_process_slots_without_ingestors_builds_one_slot(formatters):
pipelines = [
_scouter_pipeline(
'sin',
'm1',
[{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
)
]
input_data = {
'opc_servers': slot_opc_servers,
'active_ingestors': [],
'pipelines': pipelines,
**metadata,
}
mock_build_tag_config.return_value = (slot_mock, ['2'])
result = formatters.process_slots(input_data)
tags = list(mock_gather_read_tags.return_value.values())
assert list(result) == ['1']
assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
mock_gather_read_tags.assert_called_once_with(input_data['pipelines'])
mock_build_tag_config.assert_has_calls(
[
call(tags[:2], expected_opc_servers),
call(tags[2:], expected_opc_servers),
]
)
def test_process_slots_more_ingestors_than_addresses(formatters):
"""Extra ingestors get empty slots, as before"""
pipelines = [
_scouter_pipeline(
'sin',
'm1',
[{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
)
]
input_data = {
'opc_servers': slot_opc_servers,
'active_ingestors': ['i1', 'i2', 'i3'],
'pipelines': pipelines,
**metadata,
}
result = formatters.process_slots(input_data)
assert list(result) == ['1', '2', '3']
assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
assert result['2'] == {}
assert result['3'] == {}
def test_process_slots_with_tag_database(formatters):
"""A read tag with tag_id takes address, name and server from the tag database"""
pipelines = [
_scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}]),
]
input_data = {
'opc_servers': slot_opc_servers,
'connectors': [
{
'connector_id': 3,
'type': 'opc_ua',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': True,
'subscription_period_ms': 1500,
'status': 'Active',
}
],
'connector_tags': [
{'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'}
],
'active_ingestors': ['i1'],
'pipelines': pipelines,
**metadata,
}
result = formatters.process_slots(input_data)
assert result == {
'1': {
'plant': {
'server_id': '3',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': True,
'subscription_period_ms': 1500,
'tags': {
'ns=3;i=1001': {
'frequency': 1000,
'active_read': {'enabled': False},
'names': [
{
'collection': 'raw_sin',
'tag_name': 'Pressure',
'model_id': 'm1',
'model_name': 'model m1',
}
],
}
},
}
}
}
def test_process_slots_notifies_unresolved_tags_once(formatters):
"""Each kind of unresolved read tag is notified once per run, not once per slot"""
pipelines = [
_scouter_pipeline(
'sin',
'm1',
[
{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000},
{'server_id': '9', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
{'tag_id': 404, 'frequency': 1000},
{'tag_id': 8, 'frequency': 1000},
],
),
]
input_data = {
'opc_servers': slot_opc_servers,
'connectors': [],
'connector_tags': [
{'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=8', 'tag_name': 'Level'}
],
'active_ingestors': ['i1', 'i2'],
'pipelines': pipelines,
**metadata,
}
result = formatters.process_slots(input_data)
assert list(result['1']['kepserver']['tags']) == ['ns=2;s=A']
assert result['2'] == {}
formatters.send_notification.assert_has_calls(
[
call(
metadata=metadata['metadata'],
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message='Servers 2 not found in opc_servers',
message='Servers 9 not found in opc_servers',
block='orchestrator',
level=NotificationLevel.ERROR,
),
call(
metadata=metadata['metadata'],
notification_id='ORCHESTRATOR_SERVER_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message='Servers 2 not found in opc_servers',
notification_id='ORCHESTRATOR_TAG_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message='Tags 404 not found among the active Read tags of connector_tags',
block='orchestrator',
level=NotificationLevel.ERROR,
),
call(
metadata=metadata['metadata'],
notification_id='ORCHESTRATOR_CONNECTOR_NOT_FOUND_DURING_SLOT_CONFIGURATION',
message='Connectors 5 not found or not active in connectors',
block='orchestrator',
level=NotificationLevel.ERROR,
),
]
)
assert formatters.send_notification.call_count == 3
assert result == {
'1': slot_mock,
'2': slot_mock,
def test_process_slots_is_stable_across_pipeline_order(formatters):
"""The same pipelines in another order give the same slots (O6)"""
pipelines = [
_scouter_pipeline(
'sin',
'm1',
[{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'a', 'frequency': 1000}],
),
_scouter_pipeline(
'cos',
'm2',
[
{'server_id': '1', 'tag_address': 'ns=2;s=B', 'tag_name': 'b', 'frequency': 1000},
{'server_id': '1', 'tag_address': 'ns=2;s=A', 'tag_name': 'x', 'frequency': 1000},
],
),
]
def run(order):
return formatters.process_slots(
{
'opc_servers': slot_opc_servers,
'active_ingestors': ['i1', 'i2'],
'pipelines': order,
**metadata,
}
)
assert json.dumps(run(pipelines)) == json.dumps(run(list(reversed(pipelines))))
def test_process_schedules_names_scouter_tags_like_the_slot(formatters):
"""model_tags and the slot's names[] take the tag name from the same source (SC1)"""
pipeline = {
**_scouter_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000, 'aggr_func': 'avg'}]),
'updated_at': '2021-01-01',
}
connector_tags = [
{'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'}
]
connectors = [
{
'connector_id': 3,
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
}
]
schedules = formatters.process_schedules(
{'pipelines': [pipeline], 'connector_tags': connector_tags, **metadata}
)
slots = formatters.process_slots(
{
'opc_servers': [],
'connectors': connectors,
'connector_tags': connector_tags,
'active_ingestors': [],
'pipelines': [pipeline],
**metadata,
}
)
model_tags = schedules['scouter']['sin']['model_tags']
names = slots['1']['plant']['tags']['ns=3;i=1001']['names']
assert list(model_tags) == ['Pressure']
assert model_tags['Pressure']['aggr_func'] == 'avg'
assert [name['tag_name'] for name in names] == ['Pressure']
assert names[0]['collection'] == schedules['scouter']['sin']['topic']
def test_format_schedule_config(formatters):
@@ -410,6 +649,31 @@ def test_create_slot_config(formatters):
}
def test_create_slot_config_skips_unchanged_slots(formatters):
"""Only new or changed slots are written (O6); current slots come keyed by Redis key"""
slot_v2 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'names': []}}}}
slot_v1 = {'kepserver': {'server_id': '1', 'tags': {'ns=2;s=A': {'topics': ['raw_x']}}}}
input_data = {
'current_slot_config': {
'slot:opc_tags:1': slot_v2,
'slot:opc_tags:2': slot_v1,
},
'slot_config': {'1': slot_v2, '2': slot_v2, '3': {}},
}
result = formatters.create_slot_config(input_data)
assert result == {'to_delete': [], 'to_insert': {'2': slot_v2, '3': {}}}
def test_create_slot_config_without_current_slots(formatters):
slot_config = {'1': {'kepserver': {'tags': {}}}, '2': {}}
result = formatters.create_slot_config({'current_slot_config': {}, 'slot_config': slot_config})
assert result == {'to_delete': [], 'to_insert': slot_config}
def test_send_success_report(formatters):
formatters.send_success_report(
metadata=metadata,

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

@@ -6,11 +6,13 @@ from orchestrator.utils.orchestrator_functions import (
common_config,
drift,
gather_read_tags,
index_connector_tags,
minimal_retrain,
overlap_filter_config,
pi_web_api_scouter,
predictions_batch,
process_path_priority,
resolve_tag_names,
scouter,
simple_metrics,
)
@@ -311,249 +313,427 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi
assert result == expected
def test_gather_read_tags():
pipelines = [
{
'workflow_type': 'scouter',
'schedule_name': 'test_schedule',
'read_tags': [
{
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
}
],
},
{
'workflow_type': 'scouter',
'schedule_name': 'test_schedule2',
'read_tags': [
{
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address2',
},
{
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address3',
},
],
},
]
result = gather_read_tags(pipelines)
expected = {
'1:test_tag_address': {
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
'topics': ['raw_test_schedule'],
},
'2:test_tag_address2': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address2',
'topics': ['raw_test_schedule2'],
},
'2:test_tag_address3': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address3',
'topics': ['raw_test_schedule2'],
},
def _pipeline(schedule_name, model_id, read_tags, workflow_type='scouter'):
return {
'workflow_type': workflow_type,
'schedule_name': schedule_name,
'model_id': model_id,
'model': {'name': f'model {model_id}'},
'read_tags': read_tags,
}
assert result == expected
def _legacy_tag(address, name, frequency=1000, server_id='1', **extra):
return {
'server_id': server_id,
'tag_address': address,
'tag_name': name,
'frequency': frequency,
**extra,
}
def _name(schedule_name, tag_name, model_id):
return {
'collection': f'raw_{schedule_name}',
'tag_name': tag_name,
'model_id': model_id,
'model_name': f'model {model_id}',
}
opc_servers = {
'1': {
'id': '1',
'server_name': 'kepserver',
'url': 'opc.tcp://kep:49320',
'uri': 'urn:kep',
'cert_path': 'certs/client.pem',
},
'2': {'id': '2', 'server_name': 'simulator', 'url': 'opc.tcp://sim:4840', 'uri': 'urn:sim'},
}
connectors = {
'3': {
'connector_id': 3,
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': True,
'subscription_period_ms': 1500,
}
}
tags_by_id = {
'7': {'tag_id': 7, 'connector_id': 3, 'address': 'ns=3;i=1001', 'tag_name': 'Pressure'},
'8': {'tag_id': 8, 'connector_id': 5, 'address': 'ns=3;i=1002', 'tag_name': 'Level'},
}
def _gather(pipelines, servers=None, connector_index=None, tag_index=None):
return gather_read_tags(
pipelines,
opc_servers if servers is None else servers,
connectors if connector_index is None else connector_index,
tags_by_id if tag_index is None else tag_index,
)
def test_index_connector_tags():
tags = [{'tag_id': 7, 'tag_name': 'a'}, {'tag_id': '8', 'tag_name': 'b'}]
assert index_connector_tags(tags) == {'7': tags[0], '8': tags[1]}
def test_resolve_tag_names():
read_tags = [
{'tag_id': 7},
{'tag_id': '7', 'tag_name': 'own name'},
{'tag_id': 404},
{'tag_name': 'legacy'},
]
assert resolve_tag_names(read_tags, tags_by_id) == [
{'tag_id': 7, 'tag_name': 'Pressure'},
{'tag_id': '7', 'tag_name': 'own name'},
{'tag_id': 404},
{'tag_name': 'legacy'},
]
def test_gather_read_tags_multi_name():
"""One address read by two pipelines with different names: one entry, two names"""
pipelines = [
_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter', frequency=15000)]),
_pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Contador', frequency=5000)]),
]
entries, issues = _gather(pipelines)
assert len(entries) == 1
assert entries[0]['address'] == 'ns=2;s=A'
assert entries[0]['frequency'] == 5000
assert entries[0]['names'] == [_name('cos', 'Contador', 'm2'), _name('sin', 'Counter', 'm1')]
assert issues == {'tags': set(), 'connectors': set(), 'servers': set()}
def test_gather_read_tags_same_name_in_two_pipelines():
"""The same name read by two pipelines still goes to both collections"""
pipelines = [
_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'Counter')]),
_pipeline('cos', 'm2', [_legacy_tag('ns=2;s=A', 'Counter')]),
]
entries, _issues = _gather(pipelines)
assert entries[0]['names'] == [_name('cos', 'Counter', 'm2'), _name('sin', 'Counter', 'm1')]
def test_gather_read_tags_duplicate_tag_in_one_pipeline():
pipelines = [
_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=A', 'a')]),
]
entries, _issues = _gather(pipelines)
assert entries[0]['names'] == [_name('sin', 'a', 'm1')]
def test_gather_read_tags_same_address_on_two_servers():
pipelines = [
_pipeline(
'sin',
'm1',
[_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=A', 'a')],
),
]
entries, _issues = _gather(pipelines)
assert [(e['server']['server_id'], e['address']) for e in entries] == [
('1', 'ns=2;s=A'),
('2', 'ns=2;s=A'),
]
def test_gather_read_tags_legacy_server():
"""Without tag_id: opc-servers fields, no certificate paths, security from cert_path"""
entries, _issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')])])
assert entries[0]['server'] == {
'server_id': '1',
'name': 'kepserver',
'url': 'opc.tcp://kep:49320',
'server_uri': 'urn:kep',
'use_security': True,
'subscription_period_ms': None,
'from_tag_database': False,
'key': 'kepserver',
}
def test_gather_read_tags_numeric_server_id():
"""A numeric server_id in the pipeline still finds its opc-servers document"""
entries, issues = _gather([_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id=1)])])
assert entries[0]['server']['server_id'] == '1'
assert not issues['servers']
def test_gather_read_tags_with_tag_database():
"""With tag_id: address, name and server come from connector_tags + connectors"""
entries, _issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 7, 'frequency': 1000}])])
assert entries == [
{
'server': {
'server_id': '3',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': True,
'subscription_period_ms': 1500,
'from_tag_database': True,
'key': 'plant',
},
'address': 'ns=3;i=1001',
'frequency': 1000,
'active_read': {'enabled': False},
'names': [_name('sin', 'Pressure', 'm1')],
}
]
def test_gather_read_tags_pipeline_name_wins_over_tag_database():
entries, _issues = _gather(
[_pipeline('sin', 'm1', [{'tag_id': 7, 'tag_name': 'P1', 'frequency': 1000}])]
)
assert entries[0]['names'] == [_name('sin', 'P1', 'm1')]
def test_gather_read_tags_unknown_tag_id():
entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 404, 'frequency': 1000}])])
assert entries == []
assert issues == {'tags': {'404'}, 'connectors': set(), 'servers': set()}
def test_gather_read_tags_inactive_connector():
"""Connector 5 is not among the active connectors: its tag is left out"""
entries, issues = _gather([_pipeline('sin', 'm1', [{'tag_id': 8, 'frequency': 1000}])])
assert entries == []
assert issues == {'tags': set(), 'connectors': {'5'}, 'servers': set()}
def test_gather_read_tags_unknown_server():
entries, issues = _gather(
[_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a', server_id='9')])]
)
assert entries == []
assert issues == {'tags': set(), 'connectors': set(), 'servers': {'9'}}
def test_gather_read_tags_mixed_with_and_without_tag_id():
"""Both paths in one run; the same node via tag_id and via tag_address merges"""
connector_1 = {
'connector_id': 1,
'name': 'kepserver',
'url': 'opc.tcp://kep:49320',
'server_uri': 'urn:kep',
'use_security': True,
'subscription_period_ms': 2000,
}
tag_index = {
'20': {'tag_id': 20, 'connector_id': 1, 'address': 'ns=2;s=A', 'tag_name': 'A from db'},
}
pipelines = [
_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a'), _legacy_tag('ns=2;s=B', 'b')]),
_pipeline(
'cos', 'm2', [{'tag_id': 20, 'frequency': 500}, {'tag_id': 7, 'frequency': 1000}]
),
]
entries, issues = _gather(
pipelines,
connector_index={**connectors, '1': connector_1},
tag_index={**tags_by_id, **tag_index},
)
assert [(e['server']['server_id'], e['address']) for e in entries] == [
('1', 'ns=2;s=A'),
('1', 'ns=2;s=B'),
('3', 'ns=3;i=1001'),
]
assert entries[0]['names'] == [_name('cos', 'A from db', 'm2'), _name('sin', 'a', 'm1')]
assert entries[0]['frequency'] == 500
# Same server_id in connectors and opc-servers: the connectors fields win
assert entries[0]['server']['subscription_period_ms'] == 2000
assert entries[1]['server'] is entries[0]['server']
assert issues == {'tags': set(), 'connectors': set(), 'servers': set()}
def test_gather_read_tags_server_name_clash():
"""Two servers with the same name get distinct slot keys, stable by server_id"""
servers = {
'1': opc_servers['1'],
'2': {**opc_servers['2'], 'server_name': 'kepserver'},
}
pipelines = [
_pipeline(
'sin',
'm1',
[_legacy_tag('ns=2;s=A', 'a', server_id='2'), _legacy_tag('ns=2;s=B', 'b')],
),
]
entries, _issues = _gather(pipelines, servers=servers)
assert [e['server']['key'] for e in entries] == ['kepserver', 'kepserver:2']
assert [e['server']['name'] for e in entries] == ['kepserver', 'kepserver']
def test_gather_read_tags_active_read():
"""Active read: on if any pipeline asks, with the lowest cycle_multiple (D3)"""
pipelines = [
_pipeline('p1', 'm1', [_legacy_tag('ns=2;s=A', 'a')]),
_pipeline(
'p2',
'm2',
[_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 4})],
),
_pipeline(
'p3',
'm3',
[_legacy_tag('ns=2;s=A', 'a', active_read={'enabled': True, 'cycle_multiple': 2})],
),
_pipeline(
'p4',
'm4',
[_legacy_tag('ns=2;s=B', 'b', active_read={'enabled': False, 'cycle_multiple': 1})],
),
_pipeline('p5', 'm5', [_legacy_tag('ns=2;s=C', 'c', active_read={'enabled': True})]),
]
entries, _issues = _gather(pipelines)
assert [e['active_read'] for e in entries] == [
{'enabled': True, 'cycle_multiple': 2},
{'enabled': False},
{'enabled': True, 'cycle_multiple': 1},
]
def test_gather_read_tags_filters_non_scouter_workflows():
"""Test that gather_read_tags only processes scouter workflow types and ignores others"""
"""Only scouter pipelines feed the slots"""
pipelines = [
{
'workflow_type': 'scouter',
'schedule_name': 'test_scouter_schedule',
'read_tags': [
{
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
}
],
},
{
'workflow_type': 'predictions_batch',
'schedule_name': 'test_predictions_schedule',
'read_tags': [
{
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address_predictions',
}
],
},
{
'workflow_type': 'minimal_retrain',
'schedule_name': 'test_retrain_schedule',
'read_tags': [
{
'server_id': '3',
'server_name': 'test_server_name3',
'tag_address': 'test_tag_address_retrain',
}
],
},
{
'workflow_type': 'pi_web_api_scouter',
'schedule_name': 'test_pi_web_api_schedule',
'read_tags': [
{
'server_id': '4',
'server_name': 'test_server_name4',
'tag_address': 'test_tag_address_pi_web_api',
}
],
},
{
'workflow_type': 'drift',
'schedule_name': 'test_drift_schedule',
'read_tags': [
{
'server_id': '5',
'server_name': 'test_server_name5',
'tag_address': 'test_tag_address_drift',
}
],
},
{
'workflow_type': 'scouter',
'schedule_name': 'test_scouter_schedule2',
'read_tags': [
{
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address2',
}
],
},
_pipeline('sin', 'm1', [_legacy_tag('ns=2;s=A', 'a')]),
_pipeline('pred', 'm2', [_legacy_tag('ns=2;s=P', 'p')], workflow_type='predictions_batch'),
_pipeline('retrain', 'm3', [_legacy_tag('ns=2;s=R', 'r')], workflow_type='minimal_retrain'),
_pipeline('drift', 'm4', [_legacy_tag('ns=2;s=D', 'd')], workflow_type='drift'),
]
result = gather_read_tags(pipelines)
entries, _issues = _gather(pipelines)
# Only scouter workflow types should be included
expected = {
'1:test_tag_address': {
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
'topics': ['raw_test_scouter_schedule'],
},
'1:test_tag_address2': {
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address2',
'topics': ['raw_test_scouter_schedule2'],
},
}
assert [e['address'] for e in entries] == ['ns=2;s=A']
assert result == expected
# Ensure non-scouter pipelines are not included
assert '2:test_tag_address_predictions' not in result
assert '3:test_tag_address_retrain' not in result
assert '4:test_tag_address_pi_web_api' not in result
assert '5:test_tag_address_drift' not in result
def test_gather_read_tags_without_read_tags():
pipeline = _pipeline('sin', 'm1', [])
del pipeline['read_tags']
assert _gather([pipeline]) == ([], {'tags': set(), 'connectors': set(), 'servers': set()})
def test_build_tag_config():
tags = [
{
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
'frequency': 1000,
},
{
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address2',
'frequency': 1000,
},
{
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address3',
'frequency': 300,
},
{
'server_id': '3',
'server_name': 'test_server_name3',
'tag_address': 'test_tag_address4',
'frequency': 200,
},
"""Slot v2 (contract C1): no certificate paths, subscription period per server"""
pipelines = [
_pipeline(
'sin',
'm1',
[
_legacy_tag('ns=2;s=A', 'a', frequency=1000),
_legacy_tag('ns=2;s=B', 'b', frequency='300'),
_legacy_tag('ns=2;s=C', 'c', frequency=200, server_id='2'),
{'tag_id': 7, 'frequency': 100},
],
)
]
entries, _issues = _gather(pipelines)
opc_servers = {
'1': {
'server_name': 'test_server_name',
'url': 'test_url',
'uri': 'test_uri',
},
'2': {
'server_name': 'test_server_name2',
'url': 'test_url2',
'uri': 'test_uri2',
},
}
result = build_tag_config(entries)
result = build_tag_config(tags, opc_servers)
expected = {
'test_server_name': {
assert result == {
'kepserver': {
'server_id': '1',
'name': 'test_server_name',
'url': 'test_url',
'server_uri': 'test_uri',
'cert_path': None,
'private_key_path': None,
'server_cert_path': None,
'subscription_period_ms': 500,
'tags': {
'test_tag_address': {
'server_id': '1',
'server_name': 'test_server_name',
'tag_address': 'test_tag_address',
'frequency': 1000,
},
},
},
'test_server_name2': {
'server_id': '2',
'name': 'test_server_name2',
'url': 'test_url2',
'server_uri': 'test_uri2',
'cert_path': None,
'private_key_path': None,
'server_cert_path': None,
'name': 'kepserver',
'url': 'opc.tcp://kep:49320',
'server_uri': 'urn:kep',
'use_security': True,
'subscription_period_ms': 150,
'tags': {
'test_tag_address2': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address2',
'ns=2;s=A': {
'frequency': 1000,
'active_read': {'enabled': False},
'names': [_name('sin', 'a', 'm1')],
},
'test_tag_address3': {
'server_id': '2',
'server_name': 'test_server_name2',
'tag_address': 'test_tag_address3',
'frequency': 300,
'ns=2;s=B': {
'frequency': '300',
'active_read': {'enabled': False},
'names': [_name('sin', 'b', 'm1')],
},
},
},
'simulator': {
'server_id': '2',
'name': 'simulator',
'url': 'opc.tcp://sim:4840',
'server_uri': 'urn:sim',
'use_security': False,
'subscription_period_ms': 100,
'tags': {
'ns=2;s=C': {
'frequency': 200,
'active_read': {'enabled': False},
'names': [_name('sin', 'c', 'm1')],
},
},
},
'plant': {
'server_id': '3',
'name': 'plant',
'url': 'opc.tcp://plant:4840',
'server_uri': 'urn:plant',
'use_security': True,
'subscription_period_ms': 1500,
'tags': {
'ns=3;i=1001': {
'frequency': 100,
'active_read': {'enabled': False},
'names': [_name('sin', 'Pressure', 'm1')],
},
},
},
}
assert result == (expected, ['3'])
def test_build_tag_config_empty():
assert build_tag_config([]) == {}
def test_scouter_skips_tags_without_name():
"""A tag_id the tag database cannot resolve has no name and stays out of model_tags"""
config = {
'workflow_type': 'scouter',
'schedule_name': 'sin',
'model_id': 'm1',
'model': {'name': 'model m1'},
'read_tags': [{'tag_name': 'a'}, {'tag_id': 404}],
}
assert list(scouter(config)['model_tags']) == ['a']
def test_base_scouter():

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

@@ -1,9 +1,14 @@
from unittest.mock import ANY, AsyncMock, call, patch
from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
from pytest import fixture, mark
from orchestrator.activities.activities import Activities
from orchestrator.workflows.orchestrator import Orchestrator
from orchestrator.workflows.orchestrator import (
DEFAULT_CONNECTOR_TAGS_QUERY,
DEFAULT_CONNECTORS_QUERY,
TAG_DATABASE_PATCH,
Orchestrator,
)
@fixture
@@ -24,6 +29,7 @@ metadata = {
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run(workflow_mock, orchestrator):
workflow_mock.patched = MagicMock(return_value=True)
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
@@ -109,11 +115,34 @@ async def test_run(workflow_mock, orchestrator):
]
)
workflow_mock.patched.assert_called_once_with(TAG_DATABASE_PATCH)
workflow_mock.start_local_activity_method.assert_has_calls(
[
call(
Activities.find_documents_in_mongodb,
{**metadata, 'query': DEFAULT_CONNECTORS_QUERY},
retry_policy=ANY,
start_to_close_timeout=ANY,
),
call(
Activities.find_documents_in_mongodb,
{**metadata, 'query': DEFAULT_CONNECTOR_TAGS_QUERY},
retry_policy=ANY,
start_to_close_timeout=ANY,
),
]
)
workflow_mock.start_local_activity_method.assert_has_calls(
[
call(
Activities.process_schedules,
{**metadata, 'pipelines': workflow_mock.start_local_activity_method.return_value},
{
**metadata,
'pipelines': workflow_mock.start_local_activity_method.return_value,
'connector_tags': workflow_mock.start_local_activity_method.return_value,
},
retry_policy=ANY,
start_to_close_timeout=ANY,
)
@@ -127,6 +156,8 @@ async def test_run(workflow_mock, orchestrator):
{
**metadata,
'opc_servers': workflow_mock.start_local_activity_method.return_value,
'connectors': workflow_mock.start_local_activity_method.return_value,
'connector_tags': workflow_mock.start_local_activity_method.return_value,
'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.start_local_activity_method.return_value,
},
@@ -346,3 +377,78 @@ async def test_run(workflow_mock, orchestrator):
)
]
)
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_with_tag_database_queries_from_input(workflow_mock, orchestrator):
"""The input overrides the default tag database queries (SIENTIAPDE-1974, D9)"""
workflow_mock.patched = MagicMock(return_value=True)
connectors_query = {'collection': 'connectors', 'filters': {'type': 'opc_ua'}}
connector_tags_query = {'collection': 'connector_tags', 'filters': {'category': 'Read'}}
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
'connectors_query': connectors_query,
'connector_tags_query': connector_tags_query,
'schedule_name': 'test-schedule-name',
}
await orchestrator.run(input_data)
workflow_mock.start_local_activity_method.assert_has_calls(
[
call(
Activities.find_documents_in_mongodb,
{**metadata, 'query': connectors_query},
retry_policy=ANY,
start_to_close_timeout=ANY,
),
call(
Activities.find_documents_in_mongodb,
{**metadata, 'query': connector_tags_query},
retry_policy=ANY,
start_to_close_timeout=ANY,
),
]
)
@mark.asyncio
@patch('orchestrator.workflows.orchestrator.workflow', new_callable=AsyncMock)
async def test_run_replay_before_tag_database_patch(workflow_mock, orchestrator):
"""A run started before the patch replays without the tag database loads"""
workflow_mock.patched = MagicMock(return_value=False)
input_data = {
'pipelines_query': 'SELECT * FROM bucket',
'opc_servers_query': 'SELECT * FROM servers',
'schedule_name': 'test-schedule-name',
}
await orchestrator.run(input_data)
queries = [
c.args[1].get('query')
for c in workflow_mock.start_local_activity_method.call_args_list
if c.args[0] is Activities.find_documents_in_mongodb
]
assert DEFAULT_CONNECTORS_QUERY not in queries
assert DEFAULT_CONNECTOR_TAGS_QUERY not in queries
workflow_mock.start_local_activity_method.assert_has_calls(
[
call(
Activities.process_slots,
{
**metadata,
'opc_servers': workflow_mock.start_local_activity_method.return_value,
'connectors': [],
'connector_tags': [],
'active_ingestors': workflow_mock.start_local_activity_method.return_value,
'pipelines': workflow_mock.start_local_activity_method.return_value,
},
retry_policy=ANY,
start_to_close_timeout=ANY,
)
]
)