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

feat(gateway): slot v2 loader (contract C1)

Browse Source
This commit is contained in:
PedroHMCosme
2026-07-14 10:47:33 -03:00
parent 00f919cec6
commit 44df022e32
4 changed files with 53 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
tests/unit/io/__init__.py Normal file
View File

28
tests/unit/io/test_slots.py Normal file
View File

@@ -0,0 +1,28 @@
from ingestor.io.slots import parse_slot
RAW = {'server1': {'server_id': '1', 'url': 'opc.tcp://h:4840', 'use_security': True,
'subscription_period_ms': 1000, 'tags': {'ns=2;i=2': {'frequency': '15000',
'active_read': {'enabled': True, 'cycle_multiple': 3},
'names': [{'collection': 'raw_x', 'tag_name': 'Counter', 'model_name': 'm1', 'model_id': '1'},
{'collection': 'raw_y', 'tag_name': 'Contador', 'model_name': 'm2', 'model_id': '2'}]}}}}
def test_parse_slot_multi_name():
slot = parse_slot(RAW)
tag = slot['server1']['tags']['ns=2;i=2']
assert len(tag['names']) == 2
assert tag['active_read']['cycle_multiple'] == 3
def test_parse_slot_defaults_active_read_disabled():
raw = {'s': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=1': {
'frequency': '1000', 'names': [{'collection': 'c', 'tag_name': 't', 'model_name': 'm', 'model_id': '1'}]}}}}
tag = parse_slot(raw)['s']['tags']['ns=2;i=1']
assert tag['active_read'] == {'enabled': False}
def test_parse_slot_rejects_tag_without_names():
raw = {'s': {'server_id': '1', 'url': 'u', 'tags': {'ns=2;i=1': {'frequency': '1000'}}}}
import pytest
with pytest.raises((KeyError, ValueError)):
parse_slot(raw)