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

style: ruff format (new quality gate check picked up from main)

Browse Source
This commit is contained in:
PedroHMCosme
2026-09-22 11:26:26 -03:00
parent 23b40cac81
commit cdf7521530
4 changed files with 90 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
tests/unit/engines/test_subscription.py
View File

@@ -150,9 +150,7 @@ async def test_subscribe_without_subscription_raises():
async def test_unsubscribe_missing_logs_warning():
eng = make_engine()
await eng.unsubscribe('sub1')
eng.logger.warning.assert_called_once_with(
"Subscription 'sub1' not found. Cannot unsubscribe."
)
eng.logger.warning.assert_called_once_with("Subscription 'sub1' not found. Cannot unsubscribe.")
@pytest.mark.asyncio

50
tests/unit/io/test_slots.py
View File

@@ -1,10 +1,33 @@
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'}]}}}}
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():
@@ -15,8 +38,20 @@ def test_parse_slot_multi_name():
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'}]}}}}
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}
@@ -24,5 +59,6 @@ def test_parse_slot_defaults_active_read_disabled():
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)

30
tests/unit/test_converters.py
View File

@@ -3,10 +3,20 @@ import pytest
from ingestor.converters import ConversionError, to_typed_value
@pytest.mark.parametrize('raw,expected', [
('false', False), ('False', False), ('0', False), ('no', False),
('off', False), ('', False), ('true', True), ('1', True), (True, True),
])
@pytest.mark.parametrize(
'raw,expected',
[
('false', False),
('False', False),
('0', False),
('no', False),
('off', False),
('', False),
('true', True),
('1', True),
(True, True),
],
)
def test_bool_from_json_string(raw, expected):
assert to_typed_value(raw, 'boolean') is expected
@@ -16,9 +26,15 @@ def test_float_and_int():
assert to_typed_value('7', 'int') == 7
@pytest.mark.parametrize('raw,dtype', [
(None, 'int'), ('abc', 'float'), ('123.123', 'int'), (None, 'float'),
])
@pytest.mark.parametrize(
'raw,dtype',
[
(None, 'int'),
('abc', 'float'),
('123.123', 'int'),
(None, 'float'),
],
)
def test_invalid_raises_conversion_error(raw, dtype):
with pytest.raises(ConversionError):
to_typed_value(raw, dtype)