Skip to content

O
opcua-asyncio
Commit 6a913f89 authored by Fabian Beitler's avatar Fabian Beitler Committed by GitHub
Browse files
Options

Add check if BrowseName already exists in ParentNode (#306) 

* Add check if BrowseName already exists in ParentNode

* Replace try-except blocks
Log error in check_method, not everywhere else

* Remove checks, because they are not cought if UaClient sends addNode request

* Add test to try identifing if BrowseName of new Node is child of ParentNode. If yes, raise error.If Node is root, ignore test.

* Fix tests, because multiple identical BrowseNames with the same parent Node is not allowed.

* Remove exception and ignore root Node directly

* Fix test by change Browsename

* Tidy up codeformat

* Try to get all sorts of NodeId child classes
Delete root Node parent check

* Remove empty aspace arg
Check if Reference is a HasChild Reference or a subtype of it

* Change pytest fixtures to function to not interfere with other tests
Add check if Nodes are in the same Namespace as well
Add duplicated browsename test

* Seperate between root Node or others
As well pushing test again...

* Add more Meta information about Node positions

* Add more Meta information about Nodes which seems to fail

* Change Browsename since both Nodes contain the same Browsename

* Replace dirty print with log

* Register Namespace and import Nodes, because Tests have been interfered eachother

* Try if test interferes with others

* Add check to catch faulty preperation

* Revert "Try if test interferes with others"

This reverts commit b0ac20dc2e898a70a7dedb433da9b88dfb2dc3fd.
Revert "Add check to catch faulty preperation"

This reverts commit c0e5ff6c628076b99d972778fef2dd33132a29a2.

* Change parallel jobs to stages, to reduce the load from Travis

* Check if just this test causes problems

* Fix typo in decorator of pytest

* Change object name
Delete some nodes after test, to be reusable in other
Undo scope back to module

* Add more Node cleanups

* Add cleanup if xml was imported

* Add more Node cleanups to avoid duplicated browsenames while import xml in tests

* Fix list issue in test

* Add even more necessary node cleanups

* Add more Node cleanups

* Remove faulty await

* Add Node cleanup for imported xml Nodes

* Try to get rid of imported nodes from xml

* Getting rid of imported nodes from xml

* Revert Travis
Some PEP8 cleanups

* Fix some changes back to master branch state

* Revert pytest changes

* Avoid multiple Nodes with same Browsename in test server

* Add some more Node cleanups

* Revert wrong changes

* Switch try/except with if ... not/else statement

* Remove empty parent check
Replace explicit in _nodes[] check with a general in _aspace[] check
parent 0261d3ed
Hide whitespace changes
Inline Side-by-side
Showing with 205 additions and 57 deletions
asyncua/common/manage_nodes.py
View file @ 6a913f89
......@@ -311,7 +311,6 @@ async def create_data_type(parent, nodeid, bname, description=None):
or namespace index, name
"""
nodeid, qname = _parse_nodeid_qname(nodeid, bname)
addnode = ua.AddNodesItem()
addnode.RequestedNewNodeId = nodeid
addnode.BrowseName = qname
......
asyncua/server/address_space.py
View file @ 6a913f89
......@@ -246,6 +246,19 @@ class NodeManagementService:
result.StatusCode = ua.StatusCode(ua.StatusCodes.BadParentNodeIdInvalid)
return result
if item.ParentNodeId in self._aspace:
for ref in self._aspace[item.ParentNodeId].references:
# Check if the Parent has a "HasChild" Reference (or subtype of it) with the Node
if ref.ReferenceTypeId.Identifier in [ua.ObjectIds.HasChild, ua.ObjectIds.HasComponent,
ua.ObjectIds.HasProperty, ua.ObjectIds.HasSubtype,
ua.ObjectIds.HasOrderedComponent] and ref.IsForward:
if item.BrowseName.Name == ref.BrowseName.Name:
self.logger.warning(f"AddNodesItem: Requested Browsename {item.BrowseName.Name}"
f" already exists in Parent Node. ParentID:{item.ParentNodeId} --- "
f"ItemId:{item.RequestedNewNodeId}")
result.StatusCode = ua.StatusCode(ua.StatusCodes.BadBrowseNameDuplicated)
return result
nodedata = NodeData(item.RequestedNewNodeId)
self._add_node_attributes(nodedata, item, add_timestamps=check)
......
tests/conftest.py
View file @ 6a913f89
......@@ -38,7 +38,7 @@ def event_loop(request):
loop.close()
@pytest.fixture(scope='function')
@pytest.fixture(scope='module')
async def running_server(request):
"""
Spawn a server in a separate thread
......
tests/custom_nodes.xml
View file @ 6a913f89
......@@ -121,7 +121,7 @@
</Definition>
</UADataType>
<UAVariable DataType="EnumValueType" ParentNodeId="i=120" ValueRank="1" NodeId="ns=1;i=99002" ArrayDimensions="3" BrowseName="EnumValues">
<UAVariable DataType="EnumValueType" ParentNodeId="i=120" ValueRank="1" NodeId="ns=1;i=99002" ArrayDimensions="3" BrowseName="EnumValues1">
<DisplayName>EnumValues</DisplayName>
<References>
<Reference ReferenceType="HasModellingRule">i=78</Reference>
......@@ -177,7 +177,7 @@
</UAVariable>
<!-- Test the other xml format-->
<UAVariable NodeId="i=99003" BrowseName="EnumValues" ParentNodeId="i=120" DataType="i=7594" ValueRank="1">
<UAVariable NodeId="i=99003" BrowseName="EnumValues2" ParentNodeId="i=120" DataType="i=7594" ValueRank="1">
<DisplayName>EnumValues</DisplayName>
<References>
<Reference ReferenceType="HasTypeDefinition">i=68</Reference>
......
tests/test_callback_service.py
View file @ 6a913f89
......@@ -83,5 +83,6 @@ async def test_write_callback(mocker):
await var.set_value(69.0)
assert mocked_write_items.called
await server.delete_nodes([myobj, myvar])
tests/test_client.py
View file @ 6a913f89
......@@ -31,6 +31,7 @@ async def test_folder_anonymous(server, admin_client, client):
assert f == f_ro
with pytest.raises(ua.UaStatusCodeError):
await f_ro.add_folder(3, 'MyFolder2')
await server.delete_nodes([f, f_ro])
async def test_variable_anonymous(server, admin_client, client):
......
tests/test_common.py
View file @ 6a913f89
......@@ -129,6 +129,7 @@ async def test_delete_nodes(opc):
await obj.get_child(["2:FolderToDelete", "2:VarToDelete"])
childs = await fold.get_children()
assert var not in childs
await opc.opc.delete_nodes([fold])
async def test_delete_nodes_with_inverse_references(opc):
......@@ -145,10 +146,12 @@ async def test_delete_nodes_with_inverse_references(opc):
await opc.opc.delete_nodes([var])
childs = await fold.get_children()
assert var not in childs
has_desc_refs = await var2.get_referenced_nodes(refs=ua.ObjectIds.HasDescription, direction=ua.BrowseDirection.Inverse)
has_desc_refs = await var2.get_referenced_nodes(refs=ua.ObjectIds.HasDescription,
direction=ua.BrowseDirection.Inverse)
assert len(has_desc_refs) == 0
has_effect_refs = await var2.get_referenced_nodes(refs=ua.ObjectIds.HasEffect, direction=ua.BrowseDirection.Inverse)
assert len(has_effect_refs) == 0
await opc.opc.delete_nodes([fold])
async def test_delete_nodes_recursive(opc):
......@@ -165,14 +168,13 @@ async def test_delete_nodes_recursive(opc):
async def test_delete_nodes_recursive2(opc):
obj = opc.opc.nodes.objects
fold = await obj.add_folder(2, "FolderToDeleteRoot")
nfold = fold
mynodes = []
for i in range(7):
nfold = await fold.add_folder(2, "FolderToDeleteRoot")
var = await fold.add_variable(2, "VarToDeleteR", 9.1)
var = await fold.add_property(2, "ProToDeleteR", 9.1)
prop = await fold.add_property(2, "ProToDeleteR", 9.1)
o = await fold.add_object(3, "ObjToDeleteR")
nfold = await fold.add_folder(2, f"FolderToDeleteRoot{i}")
var = await nfold.add_variable(2, "VarToDeleteR", 9.1)
var = await nfold.add_property(2, "ProToDeleteR", 9.1)
prop = await nfold.add_property(2, "ProToDeleteR2", 9.1)
o = await nfold.add_object(3, "ObjToDeleteR")
mynodes.append(nfold)
mynodes.append(var)
mynodes.append(prop)
......@@ -181,10 +183,12 @@ async def test_delete_nodes_recursive2(opc):
for node in mynodes:
with pytest.raises(ua.UaStatusCodeError):
await node.read_browse_name()
await opc.opc.delete_nodes([fold])
async def test_delete_references(opc):
newtype = await opc.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.opc.nodes.objects
fold = await obj.add_folder(2, "FolderToRef")
......@@ -285,6 +289,10 @@ async def test_browse(opc):
assert folder in all_objs
assert obj2 in all_objs
assert var not in all_objs
await opc.opc.delete_nodes([folder, prop, prop2, var, obj2, obj])
await opc.opc.delete_nodes(props)
await opc.opc.delete_nodes(all_vars)
await opc.opc.delete_nodes(all_objs)
async def test_browse_references(opc):
......@@ -323,6 +331,7 @@ async def test_browsename_with_spaces(opc):
v = await o.add_variable(3, 'BNVariable with spaces and %&+?/', 1.3)
v2 = await o.get_child("3:BNVariable with spaces and %&+?/")
assert v == v2
await opc.opc.delete_nodes([v])
async def test_non_existing_path(opc):
......@@ -359,12 +368,13 @@ async def test_datetime_write_value(opc):
v1 = await objects.add_variable(4, "test_datetime", now)
tid = await v1.read_value()
assert now == tid
await opc.opc.delete_nodes([v1])
async def test_variant_array_dim(opc):
objects = opc.opc.nodes.objects
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)
await v.write_array_dimensions([0, 0, 0])
......@@ -382,11 +392,12 @@ async def test_variant_array_dim(opc):
l = [[[], [], []], [[], [], []]]
variant = ua.Variant(l, ua.VariantType.UInt32)
v = await objects.add_variable(3, 'variableWithDimsEmpty', variant)
v2 = await v.read_value()
v1 = await objects.add_variable(3, 'variableWithDimsEmpty', variant)
v2 = await v1.read_value()
assert l == v2
dv = await v.read_data_value()
dv = await v1.read_data_value()
assert [2, 3, 0] == dv.Value.Dimensions
await opc.opc.delete_nodes([v, v1])
async def test_add_numeric_variable(opc):
......@@ -396,6 +407,7 @@ async def test_add_numeric_variable(opc):
qn = ua.QualifiedName('numericnodefromstring', 3)
assert nid == v.nodeid
assert qn == await v.read_browse_name()
await opc.opc.delete_nodes([v])
async def test_add_string_variable(opc):
......@@ -405,6 +417,7 @@ async def test_add_string_variable(opc):
qn = ua.QualifiedName('stringnodefromstring', 3)
assert nid == v.nodeid
assert qn == await v.read_browse_name()
await opc.opc.delete_nodes([v])
async def test_utf8(opc):
......@@ -419,6 +432,7 @@ async def test_utf8(opc):
assert val == val2
bn2 = await v.read_browse_name()
assert bn == bn2
await opc.opc.delete_nodes([v])
async def test_null_variable(opc):
......@@ -431,6 +445,7 @@ async def test_null_variable(opc):
val = await var.read_value()
assert val is not None
assert "" == val
await opc.opc.delete_nodes([var])
async def test_variable_data_type(opc):
......@@ -438,9 +453,11 @@ async def test_variable_data_type(opc):
var = await objects.add_variable(3, 'stringfordatatype', "a string")
val = await var.read_data_type_as_variant_type()
assert ua.VariantType.String == val
await opc.opc.delete_nodes([var])
var = await objects.add_variable(3, 'stringarrayfordatatype', ["a", "b"])
val = await var.read_data_type_as_variant_type()
assert ua.VariantType.String == val
await opc.opc.delete_nodes([var])
async def test_add_string_array_variable(opc):
......@@ -452,6 +469,7 @@ async def test_add_string_array_variable(opc):
assert qn == await v.read_browse_name()
val = await v.read_value()
assert ['l', 'b'] == val
await opc.opc.delete_nodes([v])
async def test_add_numeric_node(opc):
......@@ -461,6 +479,7 @@ async def test_add_numeric_node(opc):
v1 = await objects.add_variable(nid, qn, 0)
assert nid == v1.nodeid
assert qn == await v1.read_browse_name()
await opc.opc.delete_nodes([v1])
async def test_add_string_node(opc):
......@@ -470,6 +489,7 @@ async def test_add_string_node(opc):
v2 = await objects.add_variable(nid, qn, 0)
assert nid == v2.nodeid
assert qn == await v2.read_browse_name()
await opc.opc.delete_nodes([v2])
async def test_add_find_node_(opc):
......@@ -477,6 +497,7 @@ async def test_add_find_node_(opc):
o = await objects.add_object('ns=2;i=101;', '2:AddFindObject')
o2 = await objects.get_child('2:AddFindObject')
assert o == o2
await opc.opc.delete_nodes([o, o2])
async def test_node_path(opc):
......@@ -485,6 +506,7 @@ async def test_node_path(opc):
root = opc.opc.nodes.root
o2 = await root.get_child(['0:Objects', '2:NodePathObject'])
assert o == o2
await opc.opc.delete_nodes([o, o2])
async def test_add_read_node(opc):
......@@ -494,6 +516,7 @@ async def test_add_read_node(opc):
assert nid == o.nodeid
qn = ua.QualifiedName('AddReadObject', 2)
assert qn == await o.read_browse_name()
await opc.opc.delete_nodes([o])
async def test_simple_value(opc):
......@@ -501,13 +524,15 @@ async def test_simple_value(opc):
v = await o.add_variable(3, 'VariableTestValue', 4.32)
val = await v.read_value()
assert 4.32 == val
await opc.opc.delete_nodes([v])
async def test_add_exception(opc):
objects = opc.opc.nodes.objects
await objects.add_object('ns=2;i=103;', '2:AddReadObject')
v = await objects.add_object('ns=2;i=103;', '2:AddReadObject')
with pytest.raises(ua.UaStatusCodeError):
await objects.add_object('ns=2;i=103;', '2:AddReadObject')
await opc.opc.delete_nodes([v])
async def test_negative_value(opc):
......@@ -515,6 +540,7 @@ async def test_negative_value(opc):
v = await o.add_variable(3, 'VariableNegativeValue', 4)
await v.write_value(-4.54)
assert -4.54 == await v.read_value()
await opc.opc.delete_nodes([v])
async def test_read_server_state(opc):
......@@ -532,6 +558,7 @@ async def test_bad_node(opc):
await bad.add_object(0, "0:myobj")
with pytest.raises(ua.UaStatusCodeError):
await bad.get_child("0:myobj")
await opc.opc.delete_nodes([bad])
async def test_value(opc):
......@@ -544,6 +571,7 @@ async def test_value(opc):
assert ua.DataValue == type(dv)
assert dvar.Value == dv.Value
assert dvar.Value == var
await opc.opc.delete_nodes([v])
async def test_write_value(opc):
......@@ -560,12 +588,14 @@ async def test_write_value(opc):
await v.write_value(dvar)
v3 = await v.read_data_value()
assert v3.Value == dvar.Value
await opc.opc.delete_nodes([v])
async def test_array_value(opc):
o = opc.opc.nodes.objects
v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3])
assert [1, 2, 3] == await v.read_value()
await opc.opc.delete_nodes([v])
async def test_bool_variable(opc):
......@@ -578,13 +608,15 @@ async def test_bool_variable(opc):
await v.write_value(False)
val = await v.read_value()
assert val is False
await opc.opc.delete_nodes([v])
async def test_array_size_one_value(opc):
o = opc.opc.nodes.objects
v = await o.add_variable(3, 'VariableArrayValue', [1, 2, 3])
v = await o.add_variable(3, 'VariableArrayValue2', [1, 2, 3])
await v.write_value([1])
assert [1] == await v.read_value()
await opc.opc.delete_nodes([v])
async def test_use_namespace(opc):
......@@ -594,11 +626,12 @@ async def test_use_namespace(opc):
myvar = await root.add_variable(idx, 'var_in_custom_namespace', [5])
myid = myvar.nodeid
assert idx == myid.NamespaceIndex
await opc.opc.delete_nodes([myvar])
async def test_method(opc):
o = opc.opc.nodes.objects
await o.get_child("2:ServerMethod")
v = await o.get_child("2:ServerMethod")
result = await o.call_method("2:ServerMethod", 2.1)
assert 4.2 == result
with pytest.raises(ua.UaStatusCodeError):
......@@ -674,13 +707,14 @@ async def test_add_nodes(opc):
assert o in childs
assert v in childs
assert p in childs
await opc.opc.delete_nodes([f, o, v, p, child])
async def test_modelling_rules(opc):
obj = await opc.opc.nodes.base_object_type.add_object_type(2, 'MyFooObjectType')
v = await obj.add_variable(2, "myvar", 1.1)
await v.set_modelling_rule(True)
p = await obj.add_property(2, "myvar", 1.1)
p = await obj.add_property(2, "myvar2", 1.1)
await p.set_modelling_rule(False)
refs = await obj.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
......@@ -695,6 +729,7 @@ async def test_modelling_rules(opc):
await p.set_modelling_rule(None)
refs = await p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule)
assert 0 == len(refs)
await opc.opc.delete_nodes([v, p])
async def test_incl_subtypes(opc):
......@@ -716,7 +751,7 @@ async def test_add_node_with_type(opc):
assert ua.ObjectIds.BaseObjectType == (await o.get_type_definition()).Identifier
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, 'MyFooObjectType2')
o = await f.add_object(3, 'MyObject3', custom_otype.nodeid)
assert custom_otype.nodeid.Identifier == (await o.get_type_definition()).Identifier
......@@ -724,11 +759,12 @@ async def test_add_node_with_type(opc):
references = await o.get_references(refs=ua.ObjectIds.HasTypeDefinition, direction=ua.BrowseDirection.Forward)
assert 1 == len(references)
assert custom_otype.nodeid == references[0].NodeId
await opc.opc.delete_nodes([f, o])
async def test_references_for_added_nodes(opc):
objects = opc.opc.nodes.objects
o = await objects.add_object(3, 'MyObject')
o = await objects.add_object(3, 'MyObject4')
nodes = await objects.get_referenced_nodes(
refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False
)
......@@ -881,6 +917,7 @@ async def test_instantiate_1(opc):
var1 = await mydevicederived.get_child(["0:childparam"])
var_parent = await mydevicederived.get_child(["0:sensor"])
prop_parent = await mydevicederived.get_child(["0:sensor_id"])
await opc.opc.delete_nodes([devd_t, dev_t])
async def test_instantiate_string_nodeid(opc):
......@@ -897,7 +934,7 @@ async def test_instantiate_string_nodeid(opc):
# instanciate device
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]
assert ua.NodeClass.Object == await mydevice.read_node_class()
......@@ -909,6 +946,7 @@ async def test_instantiate_string_nodeid(opc):
assert ua.ObjectIds.PropertyType == (await prop.get_type_definition()).Identifier
assert "Running" == await prop.read_value()
assert prop.nodeid != prop_t.nodeid
await opc.opc.delete_nodes([dev_t])
async def test_variable_with_datatype(opc):
......@@ -923,6 +961,7 @@ async def test_variable_with_datatype(opc):
)
tp2 = await v2.read_data_type()
assert tp2 == ua.NodeId(ua.ObjectIds.ApplicationType)
await opc.opc.delete_nodes([v1, v2])
async def test_enum(opc):
......@@ -941,6 +980,8 @@ async def test_enum(opc):
# tests
assert myenum_type.nodeid == await myvar.read_data_type()
await myvar.write_value(ua.LocalizedText("String2"))
await opc.opc.delete_nodes([es, myvar])
await opc.opc.delete_nodes([myenum_type])
async def test_supertypes(opc):
......@@ -960,6 +1001,7 @@ async def test_supertypes(opc):
dtype2 = await dtype.add_data_type(0, "MyCustomDataType2")
node = await ua_utils.get_node_supertype(dtype2)
assert dtype == node
await opc.opc.delete_nodes([dtype, dtype2])
async def test_base_data_type(opc):
......@@ -974,6 +1016,7 @@ async def test_base_data_type(opc):
d = opc.opc.get_node(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)
await opc.opc.delete_nodes([ext])
async def test_data_type_to_variant_type(opc):
......@@ -992,7 +1035,8 @@ async def test_data_type_to_variant_type(opc):
ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32 # enumeration
}
for dt, vdt in test_data.items():
assert vdt == await ua_utils.data_type_to_variant_type(opc.opc.get_node(ua.NodeId(dt)))
k = await ua_utils.data_type_to_variant_type(opc.opc.get_node(ua.NodeId(dt)))
assert vdt == k
async def test_guid_node_id():
......@@ -1027,6 +1071,10 @@ async def test_import_xml_data_type_definition(opc):
s2 = await var.read_value()
assert s2.structs[1].toto == ss.structs[1].toto == 0.1
await opc.opc.delete_nodes([datatype, var])
n = []
[n.append(opc.opc.get_node(node)) for node in nodes]
await opc.opc.delete_nodes(n)
async def test_struct_data_type(opc):
......@@ -1034,7 +1082,7 @@ async def test_struct_data_type(opc):
node = opc.opc.get_node(ua.AddNodesItem.data_type)
path = await node.get_path()
assert opc.opc.nodes.base_structure_type in path
await opc.opc.delete_nodes([node])
async def test_import_xml_enum_data_type_definition(opc):
......@@ -1045,5 +1093,31 @@ async def test_import_xml_enum_data_type_definition(opc):
var = await opc.opc.nodes.objects.add_variable(2, "MyEnumVar", e, datatype=ua.enums_datatypes[ua.MyEnum])
e2 = await var.read_value()
assert e2 == ua.MyEnum.val2
await opc.opc.delete_nodes([var])
n = []
[n.append(opc.opc.get_node(node)) for node in nodes]
await opc.opc.delete_nodes(n)
async def test_duplicated_browsenames_same_ns(opc):
parentfolder = await opc.opc.nodes.objects.add_folder(2, "parent_folder")
childfolder = await parentfolder.add_folder(2, "child_folder")
try:
childfolder2 = await parentfolder.add_folder(2, "child_folder")
await opc.opc.delete_nodes([parentfolder])
pytest.fail("Childfolder2 should never be created!")
except:
await opc.opc.delete_nodes([parentfolder])
return
async def test_duplicated_browsenames_different_ns(opc):
parentfolder = await opc.opc.nodes.objects.add_folder(2, "parent_folder")
childfolder = await parentfolder.add_folder(2, "child_folder")
try:
childfolder2 = await parentfolder.add_folder(3, "child_folder")
await opc.opc.delete_nodes([parentfolder])
pytest.fail("Childfolder2 should never be created!")
except:
await opc.opc.delete_nodes([parentfolder])
return
tests/test_crypto_connect.py
View file @ 6a913f89
......@@ -167,6 +167,7 @@ async def test_basic256_encrypt_success(srv_crypto_all_certs):
assert await clt.nodes.objects.get_children()
@pytest.mark.skip("# FIXME: how to make it fail???")
async def test_basic256_encrypt_fail(srv_crypto_all_certs):
# FIXME: how to make it fail???
_, cert = srv_crypto_all_certs
......
tests/test_custom_structures.py
View file @ 6a913f89
......@@ -37,7 +37,7 @@ def set_up_test_tree():
return test_etree
@pytest.fixture(scope="module")
@pytest.fixture(scope="function")
async def _srv(server):
class Srv:
pass
......@@ -353,6 +353,7 @@ async def test_functional_basic(srv):
basic_result = await basic_var.read_value()
assert basic_result == basic_msg
await srv.srv.delete_nodes([basic_var])
async def test_functional_advance(srv):
......@@ -395,3 +396,4 @@ async def test_functional_advance(srv):
assert basic_result == basic_msg
nested_result = await nested_var.read_value()
assert nested_result == nested_msg
await srv.srv.delete_nodes([basic_var, nested_var])
tests/test_permissions.py
View file @ 6a913f89
......@@ -73,9 +73,9 @@ async def srv_crypto_one_cert(request):
await srv.start()
yield srv
# stop the server
await srv.delete_nodes([myobj, myvar])
await srv.stop()
async def test_permissions_admin(srv_crypto_one_cert):
clt = Client(uri_crypto_cert)
await clt.set_security(
......
tests/test_server.py
View file @ 6a913f89
......@@ -185,7 +185,7 @@ async def test_references_for_added_nodes_method(server):
includesubtypes=False)
assert o in nodes
assert await m.get_parent() == o
await server.delete_nodes([o])
async def test_get_event_from_type_node_BaseEvent(server):
"""
......@@ -295,7 +295,7 @@ async def test_eventgenerator_sourceMyObject(server):
evgen = await server.get_event_generator(emitting_node=o)
await check_eventgenerator_base_event(evgen, server)
await check_event_generator_object(evgen, o)
await server.delete_nodes([o])
async def test_eventgenerator_source_collision(server):
objects = server.nodes.objects
......@@ -304,7 +304,7 @@ async def test_eventgenerator_source_collision(server):
evgen = await server.get_event_generator(event, ua.ObjectIds.Server)
await check_eventgenerator_base_event(evgen, server)
await check_event_generator_object(evgen, o, emitting_node=asyncua.Node(server.iserver.isession, ua.ObjectIds.Server))
await server.delete_nodes([o])
async def test_eventgenerator_inherited_event(server):
evgen = await server.get_event_generator(ua.ObjectIds.AuditEventType)
......@@ -358,6 +358,7 @@ async def test_create_custom_event_type_object_id(server):
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
await check_custom_type(type, ua.ObjectIds.BaseEventType, server)
await server.delete_nodes([type])
async def test_create_custom_object_type_object_id(server):
......@@ -395,28 +396,31 @@ async def test_create_custom_event_type_node_id(server):
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
await check_custom_type(etype, ua.ObjectIds.BaseEventType, server)
await server.delete_nodes([etype])
async def test_create_custom_event_type_node(server):
etype = await server.create_custom_event_type(2, 'MyEvent', asyncua.Node(server.iserver.isession,
etype = await server.create_custom_event_type(2, 'MyEvent1', asyncua.Node(server.iserver.isession,
ua.NodeId(ua.ObjectIds.BaseEventType)),
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
await check_custom_type(etype, ua.ObjectIds.BaseEventType, server)
await server.delete_nodes([etype])
async def test_get_event_from_type_node_custom_event(server):
etype = await server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType,
etype = await server.create_custom_event_type(2, 'MyEvent2', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
ev = await asyncua.common.events.get_event_obj_from_type_node(etype)
check_custom_event(ev, etype)
assert 0 == ev.PropertyNum
assert ev.PropertyString is None
await server.delete_nodes([etype])
async def test_eventgenerator_custom_event(server):
etype = await server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType,
etype = await server.create_custom_event_type(2, 'MyEvent3', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
evgen = await server.get_event_generator(etype, ua.ObjectIds.Server)
......@@ -424,13 +428,14 @@ async def test_eventgenerator_custom_event(server):
await check_eventgenerator_source_server(evgen, server)
assert 0 == evgen.event.PropertyNum
assert evgen.event.PropertyString is None
await server.delete_nodes([etype])
async def test_eventgenerator_double_custom_event(server):
event1 = await server.create_custom_event_type(3, 'MyEvent1', ua.ObjectIds.BaseEventType,
event1 = await server.create_custom_event_type(3, 'MyEvent4', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
event2 = await server.create_custom_event_type(4, 'MyEvent2', event1, [('PropertyBool', ua.VariantType.Boolean),
event2 = await server.create_custom_event_type(4, 'MyEvent5', event1, [('PropertyBool', ua.VariantType.Boolean),
('PropertyInt', ua.VariantType.Int32)])
evgen = await server.get_event_generator(event2, ua.ObjectIds.Server)
check_eventgenerator_custom_event(evgen, event2, server)
......@@ -441,12 +446,13 @@ async def test_eventgenerator_double_custom_event(server):
# Properties from MyEvent2
assert not evgen.event.PropertyBool
assert 0 == evgen.event.PropertyInt
await server.delete_nodes([event1, event2])
async def test_eventgenerator_custom_event_my_object(server):
objects = server.nodes.objects
o = await objects.add_object(3, 'MyObject')
etype = await server.create_custom_event_type(2, 'MyEvent', ua.ObjectIds.BaseEventType,
etype = await server.create_custom_event_type(2, 'MyEvent6', ua.ObjectIds.BaseEventType,
[('PropertyNum', ua.VariantType.Int32),
('PropertyString', ua.VariantType.String)])
......@@ -455,6 +461,7 @@ async def test_eventgenerator_custom_event_my_object(server):
await check_event_generator_object(evgen, o)
assert 0 == evgen.event.PropertyNum
assert evgen.event.PropertyString is None
await server.delete_nodes([o, etype])
async def test_context_manager():
......
tests/test_subscriptions.py
View file @ 6a913f89
......@@ -114,6 +114,7 @@ async def test_subscription_overload(opc, handler_class):
await s.delete()
# assert myhandler.datachange_count == 1000
# assert myhandler.event_count == 0
await opc.opc.delete_nodes(variables)
@pytest.mark.parametrize("handler_class", [MySubHandlerCounter, MySubHandlerCounterAsync])
......@@ -130,6 +131,7 @@ async def test_subscription_count(opc, handler_class):
await sleep(0.2) # let last event arrive
assert nb + 1 == myhandler.datachange_count
await sub.delete()
await opc.opc.delete_nodes([var])
@pytest.mark.parametrize("handler_class", [MySubHandlerCounter, MySubHandlerCounterAsync])
......@@ -137,7 +139,7 @@ async def test_subscription_count_list(opc, handler_class):
myhandler = handler_class()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.nodes.objects
var = await o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
var = await o.add_variable(3, 'SubVarCounter1', [0.1, 0.2])
await sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
......@@ -149,6 +151,7 @@ async def test_subscription_count_list(opc, handler_class):
await sleep(0.2) # let last event arrive
assert nb + 1 == myhandler.datachange_count
await sub.delete()
await opc.opc.delete_nodes([var])
@pytest.mark.parametrize("handler_class", [MySubHandlerCounter, MySubHandlerCounterAsync])
......@@ -156,7 +159,7 @@ async def test_subscription_count_no_change(opc, handler_class):
myhandler = handler_class()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.nodes.objects
var = await o.add_variable(3, 'SubVarCounter', [0.1, 0.2])
var = await o.add_variable(3, 'SubVarCounter2', [0.1, 0.2])
await sub.subscribe_data_change(var)
nb = 12
for i in range(nb):
......@@ -165,6 +168,7 @@ async def test_subscription_count_no_change(opc, handler_class):
await sleep(0.2) # let last event arrive
assert 1 == myhandler.datachange_count
await sub.delete()
await opc.opc.delete_nodes([var])
@pytest.mark.parametrize("handler_class", [MySubHandlerCounter, MySubHandlerCounterAsync])
......@@ -172,7 +176,7 @@ async def test_subscription_count_empty(opc, handler_class):
myhandler = handler_class()
sub = await opc.opc.create_subscription(1, myhandler)
o = opc.opc.nodes.objects
var = await o.add_variable(3, 'SubVarCounter', [0.1, 0.2, 0.3])
var = await o.add_variable(3, 'SubVarCounter3', [0.1, 0.2, 0.3])
await sub.subscribe_data_change(var)
while True:
val = await var.read_value()
......@@ -185,6 +189,7 @@ async def test_subscription_count_empty(opc, handler_class):
await sleep(0.2) # let last event arrive
assert 4 == myhandler.datachange_count
await sub.delete()
await opc.opc.delete_nodes([var])
async def test_subscription_overload_simple(opc):
......@@ -232,6 +237,7 @@ async def test_subscription_data_change(opc):
await sub.delete()
with pytest.raises(ua.UaStatusCodeError):
await sub.unsubscribe(handle1) # sub does not exist anymore
await opc.opc.delete_nodes([v1])
@pytest.mark.parametrize("opc", ["client"], indirect=True)
......@@ -259,7 +265,7 @@ async def test_set_monitoring_mode(opc, mocker):
monitoring_mode = ua.SetMonitoringModeParameters()
mock_set_monitoring = mocker.patch.object(ua, "SetMonitoringModeParameters", return_value=monitoring_mode)
mock_client_monitoring = mocker.patch("asyncua.client.ua_client.UaClient.set_monitoring_mode", new=CoroutineMock())
v = await o.add_variable(3, 'SubscriptionVariable', 123)
v = await o.add_variable(3, 'SubscriptionVariable2', 123)
sub = await opc.opc.create_subscription(100, myhandler)
await sub.set_monitoring_mode(ua.MonitoringMode.Disabled)
......@@ -278,7 +284,7 @@ async def test_set_publishing_mode(opc, mocker):
publishing_mode = ua.SetPublishingModeParameters()
mock_set_monitoring = mocker.patch.object(ua, "SetPublishingModeParameters", return_value=publishing_mode)
mock_client_monitoring = mocker.patch("asyncua.client.ua_client.UaClient.set_publishing_mode", new=CoroutineMock())
v = await o.add_variable(3, 'SubscriptionVariable', 123)
v = await o.add_variable(3, 'SubscriptionVariable3', 123)
sub = await opc.opc.create_subscription(100, myhandler)
await sub.set_publishing_mode(False)
......@@ -312,6 +318,7 @@ async def test_subscription_data_change_bool(opc):
assert v1 == node
assert val is False
await sub.delete() # should delete our monitoreditem too
await opc.opc.delete_nodes([v1])
async def test_subscription_data_change_many(opc):
......@@ -347,6 +354,7 @@ async def test_subscription_data_change_many(opc):
else:
raise RuntimeError(f"Error node {node} is neither {v1} nor {v2}")
await sub.delete()
await opc.opc.delete_nodes([v1, v2])
async def test_subscribe_server_time(opc):
......@@ -387,12 +395,13 @@ async def test_modify_monitored_item(opc):
async def test_create_delete_subscription(opc):
o = opc.opc.nodes.objects
v = await o.add_variable(3, 'SubscriptionVariable', [1, 2, 3])
v = await o.add_variable(3, 'SubscriptionVariable4', [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)
await sub.delete()
await opc.opc.delete_nodes([v])
async def test_unsubscribe_two_objects_simultaneously(opc):
......@@ -450,7 +459,7 @@ async def test_subscribe_events_to_wrong_node(opc):
with pytest.raises(ua.UaStatusCodeError):
handle = await sub.subscribe_events(v)
await sub.delete()
await opc.opc.delete_nodes([v])
async def test_get_event_from_type_node_BaseEvent(opc):
etype = opc.opc.get_node(ua.ObjectIds.BaseEventType)
......@@ -473,6 +482,7 @@ async def test_get_event_from_type_node_CustomEvent(opc):
assert child in properties
assert await etype.get_child("2:PropertyNum") in properties
assert await etype.get_child("2:PropertyString") in properties
await opc.opc.delete_nodes([etype])
async def test_events_default(opc):
......@@ -515,6 +525,7 @@ async def test_events_MyObject(opc):
assert tid == ev.Time
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([o])
async def test_events_wrong_source(opc):
......@@ -531,6 +542,7 @@ async def test_events_wrong_source(opc):
ev = await myhandler.result()
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([o])
async def test_events_CustomEvent(opc):
......@@ -562,6 +574,7 @@ async def test_events_CustomEvent(opc):
assert propertystring == ev.PropertyString
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([etype])
async def test_events_CustomEvent_MyObject(opc):
......@@ -593,6 +606,7 @@ async def test_events_CustomEvent_MyObject(opc):
assert propertystring == ev.PropertyString
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([etype, o])
async def test_several_different_events(opc):
......@@ -636,7 +650,8 @@ async def test_several_different_events(opc):
assert 7 == len(ev1s)
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([etype1, etype2])
await opc.opc.delete_nodes([o])
async def test_several_different_events_2(opc):
objects = opc.server.nodes.objects
......@@ -691,7 +706,8 @@ async def test_several_different_events_2(opc):
assert ev1s[0].PropertyNum3 is None
await sub.unsubscribe(handle)
await sub.delete()
await opc.opc.delete_nodes([etype1, etype2, etype3])
await opc.opc.delete_nodes([o])
async def test_internal_server_subscription(opc):
"""
......@@ -715,3 +731,4 @@ async def test_internal_server_subscription(opc):
internal_sub = opc.server.iserver.subscription_service.subscriptions[sub.subscription_id]
# Check that the results are not left un-acknowledged on internal Server Subscriptions.
assert len(internal_sub._not_acknowledged_results) == 0
await opc.opc.delete_nodes([sub_obj])
tests/test_xml.py
View file @ 6a913f89
......@@ -27,7 +27,7 @@ def func(parent, value, string):
async def test_xml_import(opc):
await opc.opc.import_xml(CUSTOM_NODES_XML_PATH)
nodes = await opc.opc.import_xml(CUSTOM_NODES_XML_PATH)
o = opc.opc.nodes.objects
v = await o.get_child(["1:MyXMLFolder", "1:MyXMLObject", "1:MyXMLVariable"])
val = await v.read_value()
......@@ -44,6 +44,10 @@ async def test_xml_import(opc):
o = await opc.opc.nodes.root.get_child(node_path)
input_arg = (await o.read_data_value()).Value.Value[0]
assert "Context" == input_arg.Name
await opc.opc.delete_nodes([v])
n = []
[n.append(opc.opc.get_node(node)) for node in nodes]
await opc.opc.delete_nodes(n)
async def test_xml_import_additional_ns(opc):
......@@ -62,6 +66,7 @@ async def test_xml_import_additional_ns(opc):
assert ns == r1.NodeId.NamespaceIndex
r3 = (await v1.get_references(refs=ua.ObjectIds.HasComponent))[0]
assert ns == r3.NodeId.NamespaceIndex
await opc.opc.delete_nodes([o2, v1])
async def test_xml_method(opc, tmpdir):
......@@ -98,7 +103,7 @@ async def test_xml_vars(opc, tmpdir):
v = await o.add_variable(3, "myxmlvar", 6.78, ua.VariantType.Double)
a = await o.add_variable(3, "myxmlvar-array", [6, 1], ua.VariantType.UInt16)
a2 = await o.add_variable(3, "myxmlvar-2dim", [[1, 2], [3, 4]], ua.VariantType.UInt32)
a3 = await o.add_variable(3, "myxmlvar-2dim", [[]], ua.VariantType.ByteString)
a3 = await o.add_variable(3, "myxmlvar-2dim2", [[]], ua.VariantType.ByteString)
nodes = [o, v, a, a2, a3]
await opc.opc.export_xml(nodes, tmp_path)
await opc.opc.delete_nodes(nodes)
......@@ -178,16 +183,19 @@ async def test_xml_float(opc, tmpdir):
assert o2 == o
assert await o2.read_data_type() == dtype
assert (await o2.read_data_value()).Value == dv.Value
await opc.opc.delete_nodes([o, o2])
async def test_xml_bool(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlbool", True)
await _test_xml_var_type(opc, tmpdir, o, "bool")
await opc.opc.delete_nodes([o])
async def test_xml_string(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlstring", "mystring")
await _test_xml_var_type(opc, tmpdir, o, "string")
await opc.opc.delete_nodes([o])
async def test_xml_string_with_null_description(opc, tmpdir):
......@@ -195,27 +203,32 @@ async def test_xml_string_with_null_description(opc, tmpdir):
await o.write_attribute(ua.AttributeIds.Description, ua.DataValue(None))
o2 = await _test_xml_var_type(opc, tmpdir, o, "string")
assert await o.read_description() == await o2.read_description()
await opc.opc.delete_nodes([o, o2])
async def test_xml_string_array(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlstringarray", ["mystring2", "mystring3"])
node2 = await _test_xml_var_type(opc, tmpdir, o, "stringarray")
dv = await node2.read_data_value()
await opc.opc.delete_nodes([o, node2])
async def test_xml_guid(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlguid", uuid.uuid4())
await _test_xml_var_type(opc, tmpdir, o, "guid")
await opc.opc.delete_nodes([o])
async def test_xml_guid_array(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlguid", [uuid.uuid4(), uuid.uuid4()])
await _test_xml_var_type(opc, tmpdir, o, "guid_array")
await opc.opc.delete_nodes([o])
async def test_xml_datetime(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(3, "myxmlvar-dt", datetime.datetime.utcnow(), ua.VariantType.DateTime)
await _test_xml_var_type(opc, tmpdir, o, "datetime")
await opc.opc.delete_nodes([o])
async def test_xml_datetime_array(opc, tmpdir):
......@@ -225,6 +238,7 @@ async def test_xml_datetime_array(opc, tmpdir):
datetime.datetime.now(pytz.timezone("Asia/Tokyo"))
], ua.VariantType.DateTime)
await _test_xml_var_type(opc, tmpdir, o, "datetime_array")
await opc.opc.delete_nodes([o])
# async def test_xml_qualifiedname(opc):
......@@ -238,36 +252,48 @@ async def test_xml_datetime_array(opc, tmpdir):
async def test_xml_bytestring(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext", "mytext".encode("utf8"), ua.VariantType.ByteString)
await _test_xml_var_type(opc, tmpdir, o, "bytestring")
await opc.opc.delete_nodes([o])
async def test_xml_bytestring_array(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext_array",
["mytext".encode("utf8"), "errsadf".encode("utf8")], ua.VariantType.ByteString)
["mytext".encode("utf8"), "errsadf".encode("utf8")],
ua.VariantType.ByteString)
await _test_xml_var_type(opc, tmpdir, o, "bytestring_array")
await opc.opc.delete_nodes([o])
async def test_xml_localizedtext(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext", ua.LocalizedText("mytext"))
await _test_xml_var_type(opc, tmpdir, o, "localized_text")
await opc.opc.delete_nodes([o])
async def test_xml_localizedtext_with_locale(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext", ua.LocalizedText("mytext","en-US"))
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext", ua.LocalizedText("mytext", "en-US"))
await _test_xml_var_type(opc, tmpdir, o, "localized_text")
await opc.opc.delete_nodes([o])
async def test_xml_localizedtext_array(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext_array",
[ua.LocalizedText("erert"), ua.LocalizedText("erert33")])
[ua.LocalizedText("erert"), ua.LocalizedText("erert33")])
await _test_xml_var_type(opc, tmpdir, o, "localized_text_array")
await opc.opc.delete_nodes([o])
async def test_xml_localizedtext_array_with_locale(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlltext_array",
[ua.LocalizedText(text="erert",locale="en"), ua.LocalizedText(text="erert33",locale="de")])
[ua.LocalizedText(text="erert", locale="en"),
ua.LocalizedText(text="erert33", locale="de")])
await _test_xml_var_type(opc, tmpdir, o, "localized_text_array")
await opc.opc.delete_nodes([o])
async def test_xml_nodeid(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlnodeid", ua.NodeId("mytext", 1))
await _test_xml_var_type(opc, tmpdir, o, "nodeid")
await opc.opc.delete_nodes([o])
async def test_xml_ext_obj(opc, tmpdir):
......@@ -283,6 +309,7 @@ async def test_xml_ext_obj(opc, tmpdir):
assert arg.ArrayDimensions == arg2.ArrayDimensions
assert arg.Description == arg2.Description
assert arg.DataType == arg2.DataType
await opc.opc.delete_nodes([node, node2])
async def test_xml_ext_obj_array(opc, tmpdir):
......@@ -305,26 +332,30 @@ async def test_xml_ext_obj_array(opc, tmpdir):
assert args[i].ArrayDimensions == read_args[i].ArrayDimensions
assert args[i].Description == read_args[i].Description
assert args[i].DataType == read_args[i].DataType
await opc.opc.delete_nodes([node, node2])
async def test_xml_enum(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlenum", 0, varianttype=ua.VariantType.Int32,
datatype=ua.ObjectIds.ApplicationType)
datatype=ua.ObjectIds.ApplicationType)
await _test_xml_var_type(opc, tmpdir, o, "enum")
await opc.opc.delete_nodes([o])
async def test_xml_enumvalues(opc, tmpdir):
o = await opc.opc.nodes.objects.add_variable(2, "xmlenumvalues", 0, varianttype=ua.VariantType.UInt32,
datatype=ua.ObjectIds.AttributeWriteMask)
datatype=ua.ObjectIds.AttributeWriteMask)
await _test_xml_var_type(opc, tmpdir, o, "enumvalues")
await opc.opc.delete_nodes([o])
async def test_xml_custom_uint32(opc, tmpdir):
# t = opc.opc.nodes. create_custom_data_type(2, 'MyCustomUint32', ua.ObjectIds.UInt32)
t = await opc.opc.get_node(ua.ObjectIds.UInt32).add_data_type(2, 'MyCustomUint32')
o = await opc.opc.nodes.objects.add_variable(2, "xmlcustomunit32", 0, varianttype=ua.VariantType.UInt32,
datatype=t.nodeid)
datatype=t.nodeid)
await _test_xml_var_type(opc, tmpdir, o, "cuint32")
await opc.opc.delete_nodes([o, t])
async def test_xml_var_nillable(opc):
......@@ -369,6 +400,7 @@ async def test_xml_var_nillable(opc):
var_bool = opc.opc.get_node(ua.NodeId('test_xml.bool.nillabel', 2))
assert await var_string.read_value() is None
assert await var_bool.read_value() is None
await opc.opc.delete_nodes([var_string, var_bool])
async def _test_xml_var_type(opc, tmpdir, node: Node, typename: str, test_equality: bool = True):
......@@ -385,7 +417,7 @@ async def _test_xml_var_type(opc, tmpdir, node: Node, typename: str, test_equali
assert node == node
assert dtype == await node2.read_data_type()
if test_equality:
print("DEBUG", node, dv, node2, await node2.read_value())
logger.debug(node, dv, node2, await node2.read_value())
assert dv.Value == (await node2.read_data_value()).Value
assert rank == await node2.read_value_rank()
assert dim == await node2.read_array_dimensions()
......@@ -405,3 +437,4 @@ async def test_xml_byte(opc, tmpdir):
assert o == o2
assert dtype == await o2.read_data_type()
assert dv.Value == (await o2.read_data_value()).Value
await opc.opc.delete_nodes([o2])
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