Skip to content

O
opcua-asyncio
Commit 4221be7d authored by Christian Bergmiller's avatar Christian Bergmiller
Browse files
Options

refactored tests and async fixes

parent 006689dc
Hide whitespace changes
Inline Side-by-side
Showing with 714 additions and 763 deletions
opcua/common/events.py
View file @ 4221be7d
...@@ -112,7 +112,7 @@ class Event(object): ...@@ -112,7 +112,7 @@ class Event(object):
async def get_filter_from_event_type(eventtypes): async def get_filter_from_event_type(eventtypes):
evfilter = ua.EventFilter() evfilter = ua.EventFilter()
evfilter.SelectClauses = await select_clauses_from_evtype(eventtypes) evfilter.SelectClauses = await select_clauses_from_evtype(eventtypes)
evfilter.WhereClause = where_clause_from_evtype(eventtypes) evfilter.WhereClause = await where_clause_from_evtype(eventtypes)
return evfilter return evfilter
...@@ -121,19 +121,19 @@ async def select_clauses_from_evtype(evtypes): ...@@ -121,19 +121,19 @@ async def select_clauses_from_evtype(evtypes):
selected_paths = [] selected_paths = []
for evtype in evtypes: for evtype in evtypes:
for prop in await get_event_properties_from_type_node(evtype): for prop in await get_event_properties_from_type_node(evtype):
if prop.get_browse_name() not in selected_paths: browse_name = await prop.get_browse_name()
if browse_name not in selected_paths:
op = ua.SimpleAttributeOperand() op = ua.SimpleAttributeOperand()
op.AttributeId = ua.AttributeIds.Value op.AttributeId = ua.AttributeIds.Value
op.BrowsePath = [prop.get_browse_name()] op.BrowsePath = [browse_name]
clauses.append(op) clauses.append(op)
selected_paths.append(prop.get_browse_name()) selected_paths.append(browse_name)
return clauses return clauses
def where_clause_from_evtype(evtypes): async def where_clause_from_evtype(evtypes):
cf = ua.ContentFilter() cf = ua.ContentFilter()
el = ua.ContentFilterElement() el = ua.ContentFilterElement()
# operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute # operands can be ElementOperand, LiteralOperand, AttributeOperand, SimpleAttribute
# Create a clause where the generate event type property EventType # Create a clause where the generate event type property EventType
# must be a subtype of events in evtypes argument # must be a subtype of events in evtypes argument
...@@ -144,20 +144,18 @@ def where_clause_from_evtype(evtypes): ...@@ -144,20 +144,18 @@ def where_clause_from_evtype(evtypes):
op.BrowsePath.append(ua.QualifiedName("EventType", 0)) op.BrowsePath.append(ua.QualifiedName("EventType", 0))
op.AttributeId = ua.AttributeIds.Value op.AttributeId = ua.AttributeIds.Value
el.FilterOperands.append(op) el.FilterOperands.append(op)
# now create a list of all subtypes we want to accept # now create a list of all subtypes we want to accept
subtypes = [] subtypes = []
for evtype in evtypes: for evtype in evtypes:
subtypes += [st.nodeid for st in ua_utils.get_node_subtypes(evtype)] for st in await ua_utils.get_node_subtypes(evtype):
subtypes.append(st.nodeid)
subtypes = list(set(subtypes)) # remove duplicates subtypes = list(set(subtypes)) # remove duplicates
for subtypeid in subtypes: for subtypeid in subtypes:
op = ua.LiteralOperand() op = ua.LiteralOperand()
op.Value = ua.Variant(subtypeid) op.Value = ua.Variant(subtypeid)
el.FilterOperands.append(op) el.FilterOperands.append(op)
el.FilterOperator = ua.FilterOperator.InList el.FilterOperator = ua.FilterOperator.InList
cf.Elements.append(el) cf.Elements.append(el)
return cf return cf
...@@ -165,7 +163,7 @@ async def get_event_properties_from_type_node(node): ...@@ -165,7 +163,7 @@ async def get_event_properties_from_type_node(node):
properties = [] properties = []
curr_node = node curr_node = node
while True: while True:
properties.extend(curr_node.get_properties()) properties.extend(await curr_node.get_properties())
if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType: if curr_node.nodeid.Identifier == ua.ObjectIds.BaseEventType:
break break
parents = await curr_node.get_referenced_nodes( parents = await curr_node.get_referenced_nodes(
......
tests/conftest.py
View file @ 4221be7d
import pytest import pytest
from collections import namedtuple
from opcua import Client from opcua import Client
from opcua import Server from opcua import Server
from .test_common import add_server_methods from .test_common import add_server_methods
from .util_enum_struct import add_server_custom_enum_struct
port_num1 = 48510
port_num = 48540 port_num = 48540
port_num1 = 48510
port_discovery = 48550
Opc = namedtuple('opc', ['opc', 'server'])
def pytest_generate_tests(metafunc): def pytest_generate_tests(metafunc):
if 'opc' in metafunc.fixturenames: if 'opc' in metafunc.fixturenames:
metafunc.parametrize('opc', ['client', 'server'], indirect=True) metafunc.parametrize('opc', ['client', 'server'], indirect=True)
@pytest.fixture()
async def server():
# start our own server
srv = Server()
await srv.init()
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_num}')
await add_server_methods(srv)
await srv.start()
yield srv
# stop the server
await srv.stop()
@pytest.fixture()
async def discovery_server():
# start our own server
srv = Server()
await srv.init()
await srv.set_application_uri('urn:freeopcua:python:discovery')
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_discovery}')
await srv.start()
yield srv
# stop the server
await srv.stop()
@pytest.fixture()
async def admin_client():
# start admin client
# long timeout since travis (automated testing) can be really slow
clt = Client(f'opc.tcp://admin@127.0.0.1:{port_num1}', timeout=10)
await clt.connect()
yield clt
await clt.disconnect()
@pytest.fixture()
async def client():
# start anonymous client
ro_clt = Client(f'opc.tcp://127.0.0.1:{port_num1}')
await ro_clt.connect()
yield ro_clt
await ro_clt.disconnect()
@pytest.fixture() @pytest.fixture()
async def opc(request): async def opc(request):
"""
Fixture for tests that should run for both `Server` and `Client`
:param request:
:return:
"""
if request.param == 'client': if request.param == 'client':
srv = Server() srv = Server()
await srv.init() await srv.init()
...@@ -24,7 +78,7 @@ async def opc(request): ...@@ -24,7 +78,7 @@ async def opc(request):
# long timeout since travis (automated testing) can be really slow # long timeout since travis (automated testing) can be really slow
clt = Client(f'opc.tcp://admin@127.0.0.1:{port_num}', timeout=10) clt = Client(f'opc.tcp://admin@127.0.0.1:{port_num}', timeout=10)
await clt.connect() await clt.connect()
yield clt yield Opc(clt, srv)
await clt.disconnect() await clt.disconnect()
await srv.stop() await srv.stop()
elif request.param == 'server': elif request.param == 'server':
...@@ -32,10 +86,10 @@ async def opc(request): ...@@ -32,10 +86,10 @@ async def opc(request):
# start our own server # start our own server
srv = Server() srv = Server()
await srv.init() await srv.init()
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_num}') srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_num1}')
await add_server_methods(srv) await add_server_methods(srv)
await srv.start() await srv.start()
yield srv yield Opc(srv, srv)
# stop the server # stop the server
await srv.stop() await srv.stop()
else: else:
......
tests/test_client.py
View file @ 4221be7d
...@@ -6,47 +6,10 @@ from opcua import Client ...@@ -6,47 +6,10 @@ from opcua import Client
from opcua import Server from opcua import Server
from opcua import ua from opcua import ua
from .test_common import add_server_methods
from .util_enum_struct import add_server_custom_enum_struct
port_num1 = 48510
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
pytestmark = pytest.mark.asyncio pytestmark = pytest.mark.asyncio
@pytest.fixture()
async def admin_client():
# start admin client
# long timeout since travis (automated testing) can be really slow
clt = Client(f'opc.tcp://admin@127.0.0.1:{port_num1}', timeout=10)
await clt.connect()
yield clt
await clt.disconnect()
@pytest.fixture()
async def client():
# start anonymous client
ro_clt = Client(f'opc.tcp://127.0.0.1:{port_num1}')
await ro_clt.connect()
yield ro_clt
await ro_clt.disconnect()
@pytest.fixture()
async def server():
# start our own server
srv = Server()
await srv.init()
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_num1}')
await add_server_methods(srv)
await add_server_custom_enum_struct(srv)
await srv.start()
yield srv
# stop the server
await srv.stop()
async def test_service_fault(server, admin_client): async def test_service_fault(server, admin_client):
request = ua.ReadRequest() request = ua.ReadRequest()
request.TypeId = ua.FourByteNodeId(999) # bad type! request.TypeId = ua.FourByteNodeId(999) # bad type!
......
tests/test_common.py
View file @ 4221be7d
...@@ -83,12 +83,12 @@ async def add_server_methods(srv): ...@@ -83,12 +83,12 @@ async def add_server_methods(srv):
async def test_find_servers(opc): async def test_find_servers(opc):
servers = await opc.find_servers() servers = await opc.opc.find_servers()
# FIXME : finish # FIXME : finish
async def test_add_node_bad_args(opc): async def test_add_node_bad_args(opc):
obj = opc.get_objects_node() obj = opc.opc.get_objects_node()
with pytest.raises(TypeError): with pytest.raises(TypeError):
fold = await obj.add_folder(1.2, "kk") fold = await obj.add_folder(1.2, "kk")
...@@ -107,12 +107,12 @@ async def test_add_node_bad_args(opc): ...@@ -107,12 +107,12 @@ async def test_add_node_bad_args(opc):
async def test_delete_nodes(opc): async def test_delete_nodes(opc):
obj = opc.get_objects_node() obj = opc.opc.get_objects_node()
fold = await obj.add_folder(2, "FolderToDelete") fold = await obj.add_folder(2, "FolderToDelete")
var = await fold.add_variable(2, "VarToDelete", 9.1) var = await fold.add_variable(2, "VarToDelete", 9.1)
childs = await fold.get_children() childs = await fold.get_children()
assert var in childs assert var in childs
await opc.delete_nodes([var]) await opc.opc.delete_nodes([var])
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await var.set_value(7.8) await var.set_value(7.8)
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
...@@ -122,10 +122,10 @@ async def test_delete_nodes(opc): ...@@ -122,10 +122,10 @@ async def test_delete_nodes(opc):
async def test_delete_nodes_recursive(opc): async def test_delete_nodes_recursive(opc):
obj = opc.get_objects_node() obj = opc.opc.get_objects_node()
fold = await obj.add_folder(2, "FolderToDeleteR") fold = await obj.add_folder(2, "FolderToDeleteR")
var = await fold.add_variable(2, "VarToDeleteR", 9.1) var = await fold.add_variable(2, "VarToDeleteR", 9.1)
await opc.delete_nodes([fold, var]) await opc.opc.delete_nodes([fold, var])
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await var.set_value(7.8) await var.set_value(7.8)
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
...@@ -133,7 +133,7 @@ async def test_delete_nodes_recursive(opc): ...@@ -133,7 +133,7 @@ async def test_delete_nodes_recursive(opc):
async def test_delete_nodes_recursive2(opc): async def test_delete_nodes_recursive2(opc):
obj = opc.get_objects_node() obj = opc.opc.get_objects_node()
fold = await obj.add_folder(2, "FolderToDeleteRoot") fold = await obj.add_folder(2, "FolderToDeleteRoot")
nfold = fold nfold = fold
mynodes = [] mynodes = []
...@@ -147,16 +147,16 @@ async def test_delete_nodes_recursive2(opc): ...@@ -147,16 +147,16 @@ async def test_delete_nodes_recursive2(opc):
mynodes.append(var) mynodes.append(var)
mynodes.append(prop) mynodes.append(prop)
mynodes.append(o) mynodes.append(o)
await opc.delete_nodes([fold], recursive=True) await opc.opc.delete_nodes([fold], recursive=True)
for node in mynodes: for node in mynodes:
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await node.get_browse_name() await node.get_browse_name()
async def test_delete_references(opc): async def test_delete_references(opc):
newtype = await opc.get_node(ua.ObjectIds.HierarchicalReferences).add_reference_type(0, "HasSuperSecretVariable") newtype = await opc.opc.get_node(ua.ObjectIds.HierarchicalReferences).add_reference_type(0, "HasSuperSecretVariable")
obj = opc.get_objects_node() obj = opc.opc.get_objects_node()
fold = await obj.add_folder(2, "FolderToRef") fold = await obj.add_folder(2, "FolderToRef")
var = await fold.add_variable(2, "VarToRef", 42) var = await fold.add_variable(2, "VarToRef", 42)
...@@ -203,16 +203,16 @@ async def test_delete_references(opc): ...@@ -203,16 +203,16 @@ async def test_delete_references(opc):
assert [] == await fold.get_referenced_nodes(newtype) assert [] == await fold.get_referenced_nodes(newtype)
# clean-up # clean-up
await opc.delete_nodes([fold, newtype], recursive=True) await opc.opc.delete_nodes([fold, newtype], recursive=True)
async def test_server_node(opc): async def test_server_node(opc):
node = opc.get_server_node() node = opc.opc.get_server_node()
assert ua.QualifiedName('Server', 0) == await node.get_browse_name() assert ua.QualifiedName('Server', 0) == await node.get_browse_name()
async def test_root(opc): async def test_root(opc):
root = opc.get_root_node() root = opc.opc.get_root_node()
assert ua.QualifiedName('Root', 0) == await root.get_browse_name() assert ua.QualifiedName('Root', 0) == await root.get_browse_name()
assert ua.LocalizedText('Root') == await root.get_display_name() assert ua.LocalizedText('Root') == await root.get_display_name()
nid = ua.NodeId(84, 0) nid = ua.NodeId(84, 0)
...@@ -220,14 +220,14 @@ async def test_root(opc): ...@@ -220,14 +220,14 @@ async def test_root(opc):
async def test_objects(opc): async def test_objects(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
assert ua.QualifiedName('Objects', 0) == await objects.get_browse_name() assert ua.QualifiedName('Objects', 0) == await objects.get_browse_name()
nid = ua.NodeId(85, 0) nid = ua.NodeId(85, 0)
assert nid == objects.nodeid assert nid == objects.nodeid
async def test_browse(opc): async def test_browse(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
obj = await objects.add_object(4, "browsetest") obj = await objects.add_object(4, "browsetest")
folder = await obj.add_folder(4, "folder") folder = await obj.add_folder(4, "folder")
prop = await obj.add_property(4, "property", 1) prop = await obj.add_property(4, "property", 1)
...@@ -258,7 +258,7 @@ async def test_browse(opc): ...@@ -258,7 +258,7 @@ async def test_browse(opc):
async def test_browse_references(opc): async def test_browse_references(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
folder = await objects.add_folder(4, "folder") folder = await objects.add_folder(4, "folder")
childs = await objects.get_referenced_nodes( childs = await objects.get_referenced_nodes(
...@@ -289,33 +289,33 @@ async def test_browse_references(opc): ...@@ -289,33 +289,33 @@ async def test_browse_references(opc):
async def test_browsename_with_spaces(opc): async def test_browsename_with_spaces(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'BNVariable with spaces and %&+?/', 1.3) v = await o.add_variable(3, 'BNVariable with spaces and %&+?/', 1.3)
v2 = await o.get_child("3:BNVariable with spaces and %&+?/") v2 = await o.get_child("3:BNVariable with spaces and %&+?/")
assert v == v2 assert v == v2
async def test_non_existing_path(opc): async def test_non_existing_path(opc):
root = opc.get_root_node() root = opc.opc.get_root_node()
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await root.get_child(['0:Objects', '0:Server', '0:nonexistingnode']) await root.get_child(['0:Objects', '0:Server', '0:nonexistingnode'])
async def test_bad_attribute(opc): async def test_bad_attribute(opc):
root = opc.get_root_node() root = opc.opc.get_root_node()
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await root.set_value(99) await root.set_value(99)
async def test_get_node_by_nodeid(opc): async def test_get_node_by_nodeid(opc):
root = opc.get_root_node() root = opc.opc.get_root_node()
server_time_node = await root.get_child(['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime']) server_time_node = await root.get_child(['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime'])
correct = opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) correct = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
assert server_time_node == correct assert server_time_node == correct
async def test_datetime_read(opc): async def test_datetime_read(opc):
time_node = opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) time_node = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
dt = await time_node.get_value() dt = await time_node.get_value()
utcnow = datetime.utcnow() utcnow = datetime.utcnow()
delta = utcnow - dt delta = utcnow - dt
...@@ -323,16 +323,16 @@ async def test_datetime_read(opc): ...@@ -323,16 +323,16 @@ async def test_datetime_read(opc):
async def test_datetime_write(opc): async def test_datetime_write(opc):
time_node = opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) time_node = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
now = datetime.utcnow() now = datetime.utcnow()
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
v1 = await objects.add_variable(4, "test_datetime", now) v1 = await objects.add_variable(4, "test_datetime", now)
tid = await v1.get_value() tid = await v1.get_value()
assert now == tid assert now == tid
async def test_variant_array_dim(opc): async def test_variant_array_dim(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
l = [[[1.0, 1.0, 1.0, 1.0], [2.0, 2.0, 2.0, 2.0], [3.0, 3.0, 3.0, 3.0]], l = [[[1.0, 1.0, 1.0, 1.0], [2.0, 2.0, 2.0, 2.0], [3.0, 3.0, 3.0, 3.0]],
[[5.0, 5.0, 5.0, 5.0], [7.0, 8.0, 9.0, 01.0], [1.0, 1.0, 1.0, 1.0]]] [[5.0, 5.0, 5.0, 5.0], [7.0, 8.0, 9.0, 01.0], [1.0, 1.0, 1.0, 1.0]]]
v = await objects.add_variable(3, 'variableWithDims', l) v = await objects.add_variable(3, 'variableWithDims', l)
...@@ -360,7 +360,7 @@ async def test_variant_array_dim(opc): ...@@ -360,7 +360,7 @@ async def test_variant_array_dim(opc):
async def test_add_numeric_variable(opc): async def test_add_numeric_variable(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
v = await objects.add_variable('ns=3;i=888;', '3:numericnodefromstring', 99) v = await objects.add_variable('ns=3;i=888;', '3:numericnodefromstring', 99)
nid = ua.NodeId(888, 3) nid = ua.NodeId(888, 3)
qn = ua.QualifiedName('numericnodefromstring', 3) qn = ua.QualifiedName('numericnodefromstring', 3)
...@@ -369,7 +369,7 @@ async def test_add_numeric_variable(opc): ...@@ -369,7 +369,7 @@ async def test_add_numeric_variable(opc):
async def test_add_string_variable(opc): async def test_add_string_variable(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
v = await objects.add_variable('ns=3;s=stringid;', '3:stringnodefromstring', [68]) v = await objects.add_variable('ns=3;s=stringid;', '3:stringnodefromstring', [68])
nid = ua.NodeId('stringid', 3) nid = ua.NodeId('stringid', 3)
qn = ua.QualifiedName('stringnodefromstring', 3) qn = ua.QualifiedName('stringnodefromstring', 3)
...@@ -378,7 +378,7 @@ async def test_add_string_variable(opc): ...@@ -378,7 +378,7 @@ async def test_add_string_variable(opc):
async def test_utf8(opc): async def test_utf8(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
utf_string = "æøå@%&" utf_string = "æøå@%&"
bn = ua.QualifiedName(utf_string, 3) bn = ua.QualifiedName(utf_string, 3)
nid = ua.NodeId("æølå", 3) nid = ua.NodeId("æølå", 3)
...@@ -392,7 +392,7 @@ async def test_utf8(opc): ...@@ -392,7 +392,7 @@ async def test_utf8(opc):
async def test_null_variable(opc): async def test_null_variable(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
var = await objects.add_variable(3, 'nullstring', "a string") var = await objects.add_variable(3, 'nullstring', "a string")
await var.set_value(None) await var.set_value(None)
val = await var.get_value() val = await var.get_value()
...@@ -404,7 +404,7 @@ async def test_null_variable(opc): ...@@ -404,7 +404,7 @@ async def test_null_variable(opc):
async def test_variable_data_type(opc): async def test_variable_data_type(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
var = await objects.add_variable(3, 'stringfordatatype', "a string") var = await objects.add_variable(3, 'stringfordatatype', "a string")
val = await var.get_data_type_as_variant_type() val = await var.get_data_type_as_variant_type()
assert ua.VariantType.String == val assert ua.VariantType.String == val
...@@ -414,7 +414,7 @@ async def test_variable_data_type(opc): ...@@ -414,7 +414,7 @@ async def test_variable_data_type(opc):
async def test_add_string_array_variable(opc): async def test_add_string_array_variable(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
v = await objects.add_variable('ns=3;s=stringarrayid;', '9:stringarray', ['l', 'b']) v = await objects.add_variable('ns=3;s=stringarrayid;', '9:stringarray', ['l', 'b'])
nid = ua.NodeId('stringarrayid', 3) nid = ua.NodeId('stringarrayid', 3)
qn = ua.QualifiedName('stringarray', 9) qn = ua.QualifiedName('stringarray', 9)
...@@ -425,7 +425,7 @@ async def test_add_string_array_variable(opc): ...@@ -425,7 +425,7 @@ async def test_add_string_array_variable(opc):
async def test_add_numeric_node(opc): async def test_add_numeric_node(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
nid = ua.NodeId(9999, 3) nid = ua.NodeId(9999, 3)
qn = ua.QualifiedName('AddNodeVar1', 3) qn = ua.QualifiedName('AddNodeVar1', 3)
v1 = await objects.add_variable(nid, qn, 0) v1 = await objects.add_variable(nid, qn, 0)
...@@ -434,7 +434,7 @@ async def test_add_numeric_node(opc): ...@@ -434,7 +434,7 @@ async def test_add_numeric_node(opc):
async def test_add_string_node(opc): async def test_add_string_node(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
qn = ua.QualifiedName('AddNodeVar2', 3) qn = ua.QualifiedName('AddNodeVar2', 3)
nid = ua.NodeId('AddNodeVar2Id', 3) nid = ua.NodeId('AddNodeVar2Id', 3)
v2 = await objects.add_variable(nid, qn, 0) v2 = await objects.add_variable(nid, qn, 0)
...@@ -443,22 +443,22 @@ async def test_add_string_node(opc): ...@@ -443,22 +443,22 @@ async def test_add_string_node(opc):
async def test_add_find_node_(opc): async def test_add_find_node_(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
o = await objects.add_object('ns=2;i=101;', '2:AddFindObject') o = await objects.add_object('ns=2;i=101;', '2:AddFindObject')
o2 = await objects.get_child('2:AddFindObject') o2 = await objects.get_child('2:AddFindObject')
assert o == o2 assert o == o2
async def test_node_path(opc): async def test_node_path(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
o = await objects.add_object('ns=2;i=105;', '2:NodePathObject') o = await objects.add_object('ns=2;i=105;', '2:NodePathObject')
root = opc.get_root_node() root = opc.opc.get_root_node()
o2 = await root.get_child(['0:Objects', '2:NodePathObject']) o2 = await root.get_child(['0:Objects', '2:NodePathObject'])
assert o == o2 assert o == o2
async def test_add_read_node(opc): async def test_add_read_node(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
o = await objects.add_object('ns=2;i=102;', '2:AddReadObject') o = await objects.add_object('ns=2;i=102;', '2:AddReadObject')
nid = ua.NodeId(102, 2) nid = ua.NodeId(102, 2)
assert nid == o.nodeid assert nid == o.nodeid
...@@ -467,33 +467,33 @@ async def test_add_read_node(opc): ...@@ -467,33 +467,33 @@ async def test_add_read_node(opc):
async def test_simple_value(opc): async def test_simple_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'VariableTestValue', 4.32) v = await o.add_variable(3, 'VariableTestValue', 4.32)
val = await v.get_value() val = await v.get_value()
assert 4.32 == val assert 4.32 == val
async def test_add_exception(opc): async def test_add_exception(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
await objects.add_object('ns=2;i=103;', '2:AddReadObject') await objects.add_object('ns=2;i=103;', '2:AddReadObject')
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await objects.add_object('ns=2;i=103;', '2:AddReadObject') await objects.add_object('ns=2;i=103;', '2:AddReadObject')
async def test_negative_value(opc): async def test_negative_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'VariableNegativeValue', 4) v = await o.add_variable(3, 'VariableNegativeValue', 4)
await v.set_value(-4.54) await v.set_value(-4.54)
assert -4.54 == await v.get_value() assert -4.54 == await v.get_value()
async def test_read_server_state(opc): async def test_read_server_state(opc):
statenode = opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State)) statenode = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State))
assert 0 == await statenode.get_value() assert 0 == await statenode.get_value()
async def test_bad_node(opc): async def test_bad_node(opc):
bad = opc.get_node(ua.NodeId(999, 999)) bad = opc.opc.get_node(ua.NodeId(999, 999))
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
await bad.get_browse_name() await bad.get_browse_name()
with pytest.raises(ua.UaStatusCodeError): with pytest.raises(ua.UaStatusCodeError):
...@@ -505,7 +505,7 @@ async def test_bad_node(opc): ...@@ -505,7 +505,7 @@ async def test_bad_node(opc):
async def test_value(opc): async def test_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
var = ua.Variant(1.98, ua.VariantType.Double) var = ua.Variant(1.98, ua.VariantType.Double)
v = await o.add_variable(3, 'VariableValue', var) v = await o.add_variable(3, 'VariableValue', var)
assert 1.98 == await v.get_value() assert 1.98 == await v.get_value()
...@@ -517,7 +517,7 @@ async def test_value(opc): ...@@ -517,7 +517,7 @@ async def test_value(opc):
async def test_set_value(opc): async def test_set_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
var = ua.Variant(1.98, ua.VariantType.Double) var = ua.Variant(1.98, ua.VariantType.Double)
dvar = ua.DataValue(var) dvar = ua.DataValue(var)
v = await o.add_variable(3, 'VariableValue', var) v = await o.add_variable(3, 'VariableValue', var)
...@@ -533,13 +533,13 @@ async def test_set_value(opc): ...@@ -533,13 +533,13 @@ async def test_set_value(opc):
async def test_array_value(opc): async def test_array_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3]) v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3])
assert [1, 2, 3] == await v.get_value() assert [1, 2, 3] == await v.get_value()
async def test_bool_variable(opc): async def test_bool_variable(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'BoolVariable', True) v = await o.add_variable(3, 'BoolVariable', True)
dt = await v.get_data_type_as_variant_type() dt = await v.get_data_type_as_variant_type()
assert ua.VariantType.Boolean == dt assert ua.VariantType.Boolean == dt
...@@ -551,23 +551,23 @@ async def test_bool_variable(opc): ...@@ -551,23 +551,23 @@ async def test_bool_variable(opc):
async def test_array_size_one_value(opc): async def test_array_size_one_value(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3]) v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3])
await v.set_value([1]) await v.set_value([1])
assert [1] == await v.get_value() assert [1] == await v.get_value()
async def test_use_namespace(opc): async def test_use_namespace(opc):
idx = await opc.get_namespace_index("urn:freeopcua:python:server") idx = await opc.opc.get_namespace_index("urn:freeopcua:python:server")
assert 1 == idx assert 1 == idx
root = opc.get_root_node() root = opc.opc.get_root_node()
myvar = await root.add_variable(idx, 'var_in_custom_namespace', [5]) myvar = await root.add_variable(idx, 'var_in_custom_namespace', [5])
myid = myvar.nodeid myid = myvar.nodeid
assert idx == myid.NamespaceIndex assert idx == myid.NamespaceIndex
async def test_method(opc): async def test_method(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
await o.get_child("2:ServerMethod") await o.get_child("2:ServerMethod")
result = await o.call_method("2:ServerMethod", 2.1) result = await o.call_method("2:ServerMethod", 2.1)
assert 4.2 == result assert 4.2 == result
...@@ -579,7 +579,7 @@ async def test_method(opc): ...@@ -579,7 +579,7 @@ async def test_method(opc):
async def test_method_array(opc): async def test_method_array(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
m = await o.get_child("2:ServerMethodArray") m = await o.get_child("2:ServerMethodArray")
result = await o.call_method(m, "sin", ua.Variant(math.pi)) result = await o.call_method(m, "sin", ua.Variant(math.pi))
assert result < 0.01 assert result < 0.01
...@@ -592,7 +592,7 @@ async def test_method_array(opc): ...@@ -592,7 +592,7 @@ async def test_method_array(opc):
async def test_method_array2(opc): async def test_method_array2(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
m = o.get_child("2:ServerMethodArray2") m = o.get_child("2:ServerMethodArray2")
result = await o.call_method(m, [1.1, 3.4, 9]) result = await o.call_method(m, [1.1, 3.4, 9])
assert [2.2, 6.8, 18] == result assert [2.2, 6.8, 18] == result
...@@ -601,7 +601,7 @@ async def test_method_array2(opc): ...@@ -601,7 +601,7 @@ async def test_method_array2(opc):
async def test_method_tuple(opc): async def test_method_tuple(opc):
o = opc.get_objects_node() o = opc.opc.get_objects_node()
m = await o.get_child("2:ServerMethodTuple") m = await o.get_child("2:ServerMethodTuple")
result = await o.call_method(m) result = await o.call_method(m)
assert [1, 2, 3] == result assert [1, 2, 3] == result
...@@ -611,7 +611,7 @@ async def test_method_tuple(opc): ...@@ -611,7 +611,7 @@ async def test_method_tuple(opc):
async def test_method_none(opc): async def test_method_none(opc):
# this test calls the function linked to the type's method.. # this test calls the function linked to the type's method..
o = await opc.get_node(ua.ObjectIds.BaseObjectType).get_child("2:ObjectWithMethodsType") o = await opc.opc.get_node(ua.ObjectIds.BaseObjectType).get_child("2:ObjectWithMethodsType")
m = await o.get_child("2:ServerMethodDefault") m = await o.get_child("2:ServerMethodDefault")
result = await o.call_method(m) result = await o.call_method(m)
assert result is None assert result is None
...@@ -620,7 +620,7 @@ async def test_method_none(opc): ...@@ -620,7 +620,7 @@ async def test_method_none(opc):
async def test_add_nodes(opc): async def test_add_nodes(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
f = await objects.add_folder(3, 'MyFolder') f = await objects.add_folder(3, 'MyFolder')
child = await objects.get_child("3:MyFolder") child = await objects.get_child("3:MyFolder")
assert child == f assert child == f
...@@ -640,7 +640,7 @@ async def test_add_nodes(opc): ...@@ -640,7 +640,7 @@ async def test_add_nodes(opc):
async def test_modelling_rules(opc): async def test_modelling_rules(opc):
obj = await opc.nodes.base_object_type.add_object_type(2, 'MyFooObjectType') obj = await opc.opc.nodes.base_object_type.add_object_type(2, 'MyFooObjectType')
v = await obj.add_variable(2, "myvar", 1.1) v = await obj.add_variable(2, "myvar", 1.1)
await v.set_modelling_rule(True) await v.set_modelling_rule(True)
p = await obj.add_property(2, "myvar", 1.1) p = await obj.add_property(2, "myvar", 1.1)
...@@ -650,10 +650,10 @@ async def test_modelling_rules(opc): ...@@ -650,10 +650,10 @@ async def test_modelling_rules(opc):
assert 0 == len(refs) assert 0 == len(refs)
refs = await v.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) refs = await v.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
assert opc.get_node(ua.ObjectIds.ModellingRule_Mandatory) == refs[0] assert opc.opc.get_node(ua.ObjectIds.ModellingRule_Mandatory) == refs[0]
refs = await p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) refs = await p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
assert opc.get_node(ua.ObjectIds.ModellingRule_Optional) == refs[0] assert opc.opc.get_node(ua.ObjectIds.ModellingRule_Optional) == refs[0]
await p.set_modelling_rule(None) await p.set_modelling_rule(None)
refs = await p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) refs = await p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
...@@ -661,7 +661,7 @@ async def test_modelling_rules(opc): ...@@ -661,7 +661,7 @@ async def test_modelling_rules(opc):
async def test_incl_subtypes(opc): async def test_incl_subtypes(opc):
base_type = await opc.get_root_node().get_child(["0:Types", "0:ObjectTypes", "0:BaseObjectType"]) base_type = await opc.opc.get_root_node().get_child(["0:Types", "0:ObjectTypes", "0:BaseObjectType"])
descs = await base_type.get_children_descriptions(includesubtypes=True) descs = await base_type.get_children_descriptions(includesubtypes=True)
assert len(descs) > 10 assert len(descs) > 10
descs = await base_type.get_children_descriptions(includesubtypes=False) descs = await base_type.get_children_descriptions(includesubtypes=False)
...@@ -669,7 +669,7 @@ async def test_incl_subtypes(opc): ...@@ -669,7 +669,7 @@ async def test_incl_subtypes(opc):
async def test_add_node_with_type(opc): async def test_add_node_with_type(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
f = await objects.add_folder(3, 'MyFolder_TypeTest') f = await objects.add_folder(3, 'MyFolder_TypeTest')
o = await f.add_object(3, 'MyObject1', ua.ObjectIds.BaseObjectType) o = await f.add_object(3, 'MyObject1', ua.ObjectIds.BaseObjectType)
...@@ -678,7 +678,7 @@ async def test_add_node_with_type(opc): ...@@ -678,7 +678,7 @@ async def test_add_node_with_type(opc):
o = await f.add_object(3, 'MyObject2', ua.NodeId(ua.ObjectIds.BaseObjectType, 0)) o = await f.add_object(3, 'MyObject2', ua.NodeId(ua.ObjectIds.BaseObjectType, 0))
assert ua.ObjectIds.BaseObjectType == (await o.get_type_definition()).Identifier assert ua.ObjectIds.BaseObjectType == (await o.get_type_definition()).Identifier
base_otype = opc.get_node(ua.ObjectIds.BaseObjectType) base_otype = opc.opc.get_node(ua.ObjectIds.BaseObjectType)
custom_otype = await base_otype.add_object_type(2, 'MyFooObjectType') custom_otype = await base_otype.add_object_type(2, 'MyFooObjectType')
o = await f.add_object(3, 'MyObject3', custom_otype.nodeid) o = await f.add_object(3, 'MyObject3', custom_otype.nodeid)
...@@ -690,7 +690,7 @@ async def test_add_node_with_type(opc): ...@@ -690,7 +690,7 @@ async def test_add_node_with_type(opc):
async def test_references_for_added_nodes(opc): async def test_references_for_added_nodes(opc):
objects = opc.get_objects_node() objects = opc.opc.get_objects_node()
o = await objects.add_object(3, 'MyObject') o = await objects.add_object(3, 'MyObject')
nodes = await objects.get_referenced_nodes( nodes = await objects.get_referenced_nodes(
refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False
...@@ -748,7 +748,7 @@ async def test_references_for_added_nodes(opc): ...@@ -748,7 +748,7 @@ async def test_references_for_added_nodes(opc):
async def test_path_string(opc): async def test_path_string(opc):
o = await (await opc.nodes.objects.add_folder(1, "titif")).add_object(3, "opath") o = await (await opc.opc.nodes.objects.add_folder(1, "titif")).add_object(3, "opath")
path = await o.get_path(as_string=True) path = await o.get_path(as_string=True)
assert ["0:Root", "0:Objects", "1:titif", "3:opath"] == path assert ["0:Root", "0:Objects", "1:titif", "3:opath"] == path
path = await o.get_path(2, as_string=True) path = await o.get_path(2, as_string=True)
...@@ -756,28 +756,28 @@ async def test_path_string(opc): ...@@ -756,28 +756,28 @@ async def test_path_string(opc):
async def test_path(opc): async def test_path(opc):
of = await opc.nodes.objects.add_folder(1, "titif") of = await opc.opc.nodes.objects.add_folder(1, "titif")
op = await of.add_object(3, "opath") op = await of.add_object(3, "opath")
path = await op.get_path() path = await op.get_path()
assert [opc.nodes.root, opc.nodes.objects, of, op] == path assert [opc.opc.nodes.root, opc.opc.nodes.objects, of, op] == path
path = await op.get_path(2) path = await op.get_path(2)
assert [of, op] == path assert [of, op] == path
target = opc.get_node("i=13387") target = opc.opc.get_node("i=13387")
path = await target.get_path() path = await target.get_path()
assert [ assert [
opc.nodes.root, opc.nodes.types, opc.nodes.object_types, opc.nodes.base_object_type, opc.opc.nodes.root, opc.opc.nodes.types, opc.opc.nodes.object_types, opc.opc.nodes.base_object_type,
opc.nodes.folder_type, opc.get_node(ua.ObjectIds.FileDirectoryType), target opc.opc.nodes.folder_type, opc.opc.get_node(ua.ObjectIds.FileDirectoryType), target
] == path ] == path
async def test_get_endpoints(opc): async def test_get_endpoints(opc):
endpoints = await opc.get_endpoints() endpoints = await opc.opc.get_endpoints()
assert len(endpoints) > 0 assert len(endpoints) > 0
assert endpoints[0].EndpointUrl.startswith("opc.tcp://") assert endpoints[0].EndpointUrl.startswith("opc.opc.tcp://")
async def test_copy_node(opc): async def test_copy_node(opc):
dev_t = await opc.nodes.base_data_type.add_object_type(0, "MyDevice") dev_t = await opc.opc.nodes.base_data_type.add_object_type(0, "MyDevice")
v_t = await dev_t.add_variable(0, "sensor", 1.0) v_t = await dev_t.add_variable(0, "sensor", 1.0)
p_t = await dev_t.add_property(0, "sensor_id", "0340") p_t = await dev_t.add_property(0, "sensor_id", "0340")
ctrl_t = await dev_t.add_object(0, "controller") ctrl_t = await dev_t.add_object(0, "controller")
...@@ -786,7 +786,7 @@ async def test_copy_node(opc): ...@@ -786,7 +786,7 @@ async def test_copy_node(opc):
devd_t = await dev_t.add_object_type(0, "MyDeviceDervived") devd_t = await dev_t.add_object_type(0, "MyDeviceDervived")
v_t = await devd_t.add_variable(0, "childparam", 1.0) v_t = await devd_t.add_variable(0, "childparam", 1.0)
p_t = await devd_t.add_property(0, "sensorx_id", "0340") p_t = await devd_t.add_property(0, "sensorx_id", "0340")
nodes = await copy_node(opc.nodes.objects, dev_t) nodes = await copy_node(opc.opc.nodes.objects, dev_t)
mydevice = nodes[0] mydevice = nodes[0]
assert ua.NodeClass.ObjectType == await mydevice.get_node_class() assert ua.NodeClass.ObjectType == await mydevice.get_node_class()
assert 4 == len(await mydevice.get_children()) assert 4 == len(await mydevice.get_children())
...@@ -799,7 +799,7 @@ async def test_copy_node(opc): ...@@ -799,7 +799,7 @@ async def test_copy_node(opc):
async def test_instantiate_1(opc): async def test_instantiate_1(opc):
# Create device type # Create device type
dev_t = await opc.nodes.base_object_type.add_object_type(0, "MyDevice") dev_t = await opc.opc.nodes.base_object_type.add_object_type(0, "MyDevice")
v_t = await dev_t.add_variable(0, "sensor", 1.0) v_t = await dev_t.add_variable(0, "sensor", 1.0)
await v_t.set_modelling_rule(True) await v_t.set_modelling_rule(True)
p_t = await dev_t.add_property(0, "sensor_id", "0340") p_t = await dev_t.add_property(0, "sensor_id", "0340")
...@@ -821,7 +821,7 @@ async def test_instantiate_1(opc): ...@@ -821,7 +821,7 @@ async def test_instantiate_1(opc):
await p_t.set_modelling_rule(True) await p_t.set_modelling_rule(True)
# instanciate device # instanciate device
nodes = await instantiate(opc.nodes.objects, dev_t, bname="2:Device0001") nodes = await instantiate(opc.opc.nodes.objects, dev_t, bname="2:Device0001")
mydevice = nodes[0] mydevice = nodes[0]
assert ua.NodeClass.Object == await mydevice.get_node_class() assert ua.NodeClass.Object == await mydevice.get_node_class()
...@@ -838,7 +838,7 @@ async def test_instantiate_1(opc): ...@@ -838,7 +838,7 @@ async def test_instantiate_1(opc):
assert prop.nodeid != prop_t.nodeid assert prop.nodeid != prop_t.nodeid
# instanciate device subtype # instanciate device subtype
nodes = await instantiate(opc.nodes.objects, devd_t, bname="2:Device0002") nodes = await instantiate(opc.opc.nodes.objects, devd_t, bname="2:Device0002")
mydevicederived = nodes[0] mydevicederived = nodes[0]
prop1 = await mydevicederived.get_child(["0:sensorx_id"]) prop1 = await mydevicederived.get_child(["0:sensorx_id"])
var1 = await mydevicederived.get_child(["0:childparam"]) var1 = await mydevicederived.get_child(["0:childparam"])
...@@ -848,7 +848,7 @@ async def test_instantiate_1(opc): ...@@ -848,7 +848,7 @@ async def test_instantiate_1(opc):
async def test_instantiate_string_nodeid(opc): async def test_instantiate_string_nodeid(opc):
# Create device type # Create device type
dev_t = await opc.nodes.base_object_type.add_object_type(0, "MyDevice2") dev_t = await opc.opc.nodes.base_object_type.add_object_type(0, "MyDevice2")
v_t = await dev_t.add_variable(0, "sensor", 1.0) v_t = await dev_t.add_variable(0, "sensor", 1.0)
await v_t.set_modelling_rule(True) await v_t.set_modelling_rule(True)
p_t = await dev_t.add_property(0, "sensor_id", "0340") p_t = await dev_t.add_property(0, "sensor_id", "0340")
...@@ -859,7 +859,7 @@ async def test_instantiate_string_nodeid(opc): ...@@ -859,7 +859,7 @@ async def test_instantiate_string_nodeid(opc):
await prop_t.set_modelling_rule(True) await prop_t.set_modelling_rule(True)
# instanciate device # instanciate device
nodes = await instantiate(opc.nodes.objects, dev_t, nodeid=ua.NodeId("InstDevice", 2, ua.NodeIdType.String), nodes = await instantiate(opc.opc.nodes.objects, dev_t, nodeid=ua.NodeId("InstDevice", 2, ua.NodeIdType.String),
bname="2:InstDevice") bname="2:InstDevice")
mydevice = nodes[0] mydevice = nodes[0]
...@@ -875,13 +875,13 @@ async def test_instantiate_string_nodeid(opc): ...@@ -875,13 +875,13 @@ async def test_instantiate_string_nodeid(opc):
async def test_variable_with_datatype(opc): async def test_variable_with_datatype(opc):
v1 = await opc.nodes.objects.add_variable( v1 = await opc.opc.nodes.objects.add_variable(
3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType) 3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType)
) )
tp1 = await v1.get_data_type() tp1 = await v1.get_data_type()
assert tp1 == ua.NodeId(ua.ObjectIds.ApplicationType) assert tp1 == ua.NodeId(ua.ObjectIds.ApplicationType)
v2 = await opc.nodes.objects.add_variable( v2 = await opc.opc.nodes.objects.add_variable(
3, 'VariableEnumType2', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType) 3, 'VariableEnumType2', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType)
) )
tp2 = await v2.get_data_type() tp2 = await v2.get_data_type()
...@@ -890,7 +890,7 @@ async def test_variable_with_datatype(opc): ...@@ -890,7 +890,7 @@ async def test_variable_with_datatype(opc):
async def test_enum(opc): async def test_enum(opc):
# create enum type # create enum type
enums = await opc.get_root_node().get_child(["0:Types", "0:DataTypes", "0:BaseDataType", "0:Enumeration"]) enums = await opc.opc.get_root_node().get_child(["0:Types", "0:DataTypes", "0:BaseDataType", "0:Enumeration"])
myenum_type = await enums.add_data_type(0, "MyEnum") myenum_type = await enums.add_data_type(0, "MyEnum")
es = await myenum_type.add_variable( es = await myenum_type.add_variable(
0, "EnumStrings", [ua.LocalizedText("String0"), ua.LocalizedText("String1"), ua.LocalizedText("String2")], 0, "EnumStrings", [ua.LocalizedText("String0"), ua.LocalizedText("String1"), ua.LocalizedText("String2")],
...@@ -898,7 +898,7 @@ async def test_enum(opc): ...@@ -898,7 +898,7 @@ async def test_enum(opc):
) )
# es.set_value_rank(1) # es.set_value_rank(1)
# instantiate # instantiate
o = opc.get_objects_node() o = opc.opc.get_objects_node()
myvar = await o.add_variable(2, "MyEnumVar", ua.LocalizedText("String1"), datatype=myenum_type.nodeid) myvar = await o.add_variable(2, "MyEnumVar", ua.LocalizedText("String1"), datatype=myenum_type.nodeid)
# myvar.set_writable(True) # myvar.set_writable(True)
# tests # tests
...@@ -907,13 +907,13 @@ async def test_enum(opc): ...@@ -907,13 +907,13 @@ async def test_enum(opc):
async def test_supertypes(opc): async def test_supertypes(opc):
nint32 = opc.get_node(ua.ObjectIds.Int32) nint32 = opc.opc.get_node(ua.ObjectIds.Int32)
node = await ua_utils.get_node_supertype(nint32) node = await ua_utils.get_node_supertype(nint32)
assert opc.get_node(ua.ObjectIds.Integer) == node assert opc.opc.get_node(ua.ObjectIds.Integer) == node
nodes = await ua_utils.get_node_supertypes(nint32) nodes = await ua_utils.get_node_supertypes(nint32)
assert opc.get_node(ua.ObjectIds.Number) == nodes[1] assert opc.opc.get_node(ua.ObjectIds.Number) == nodes[1]
assert opc.get_node(ua.ObjectIds.Integer) == nodes[0] assert opc.opc.get_node(ua.ObjectIds.Integer) == nodes[0]
# test custom # test custom
dtype = await nint32.add_data_type(0, "MyCustomDataType") dtype = await nint32.add_data_type(0, "MyCustomDataType")
...@@ -926,16 +926,16 @@ async def test_supertypes(opc): ...@@ -926,16 +926,16 @@ async def test_supertypes(opc):
async def test_base_data_type(opc): async def test_base_data_type(opc):
nint32 = opc.get_node(ua.ObjectIds.Int32) nint32 = opc.opc.get_node(ua.ObjectIds.Int32)
dtype = await nint32.add_data_type(0, "MyCustomDataType") dtype = await nint32.add_data_type(0, "MyCustomDataType")
dtype2 = await dtype.add_data_type(0, "MyCustomDataType2") dtype2 = await dtype.add_data_type(0, "MyCustomDataType2")
assert nint32 == await ua_utils.get_base_data_type(dtype) assert nint32 == await ua_utils.get_base_data_type(dtype)
assert nint32 == await ua_utils.get_base_data_type(dtype2) assert nint32 == await ua_utils.get_base_data_type(dtype2)
ext = await opc.nodes.objects.add_variable(0, "MyExtensionObject", ua.Argument()) ext = await opc.opc.nodes.objects.add_variable(0, "MyExtensionObject", ua.Argument())
d = await ext.get_data_type() d = await ext.get_data_type()
d = opc.get_node(d) d = opc.opc.get_node(d)
assert opc.get_node(ua.ObjectIds.Structure) == await ua_utils.get_base_data_type(d) assert opc.opc.get_node(ua.ObjectIds.Structure) == await ua_utils.get_base_data_type(d)
assert ua.VariantType.ExtensionObject == await ua_utils.data_type_to_variant_type(d) assert ua.VariantType.ExtensionObject == await ua_utils.data_type_to_variant_type(d)
...@@ -955,4 +955,4 @@ async def test_data_type_to_variant_type(opc): ...@@ -955,4 +955,4 @@ async def test_data_type_to_variant_type(opc):
ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32 # enumeration ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32 # enumeration
} }
for dt, vdt in test_data.items(): for dt, vdt in test_data.items():
assert vdt == await ua_utils.data_type_to_variant_type(opc.get_node(ua.NodeId(dt))) assert vdt == await ua_utils.data_type_to_variant_type(opc.opc.get_node(ua.NodeId(dt)))
tests/test_server.py
View file @ 4221be7d
...@@ -7,12 +7,7 @@ import pytest ...@@ -7,12 +7,7 @@ import pytest
import logging import logging
import os import os
import shelve import shelve
from datetime import timedelta
from .test_common import add_server_methods
from .tests_xml import XmlTests
from .tests_subscriptions import SubscriptionTests
from datetime import timedelta, datetime
from tempfile import NamedTemporaryFile
import opcua import opcua
from opcua import Server from opcua import Server
...@@ -23,38 +18,10 @@ from opcua.common.event_objects import BaseEvent, AuditEvent, AuditChannelEvent, ...@@ -23,38 +18,10 @@ from opcua.common.event_objects import BaseEvent, AuditEvent, AuditChannelEvent,
AuditOpenSecureChannelEvent AuditOpenSecureChannelEvent
from opcua.common import ua_utils from opcua.common import ua_utils
port_num = 48540
port_discovery = 48550
pytestmark = pytest.mark.asyncio pytestmark = pytest.mark.asyncio
_logger = logging.getLogger(__name__) _logger = logging.getLogger(__name__)
@pytest.fixture()
async def server():
# start our own server
srv = Server()
await srv.init()
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_num}')
await add_server_methods(srv)
await srv.start()
yield srv
# stop the server
await srv.stop()
@pytest.fixture()
async def discovery_server():
# start our own server
srv = Server()
await srv.init()
await srv.set_application_uri('urn:freeopcua:python:discovery')
srv.set_endpoint(f'opc.tcp://127.0.0.1:{port_discovery}')
await srv.start()
yield srv
# stop the server
await srv.stop()
async def test_discovery(server, discovery_server): async def test_discovery(server, discovery_server):
client = Client(discovery_server.endpoint.geturl()) client = Client(discovery_server.endpoint.geturl())
async with client: async with client:
......
tests/test_subscriptions.py 0 → 100644
View file @ 4221be7d
import time
import pytest
from asyncio import Future, sleep, wait_for
from datetime import datetime, timedelta
import opcua
from opcua import ua
pytestmark = pytest.mark.asyncio
class SubHandler:
"""
Dummy subscription client
"""
def datachange_notification(self, node, val, data):
pass
def event_notification(self, event):
pass
class MySubHandler:
"""
More advanced subscription client using Future, so we can wait for events in tests
"""
def __init__(self):
self.future = Future()
def reset(self):
self.future = Future()
async def result(self):
return await wait_for(self.future, 2)
def datachange_notification(self, node, val, data):
self.future.set_result((node, val, data))
def event_notification(self, event):
self.future.set_result(event)
class MySubHandler2:
def __init__(self):
self.results = []
def datachange_notification(self, node, val, data):
self.results.append((node, val))
def event_notification(self, event):
self.results.append(event)
class MySubHandlerCounter():
def __init__(self):
self.datachange_count = 0
self.event_count = 0
def datachange_notification(self, node, val, data):
self.datachange_count += 1
def event_notification(self, event):
self.event_count += 1
async def test_subscription_failure(opc):
myhandler = MySubHandler()
o = opc.get_objects_node()
sub = await opc.create_subscription(100, myhandler)
with pytest.raises(ua.UaStatusCodeError):
handle1 = sub.subscribe_data_change(o) # we can only subscribe to variables so this should fail
sub.delete()
async def test_subscription_overload(opc):
nb = 10
myhandler = MySubHandler()
o = opc.opc.get_objects_node()
sub = await opc.opc.create_subscription(1, myhandler)
variables = []
subs = []
for i in range(nb):
v = await o.add_variable(3, f'SubscriptionVariableOverload{i}', 99)
variables.append(v)
for i in range(nb):
sub.subscribe_data_change(variables)
for i in range(nb):
for j in range(nb):
await variables[i].set_value(j)
s = await opc.opc.create_subscription(1, myhandler)
s.subscribe_data_change(variables)
subs.append(s)
sub.subscribe_data_change(variables[i])
for i in range(nb):
for j in range(nb):
await variables[i].set_value(j)
sub.delete()
for s in subs:
s.delete()
async def test_subscription_count(opc):
myhandler = MySubHandlerCounter()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.get_objects_node()
var = await o.add_variable(3, 'SubVarCounter', 0.1)
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = await var.get_value()
await var.set_value(val + 1)
await sleep(0.2) # let last event arrive
assert nb + 1 == myhandler.datachange_count
sub.delete()
async def test_subscription_count_list(opc):
myhandler = MySubHandlerCounter()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.get_objects_node()
var = await o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = await var.get_value()
val.append(i)
await var.set_value(val)
await sleep(0.2) # let last event arrive
assert nb + 1 == myhandler.datachange_count
sub.delete()
async def test_subscription_count_no_change(opc):
myhandler = MySubHandlerCounter()
sub = opc.create_subscription(1, myhandler)
o = opc.get_objects_node()
var = o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = var.get_value()
var.set_value(val)
await sleep(0.2) # let last event arrive
assert 1 == myhandler.datachange_count
sub.delete()
async def test_subscription_count_empty(opc):
myhandler = MySubHandlerCounter()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.get_objects_node()
var = await o.add_variable(3, 'SubVarCounter', [0.1, 0.2, 0.3])
sub.subscribe_data_change(var)
while True:
val = await var.get_value()
val.pop()
await var.set_value(val, ua.VariantType.Double)
if not val:
break
await sleep(0.2) # let last event arrive
assert 4 == myhandler.datachange_count
sub.delete()
async def test_subscription_overload_simple(opc):
nb = 10
myhandler = MySubHandler()
o = opc.opc.get_objects_node()
sub = await opc.opc.create_subscription(1, myhandler)
variables = []
for i in range(nb):
variables.append(await o.add_variable(3, f'SubVarOverload{i}', i))
for i in range(nb):
sub.subscribe_data_change(variables)
sub.delete()
async def test_subscription_data_change(opc):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler()
o = opc.opc.get_objects_node()
# subscribe to a variable
startv1 = [1, 2, 3]
v1 = await o.add_variable(3, 'SubscriptionVariableV1', startv1)
sub = await opc.opc.create_subscription(100, myhandler)
handle1 = await sub.subscribe_data_change(v1)
# Now check we get the start value
node, val, data = await myhandler.result()
assert startv1 == val
assert v1 == node
myhandler.reset() # reset future object
# modify v1 and check we get value
await v1.set_value([5])
node, val, data = await myhandler.result()
assert v1 == node
assert [5] == val
with pytest.raises(ua.UaStatusCodeError):
await sub.unsubscribe(999) # non existing handle
await sub.unsubscribe(handle1)
with pytest.raises(ua.UaStatusCodeError):
await sub.unsubscribe(handle1) # second try should fail
sub.delete()
with pytest.raises(ua.UaStatusCodeError):
await sub.unsubscribe(handle1) # sub does not exist anymore
async def test_subscription_data_change_bool(opc):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler()
o = opc.opc.get_objects_node()
# subscribe to a variable
startv1 = True
v1 = await o.add_variable(3, 'SubscriptionVariableBool', startv1)
sub = await opc.opc.create_subscription(100, myhandler)
handle1 = await sub.subscribe_data_change(v1)
# Now check we get the start value
node, val, data = await myhandler.result()
assert startv1 == val
assert v1 == node
myhandler.reset() # reset future object
# modify v1 and check we get value
await v1.set_value(False)
node, val, data = await myhandler.result()
assert v1 == node
assert val is False
sub.delete() # should delete our monitoreditem too
async def test_subscription_data_change_many(opc):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler2()
o = opc.opc.get_objects_node()
startv1 = True
v1 = await o.add_variable(3, 'SubscriptionVariableMany1', startv1)
startv2 = [1.22, 1.65]
v2 = await o.add_variable(3, 'SubscriptionVariableMany2', startv2)
sub = await opc.opc.create_subscription(100, myhandler)
handle1, handle2 = await sub.subscribe_data_change([v1, v2])
# Now check we get the start values
nodes = [v1, v2]
count = 0
while not len(myhandler.results) > 1:
count += 1
await sleep(0.1)
if count > 100:
raise RuntimeError("Did not get result from subscription")
for node, val in myhandler.results:
assert node in nodes
nodes.remove(node)
if node == v1:
assert val == startv1
elif node == v2:
assert val == startv2
else:
raise RuntimeError("Error node {0} is neither {1} nor {2}".format(node, v1, v2))
sub.delete()
async def test_subscribe_server_time(opc):
myhandler = MySubHandler()
server_time_node = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
sub = await opc.opc.create_subscription(200, myhandler)
handle = await sub.subscribe_data_change(server_time_node)
node, val, data = await myhandler.result()
assert server_time_node == node
delta = datetime.utcnow() - val
assert delta < timedelta(seconds=2)
await sub.unsubscribe(handle)
sub.delete()
async def test_create_delete_subscription(opc):
o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'SubscriptionVariable', [1, 2, 3])
sub = await opc.opc.create_subscription(100, MySubHandler())
handle = await sub.subscribe_data_change(v)
await sleep(0.1)
await sub.unsubscribe(handle)
sub.delete()
async def test_subscribe_events(opc):
sub = await opc.opc.create_subscription(100, MySubHandler())
handle = await sub.subscribe_events()
await sleep(0.1)
await sub.unsubscribe(handle)
sub.delete()
async def test_subscribe_events_to_wrong_node(opc):
sub = await opc.opc.create_subscription(100, MySubHandler())
with pytest.raises(ua.UaStatusCodeError):
handle = await sub.subscribe_events(opc.opc.get_node("i=85"))
o = opc.opc.get_objects_node()
v = await o.add_variable(3, 'VariableNoEventNofierAttribute', 4)
with pytest.raises(ua.UaStatusCodeError):
handle = await sub.subscribe_events(v)
sub.delete()
async def test_get_event_from_type_node_BaseEvent(opc):
etype = opc.opc.get_node(ua.ObjectIds.BaseEventType)
properties = await opcua.common.events.get_event_properties_from_type_node(etype)
for child in await etype.get_properties():
assert child in properties
async def test_get_event_from_type_node_CustomEvent(opc):
etype = await opc.server.create_custom_event_type(
2, 'MyEvent', ua.ObjectIds.AuditEventType,
[('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
properties = await opcua.common.events.get_event_properties_from_type_node(etype)
for child in await opc.opc.get_node(ua.ObjectIds.BaseEventType).get_properties():
assert child in properties
for child in await opc.opc.get_node(ua.ObjectIds.AuditEventType).get_properties():
assert child in properties
for child in await opc.opc.get_node(etype.nodeid).get_properties():
assert child in properties
assert await etype.get_child("2:PropertyNum") in properties
assert await etype.get_child("2:PropertyString") in properties
async def test_events_default(opc):
evgen = await opc.server.get_event_generator()
myhandler = MySubHandler()
sub = await opc.opc.create_subscription(100, myhandler)
handle = await sub.subscribe_events()
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = await myhandler.result()
assert ev is not None # we did not receive event
assert ua.NodeId(ua.ObjectIds.BaseEventType) == ev.EventType
assert 1 == ev.Severity
assert (await opc.opc.get_server_node().get_browse_name()).Name == ev.SourceName
assert opc.opc.get_server_node().nodeid == ev.SourceNode
assert msg == ev.Message.Text
assert tid == ev.Time
await sub.unsubscribe(handle)
sub.delete()
async def test_events_MyObject(opc):
objects = opc.server.get_objects_node()
o = await objects.add_object(3, 'MyObject')
evgen = await opc.server.get_event_generator(source=o)
myhandler = MySubHandler()
sub = await opc.opc.create_subscription(100, myhandler)
handle = await sub.subscribe_events(o)
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = await myhandler.result()
assert ev is not None # we did not receive event
assert ua.NodeId(ua.ObjectIds.BaseEventType) == ev.EventType
assert 1 == ev.Severity
assert 'MyObject' == ev.SourceName
assert o.nodeid == ev.SourceNode
assert msg == ev.Message.Text
assert tid == ev.Time
await sub.unsubscribe(handle)
sub.delete()
async def test_events_wrong_source(opc):
objects = opc.server.get_objects_node()
o = await objects.add_object(3, 'MyObject')
evgen = await opc.server.get_event_generator(source=o)
myhandler = MySubHandler()
sub = await opc.opc.create_subscription(100, myhandler)
handle = await sub.subscribe_events()
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
with pytest.raises(TimeoutError): # we should not receive event
ev = await myhandler.result()
await sub.unsubscribe(handle)
sub.delete()
async def test_events_CustomEvent(opc):
etype = opc.server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float),
('PropertyString', ua.VariantType.String)])
evgen = opc.server.get_event_generator(etype)
myhandler = MySubHandler()
sub = opc.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(evtypes=etype)
propertynum = 2
propertystring = "This is my test"
evgen.event.PropertyNum = propertynum
evgen.event.PropertyString = propertystring
serverity = 500
evgen.event.Severity = serverity
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = await myhandler.result()
assert ev is not None # we did not receive event
assert etype.nodeid == ev.EventType
assert serverity == ev.Severity
assert opc.opc.get_server_node().get_browse_name().Name == ev.SourceName
assert opc.opc.get_server_node().nodeid == ev.SourceNode
assert msg == ev.Message.Text
assert tid == ev.Time
assert propertynum == ev.PropertyNum
assert propertystring == ev.PropertyString
sub.unsubscribe(handle)
sub.delete()
async def test_events_CustomEvent_MyObject(opc):
objects = opc.server.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype = opc.server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float),
('PropertyString', ua.VariantType.String)])
evgen = opc.server.get_event_generator(etype, o)
myhandler = MySubHandler()
sub = opc.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, etype)
propertynum = 2
propertystring = "This is my test"
evgen.event.PropertyNum = propertynum
evgen.event.PropertyString = propertystring
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = await myhandler.result()
assert ev is not None # we did not receive event
assert etype.nodeid == ev.EventType
assert 1 == ev.Severity
assert 'MyObject' == ev.SourceName
assert o.nodeid == ev.SourceNode
assert msg == ev.Message.Text
assert tid == ev.Time
assert propertynum == ev.PropertyNum
assert propertystring == ev.PropertyString
sub.unsubscribe(handle)
sub.delete()
async def test_several_different_events(opc):
objects = opc.server.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype1 = opc.server.create_custom_event_type(2, 'MyEvent1', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float),
('PropertyString', ua.VariantType.String)])
evgen1 = await opc.server.get_event_generator(etype1, o)
etype2 = await opc.server.create_custom_event_type(2, 'MyEvent2', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float),
('PropertyString', ua.VariantType.String)])
evgen2 = await opc.server.get_event_generator(etype2, o)
myhandler = MySubHandler2()
sub = await opc.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, etype1)
propertynum1 = 1
propertystring1 = "This is my test 1"
evgen1.event.PropertyNum = propertynum1
evgen1.event.PropertyString = propertystring1
propertynum2 = 2
propertystring2 = "This is my test 2"
evgen2.event.PropertyNum = propertynum2
evgen2.event.PropertyString = propertystring2
for i in range(3):
evgen1.trigger()
evgen2.trigger()
await sleep(1) # ToDo: replace
assert 3 == len(myhandler.results)
ev = myhandler.results[-1]
assert etype1.nodeid == ev.EventType
handle = sub.subscribe_events(o, etype2)
for i in range(4):
evgen1.trigger()
evgen2.trigger()
await sleep(1) # ToDo: replace
ev1s = [ev for ev in myhandler.results if ev.EventType == etype1.nodeid]
ev2s = [ev for ev in myhandler.results if ev.EventType == etype2.nodeid]
assert 11 == len(myhandler.results)
assert 4 == len(ev2s)
assert 7 == len(ev1s)
sub.unsubscribe(handle)
sub.delete()
async def test_several_different_events_2(opc):
objects = opc.server.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype1 = opc.server.create_custom_event_type(
2, 'MyEvent1', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen1 = opc.server.get_event_generator(etype1, o)
etype2 = opc.server.create_custom_event_type(
2, 'MyEvent2', ua.ObjectIds.BaseEventType,
[('PropertyNum2', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen2 = opc.server.get_event_generator(etype2, o)
etype3 = opc.server.create_custom_event_type(
2, 'MyEvent3', ua.ObjectIds.BaseEventType,
[('PropertyNum3', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)]
)
evgen3 = opc.server.get_event_generator(etype3, o)
myhandler = MySubHandler2()
sub = opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, [etype1, etype3])
propertynum1 = 1
propertystring1 = "This is my test 1"
evgen1.event.PropertyNum = propertynum1
evgen1.event.PropertyString = propertystring1
propertynum2 = 2
propertystring2 = "This is my test 2"
evgen2.event.PropertyNum2 = propertynum2
evgen2.event.PropertyString = propertystring2
propertynum3 = 3
propertystring3 = "This is my test 3"
evgen3.event.PropertyNum3 = propertynum3
evgen3.event.PropertyString = propertystring2
for i in range(3):
evgen1.trigger()
evgen2.trigger()
evgen3.trigger()
evgen3.event.PropertyNum3 = 9999
evgen3.trigger()
await sleep(1)
ev1s = [ev for ev in myhandler.results if ev.EventType == etype1.nodeid]
ev2s = [ev for ev in myhandler.results if ev.EventType == etype2.nodeid]
ev3s = [ev for ev in myhandler.results if ev.EventType == etype3.nodeid]
assert 7 == len(myhandler.results)
assert 3 == len(ev1s)
assert 0 == len(ev2s)
assert 4 == len(ev3s)
assert propertynum1 == ev1s[0].PropertyNum
assert propertynum3 == ev3s[0].PropertyNum3
assert 9999 == ev3s[-1].PropertyNum3
assert ev1s[0].PropertyNum3 is None
sub.unsubscribe(handle)
sub.delete()
tests/tests_subscriptions.py deleted 100644 → 0
View file @ 006689dc
from concurrent.futures import Future, TimeoutError
import time
from datetime import datetime, timedelta
import opcua
from opcua import ua
class SubHandler():
"""
Dummy subscription client
"""
def datachange_notification(self, node, val, data):
pass
def event_notification(self, event):
pass
class MySubHandler():
"""
More advanced subscription client using Future, so we can wait for events in tests
"""
def __init__(self):
self.future = Future()
def reset(self):
self.future = Future()
def datachange_notification(self, node, val, data):
self.future.set_result((node, val, data))
def event_notification(self, event):
self.future.set_result(event)
class MySubHandler2():
def __init__(self):
self.results = []
def datachange_notification(self, node, val, data):
self.results.append((node, val))
def event_notification(self, event):
self.results.append(event)
class MySubHandlerCounter():
def __init__(self):
self.datachange_count = 0
self.event_count = 0
def datachange_notification(self, node, val, data):
self.datachange_count += 1
def event_notification(self, event):
self.event_count += 1
class SubscriptionTests(object):
def test_subscription_failure(self):
myhandler = MySubHandler()
o = self.opc.get_objects_node()
sub = self.opc.create_subscription(100, myhandler)
with self.assertRaises(ua.UaStatusCodeError):
handle1 = sub.subscribe_data_change(o) # we can only subscribe to variables so this should fail
sub.delete()
def test_subscription_overload(self):
nb = 10
myhandler = MySubHandler()
o = self.opc.get_objects_node()
sub = self.opc.create_subscription(1, myhandler)
variables = []
subs = []
for i in range(nb):
v = o.add_variable(3, 'SubscriptionVariableOverload' + str(i), 99)
variables.append(v)
for i in range(nb):
sub.subscribe_data_change(variables)
for i in range(nb):
for j in range(nb):
variables[i].set_value(j)
s = self.opc.create_subscription(1, myhandler)
s.subscribe_data_change(variables)
subs.append(s)
sub.subscribe_data_change(variables[i])
for i in range(nb):
for j in range(nb):
variables[i].set_value(j)
sub.delete()
for s in subs:
s.delete()
def test_subscription_count(self):
myhandler = MySubHandlerCounter()
sub = self.opc.create_subscription(1, myhandler)
o = self.opc.get_objects_node()
var = o.add_variable(3, 'SubVarCounter', 0.1)
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = var.get_value()
var.set_value(val +1)
time.sleep(0.2) # let last event arrive
self.assertEqual(myhandler.datachange_count, nb + 1)
sub.delete()
def test_subscription_count_list(self):
myhandler = MySubHandlerCounter()
sub = self.opc.create_subscription(1, myhandler)
o = self.opc.get_objects_node()
var = o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = var.get_value()
val.append(i)
var.set_value(val)
time.sleep(0.2) # let last event arrive
self.assertEqual(myhandler.datachange_count, nb + 1)
sub.delete()
def test_subscription_count_no_change(self):
myhandler = MySubHandlerCounter()
sub = self.opc.create_subscription(1, myhandler)
o = self.opc.get_objects_node()
var = o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
val = var.get_value()
var.set_value(val)
time.sleep(0.2) # let last event arrive
self.assertEqual(myhandler.datachange_count, 1)
sub.delete()
def test_subscription_count_empty(self):
myhandler = MySubHandlerCounter()
sub = self.opc.create_subscription(1, myhandler)
o = self.opc.get_objects_node()
var = o.add_variable(3, 'SubVarCounter', [0.1, 0.2, 0.3])
sub.subscribe_data_change(var)
while True:
val = var.get_value()
val.pop()
var.set_value(val, ua.VariantType.Double)
if not val:
break
time.sleep(0.2) # let last event arrive
self.assertEqual(myhandler.datachange_count, 4)
sub.delete()
def test_subscription_overload_simple(self):
nb = 10
myhandler = MySubHandler()
o = self.opc.get_objects_node()
sub = self.opc.create_subscription(1, myhandler)
variables = [o.add_variable(3, 'SubVarOverload' + str(i), i) for i in range(nb)]
for i in range(nb):
sub.subscribe_data_change(variables)
sub.delete()
def test_subscription_data_change(self):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler()
o = self.opc.get_objects_node()
# subscribe to a variable
startv1 = [1, 2, 3]
v1 = o.add_variable(3, 'SubscriptionVariableV1', startv1)
sub = self.opc.create_subscription(100, myhandler)
handle1 = sub.subscribe_data_change(v1)
# Now check we get the start value
node, val, data = myhandler.future.result()
self.assertEqual(val, startv1)
self.assertEqual(node, v1)
myhandler.reset() # reset future object
# modify v1 and check we get value
v1.set_value([5])
node, val, data = myhandler.future.result()
self.assertEqual(node, v1)
self.assertEqual(val, [5])
with self.assertRaises(ua.UaStatusCodeError):
sub.unsubscribe(999) # non existing handle
sub.unsubscribe(handle1)
with self.assertRaises(ua.UaStatusCodeError):
sub.unsubscribe(handle1) # second try should fail
sub.delete()
with self.assertRaises(ua.UaStatusCodeError):
sub.unsubscribe(handle1) # sub does not exist anymore
def test_subscription_data_change_bool(self):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler()
o = self.opc.get_objects_node()
# subscribe to a variable
startv1 = True
v1 = o.add_variable(3, 'SubscriptionVariableBool', startv1)
sub = self.opc.create_subscription(100, myhandler)
handle1 = sub.subscribe_data_change(v1)
# Now check we get the start value
node, val, data = myhandler.future.result()
self.assertEqual(val, startv1)
self.assertEqual(node, v1)
myhandler.reset() # reset future object
# modify v1 and check we get value
v1.set_value(False)
node, val, data = myhandler.future.result()
self.assertEqual(node, v1)
self.assertEqual(val, False)
sub.delete() # should delete our monitoreditem too
def test_subscription_data_change_many(self):
"""
test subscriptions. This is far too complicated for
a unittest but, setting up subscriptions requires a lot
of code, so when we first set it up, it is best
to test as many things as possible
"""
myhandler = MySubHandler2()
o = self.opc.get_objects_node()
startv1 = True
v1 = o.add_variable(3, 'SubscriptionVariableMany1', startv1)
startv2 = [1.22, 1.65]
v2 = o.add_variable(3, 'SubscriptionVariableMany2', startv2)
sub = self.opc.create_subscription(100, myhandler)
handle1, handle2 = sub.subscribe_data_change([v1, v2])
# Now check we get the start values
nodes = [v1, v2]
count = 0
while not len(myhandler.results) > 1:
count += 1
time.sleep(0.1)
if count > 100:
self.fail("Did not get result from subscription")
for node, val in myhandler.results:
self.assertIn(node, nodes)
nodes.remove(node)
if node == v1:
self.assertEqual(startv1, val)
elif node == v2:
self.assertEqual(startv2, val)
else:
self.fail("Error node {0} is neither {1} nor {2}".format(node, v1, v2))
sub.delete()
def test_subscribe_server_time(self):
myhandler = MySubHandler()
server_time_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
sub = self.opc.create_subscription(200, myhandler)
handle = sub.subscribe_data_change(server_time_node)
node, val, data = myhandler.future.result()
self.assertEqual(node, server_time_node)
delta = datetime.utcnow() - val
self.assertTrue(delta < timedelta(seconds=2))
sub.unsubscribe(handle)
sub.delete()
def test_create_delete_subscription(self):
o = self.opc.get_objects_node()
v = o.add_variable(3, 'SubscriptionVariable', [1, 2, 3])
sub = self.opc.create_subscription(100, MySubHandler())
handle = sub.subscribe_data_change(v)
time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_subscribe_events(self):
sub = self.opc.create_subscription(100, MySubHandler())
handle = sub.subscribe_events()
time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_subscribe_events_to_wrong_node(self):
sub = self.opc.create_subscription(100, MySubHandler())
with self.assertRaises(ua.UaStatusCodeError):
handle = sub.subscribe_events(self.opc.get_node("i=85"))
o = self.opc.get_objects_node()
v = o.add_variable(3, 'VariableNoEventNofierAttribute', 4)
with self.assertRaises(ua.UaStatusCodeError):
handle = sub.subscribe_events(v)
sub.delete()
def test_get_event_from_type_node_BaseEvent(self):
etype = self.opc.get_node(ua.ObjectIds.BaseEventType)
properties = opcua.common.events.get_event_properties_from_type_node(etype)
for child in etype.get_properties():
self.assertTrue(child in properties)
def test_get_event_from_type_node_CustomEvent(self):
etype = self.srv.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.AuditEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
properties = opcua.common.events.get_event_properties_from_type_node(etype)
for child in self.opc.get_node(ua.ObjectIds.BaseEventType).get_properties():
self.assertTrue(child in properties)
for child in self.opc.get_node(ua.ObjectIds.AuditEventType).get_properties():
self.assertTrue(child in properties)
for child in self.opc.get_node(etype.nodeid).get_properties():
self.assertTrue(child in properties)
self.assertTrue(etype.get_child("2:PropertyNum") in properties)
self.assertTrue(etype.get_child("2:PropertyString") in properties)
def test_events_default(self):
evgen = self.srv.get_event_generator()
myhandler = MySubHandler()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events()
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = myhandler.future.result()
self.assertIsNot(ev, None) # we did not receive event
self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType))
self.assertEqual(ev.Severity, 1)
self.assertEqual(ev.SourceName, self.opc.get_server_node().get_browse_name().Name)
self.assertEqual(ev.SourceNode, self.opc.get_server_node().nodeid)
self.assertEqual(ev.Message.Text, msg)
self.assertEqual(ev.Time, tid)
# time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_events_MyObject(self):
objects = self.srv.get_objects_node()
o = objects.add_object(3, 'MyObject')
evgen = self.srv.get_event_generator(source=o)
myhandler = MySubHandler()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o)
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = myhandler.future.result(10)
self.assertIsNot(ev, None) # we did not receive event
self.assertEqual(ev.EventType, ua.NodeId(ua.ObjectIds.BaseEventType))
self.assertEqual(ev.Severity, 1)
self.assertEqual(ev.SourceName, 'MyObject')
self.assertEqual(ev.SourceNode, o.nodeid)
self.assertEqual(ev.Message.Text, msg)
self.assertEqual(ev.Time, tid)
# time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_events_wrong_source(self):
objects = self.srv.get_objects_node()
o = objects.add_object(3, 'MyObject')
evgen = self.srv.get_event_generator(source=o)
myhandler = MySubHandler()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events()
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
with self.assertRaises(TimeoutError): # we should not receive event
ev = myhandler.future.result(2)
# time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_events_CustomEvent(self):
etype = self.srv.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen = self.srv.get_event_generator(etype)
myhandler = MySubHandler()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(evtypes=etype)
propertynum = 2
propertystring = "This is my test"
evgen.event.PropertyNum = propertynum
evgen.event.PropertyString = propertystring
serverity = 500
evgen.event.Severity = serverity
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = myhandler.future.result(10)
self.assertIsNot(ev, None) # we did not receive event
self.assertEqual(ev.EventType, etype.nodeid)
self.assertEqual(ev.Severity, serverity)
self.assertEqual(ev.SourceName, self.opc.get_server_node().get_browse_name().Name)
self.assertEqual(ev.SourceNode, self.opc.get_server_node().nodeid)
self.assertEqual(ev.Message.Text, msg)
self.assertEqual(ev.Time, tid)
self.assertEqual(ev.PropertyNum, propertynum)
self.assertEqual(ev.PropertyString, propertystring)
# time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_events_CustomEvent_MyObject(self):
objects = self.srv.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype = self.srv.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen = self.srv.get_event_generator(etype, o)
myhandler = MySubHandler()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, etype)
propertynum = 2
propertystring = "This is my test"
evgen.event.PropertyNum = propertynum
evgen.event.PropertyString = propertystring
tid = datetime.utcnow()
msg = "this is my msg "
evgen.trigger(tid, msg)
ev = myhandler.future.result(10)
self.assertIsNot(ev, None) # we did not receive event
self.assertEqual(ev.EventType, etype.nodeid)
self.assertEqual(ev.Severity, 1)
self.assertEqual(ev.SourceName, 'MyObject')
self.assertEqual(ev.SourceNode, o.nodeid)
self.assertEqual(ev.Message.Text, msg)
self.assertEqual(ev.Time, tid)
self.assertEqual(ev.PropertyNum, propertynum)
self.assertEqual(ev.PropertyString, propertystring)
# time.sleep(0.1)
sub.unsubscribe(handle)
sub.delete()
def test_several_different_events(self):
objects = self.srv.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype1 = self.srv.create_custom_event_type(2, 'MyEvent1', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen1 = self.srv.get_event_generator(etype1, o)
etype2 = self.srv.create_custom_event_type(2, 'MyEvent2', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen2 = self.srv.get_event_generator(etype2, o)
myhandler = MySubHandler2()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, etype1)
propertynum1 = 1
propertystring1 = "This is my test 1"
evgen1.event.PropertyNum = propertynum1
evgen1.event.PropertyString = propertystring1
propertynum2 = 2
propertystring2 = "This is my test 2"
evgen2.event.PropertyNum = propertynum2
evgen2.event.PropertyString = propertystring2
for i in range(3):
evgen1.trigger()
evgen2.trigger()
time.sleep(1)
self.assertEqual(len(myhandler.results), 3)
ev = myhandler.results[-1]
self.assertEqual(ev.EventType, etype1.nodeid)
handle = sub.subscribe_events(o, etype2)
for i in range(4):
evgen1.trigger()
evgen2.trigger()
time.sleep(1)
ev1s = [ev for ev in myhandler.results if ev.EventType == etype1.nodeid]
ev2s = [ev for ev in myhandler.results if ev.EventType == etype2.nodeid]
self.assertEqual(len(myhandler.results), 11)
self.assertEqual(len(ev2s), 4)
self.assertEqual(len(ev1s), 7)
sub.unsubscribe(handle)
sub.delete()
def test_several_different_events_2(self):
objects = self.srv.get_objects_node()
o = objects.add_object(3, 'MyObject')
etype1 = self.srv.create_custom_event_type(2, 'MyEvent1', ua.ObjectIds.BaseEventType, [('PropertyNum', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen1 = self.srv.get_event_generator(etype1, o)
etype2 = self.srv.create_custom_event_type(2, 'MyEvent2', ua.ObjectIds.BaseEventType, [('PropertyNum2', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen2 = self.srv.get_event_generator(etype2, o)
etype3 = self.srv.create_custom_event_type(2, 'MyEvent3', ua.ObjectIds.BaseEventType, [('PropertyNum3', ua.VariantType.Float), ('PropertyString', ua.VariantType.String)])
evgen3 = self.srv.get_event_generator(etype3, o)
myhandler = MySubHandler2()
sub = self.opc.create_subscription(100, myhandler)
handle = sub.subscribe_events(o, [etype1, etype3])
propertynum1 = 1
propertystring1 = "This is my test 1"
evgen1.event.PropertyNum = propertynum1
evgen1.event.PropertyString = propertystring1
propertynum2 = 2
propertystring2 = "This is my test 2"
evgen2.event.PropertyNum2 = propertynum2
evgen2.event.PropertyString = propertystring2
propertynum3 = 3
propertystring3 = "This is my test 3"
evgen3.event.PropertyNum3 = propertynum3
evgen3.event.PropertyString = propertystring2
for i in range(3):
evgen1.trigger()
evgen2.trigger()
evgen3.trigger()
evgen3.event.PropertyNum3 = 9999
evgen3.trigger()
time.sleep(1)
ev1s = [ev for ev in myhandler.results if ev.EventType == etype1.nodeid]
ev2s = [ev for ev in myhandler.results if ev.EventType == etype2.nodeid]
ev3s = [ev for ev in myhandler.results if ev.EventType == etype3.nodeid]
self.assertEqual(len(myhandler.results), 7)
self.assertEqual(len(ev1s), 3)
self.assertEqual(len(ev2s), 0)
self.assertEqual(len(ev3s), 4)
self.assertEqual(ev1s[0].PropertyNum, propertynum1)
self.assertEqual(ev3s[0].PropertyNum3, propertynum3)
self.assertEqual(ev3s[-1].PropertyNum3, 9999)
self.assertEqual(ev1s[0].PropertyNum3, None)
sub.unsubscribe(handle)
sub.delete()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7