add setup.cfg, small linting

asyncua/common/xmlimporter.py

@@ -4,7 +4,7 @@
 """
 import logging
 import uuid
-from typing import Coroutine, Union, Dict
+from typing import Union, Dict
 from copy import copy
 
 from asyncua import ua
@@ -94,7 +94,7 @@ def _add_missing_parents(self, dnodes):
                 from IPython import embed
                 embed()
 
-    async def _add_node_data(self, nodedata) -> "Node":
+    async def _add_node_data(self, nodedata) -> ua.NodeId:
         if nodedata.nodetype == "UAObject":
             node = await self.add_object(nodedata)
         elif nodedata.nodetype == "UAObjectType":

examples/test_perf.py

@@ -2,7 +2,6 @@
 import time
 import sys
 import logging
-import uvloop
 import cProfile
 
 sys.path.insert(0, "..")
@@ -30,16 +29,16 @@ async def mymain():
     async with server:
         while True:
             await asyncio.sleep(10)
-        #nb = 100000
-        #start = time.time()
-        #for i in range(nb):
-            #await server.write_attribute_value(myvar.nodeid, ua.DataValue(i))
-            #await myvar.write_value(i)
+        nb = 100000
+        start = time.time()
+        for i in range(nb):
+            await server.write_attribute_value(myvar.nodeid, ua.DataValue(i))
+            await myvar.write_value(i)
     print("\n Write frequency: \n", nb / (time.time() - start))
 
 
 if __name__ == "__main__":
     #uvloop.install()
     logging.basicConfig(level=logging.WARNING)
-    #cProfile.run('asyncio.run(mymain(), debug=True)', filename="perf.cprof")
-    asyncio.run(mymain())
+    cProfile.run('asyncio.run(mymain(), debug=True)', filename="perf.cprof")
+    #asyncio.run(mymain())

setup.cfg

@@ -1,5 +1,8 @@
-[aliases]
-test=pytest
 [flake8]
-max-line-length = 120
-exclude = opcua/ua/,opcua/common/event_objects.py,opcua/server/standard_address_space/standard_address_space*.py
+max-line-length = 160
+exclude = tests,schemas
+ignore = E501
+per-file-ignores =
+    __init__.py: F401
+[pycodestyle]
+max-line-length = 160

test_external_server.py

@@ -2,20 +2,19 @@
 Test an OPC-UA server with freeopcua python client
 """
 import sys
-import pytest
 import asyncio
 import logging
 from datetime import datetime
 
 from asyncua import ua
 from asyncua import Client
+import unittest
 
 
 class MySubHandler:
     """
     More advanced subscription client using Future, so we can wait for events in tests
     """
-
     def __init__(self):
         self.future = asyncio.Future()
 
@@ -31,7 +30,7 @@ def event_notification(self, event):
 
 class MySubHandler2(object):
     def __init__(self):
-        self.results = [] 
+        self.results = []
 
     def datachange_notification(self, node, val, data):
         self.results.append((node, val))
@@ -48,60 +47,69 @@ def wrapper(self):
             func(self, client)
         finally:
             client.disconnect()
-    return wrapper
 
+    return wrapper
 
 
 def test_connect_anonymous(self):
     c = Client(URL)
     c.connect()
     c.disconnect()
 
+
 def FINISH_test_connect_basic256(self):
     c = Client(URL)
     c.set_security_string("basic256,sign,XXXX")
     c.connect()
     c.disconnect()
 
+
 def test_find_servers(self):
     c = Client(URL)
     res = c.connect_and_find_servers()
     assert len(res) > 0
 
+
 def test_find_endpoints(self):
     c = Client(URL)
     res = c.connect_and_get_server_endpoints()
     assert len(res) > 0
 
+
 # @connect
 def test_get_root(self, client):
     root = client.nodes.root
     self.assertEqual(root.read_browse_name(), ua.QualifiedName("Root", 0))
 
+
 # @connect
 def test_get_root_children(self, client):
     root = client.nodes.root
     childs = root.get_children()
     assert len(childs) > 2
 
+
 # @connect
 async def test_get_namespace_array(self, client):
     array = await client.get_namespace_array()
     assert len(array) > 0
 
+
 # @connect
 def test_get_server_node(self, client):
     srv = client.nodes.server
     self.assertEqual(srv.read_browse_name(), ua.QualifiedName("Server", 0))
     #childs = srv.get_children()
     #assert len(childs) > 4)
 
+
 # @connect
 def test_browsepathtonodeid(self, client):
     root = client.nodes.root
-    node = root.get_child(["0:Objects", "0:Server" , "0:ServerArray"])
+    node = root.get_child(["0:Objects", "0:Server", "0:ServerArray"])
     self.assertEqual(node.read_browse_name(), ua.QualifiedName("ServerArray", 0))
 
+
 # @connect
 def test_subscribe_server_time(self, client):
     msclt = MySubHandler()
@@ -132,4 +140,3 @@ def test_subscribe_server_time(self, client):
         URL = sys.argv[1]
 
     unittest.main(verbosity=30, argv=sys.argv[:1])
-