tests/ide_tests/opcua_browse_encrypted.sikuli/opcua_service.bash
changeset 3820 46f3ca3f0157
parent 3718 7841b651d601
--- a/tests/ide_tests/opcua_browse_encrypted.sikuli/opcua_service.bash	Wed May 31 23:16:29 2023 +0200
+++ b/tests/ide_tests/opcua_browse_encrypted.sikuli/opcua_service.bash	Sun Jun 18 16:28:42 2023 +0200
@@ -1,13 +1,17 @@
 #!/bin/bash
 
+set -x -e
+
 echo "Instant encrypted OPC-UA server for test"
 
+rm -f my_cert.pem my_cert.der my_private_key.pem
+
 yes "" | openssl req -x509 -newkey rsa:2048 -keyout my_private_key.pem -out my_cert.pem \
         -days 355 -nodes -addext "subjectAltName = URI:urn:example.org:FreeOpcUa:python-opcua"
 openssl x509 -outform der -in my_cert.pem -out my_cert.der
 
 PROJECT_FILES_DIR=$BEREMIZPATH/tests/projects/opcua_browse_encrypted/project_files
-mkdir $PROJECT_FILES_DIR
+mkdir $PROJECT_FILES_DIR -p
 cp my_cert.der my_private_key.pem $PROJECT_FILES_DIR
 
 echo "CERTS READY"
@@ -18,37 +22,51 @@
 import sys
 import os
 import time
+import asyncio
 
-from opcua import ua, Server
+from asyncua import ua, Server
+from asyncua.server.users import User, UserRole
 
-server = Server()
-host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1")
-endpoint = "opc.tcp://"+host+":4840/freeopcua/server/"
-server.set_endpoint(endpoint)
+# Asyncua can't work without (over)simple shared cerificates/privkey.
+# No user is involved in that case, but asyncua needs it.
+# Over permessive User Manager hereafter helps cuting that corner.
+class AllAdminUserManager:
+    def get_user(self, iserver, username=None, password=None, certificate=None):
+        return User(role=UserRole.Admin)
 
-server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
-server.load_certificate("my_cert.der")
-server.load_private_key("my_private_key.pem")
+async def main():
+    server = Server(user_manager=AllAdminUserManager())
+    host = os.environ.get("OPCUA_DEFAULT_HOST", "127.0.0.1")
+    endpoint = "opc.tcp://"+host+":4840/freeopcua/server/"
+    await server.init()
+    server.set_endpoint(endpoint)
 
-uri = "http://beremiz.github.io"
-idx = server.register_namespace(uri)
+    server.set_security_policy([ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt])
+    await server.load_certificate("my_cert.der")
+    await server.load_private_key("my_private_key.pem")
 
-objects = server.get_objects_node()
+    uri = "http://beremiz.github.io"
+    idx = await server.register_namespace(uri)
 
-testobj = objects.add_object(idx, "TestObject")
-testvarout = testobj.add_variable(idx, "TestOut", 1.2)
-testvar = testobj.add_variable(idx, "TestIn", 5.6)
-testvar.set_writable()
+    objects = server.get_objects_node()
 
-server.start()
+    testobj = await objects.add_object(idx, "TestObject")
+    testvarout = await testobj.add_variable(idx, "TestOut", 1.2)
+    testvar = await testobj.add_variable(idx, "TestIn", 5.6)
+    await testvar.set_writable()
 
-try:
-    while True:
-        time.sleep(1)
-        inval=testvar.get_value()
-        print inval
-        testvarout.set_value(inval*2)
-        sys.stdout.flush()
-finally:
-    server.stop()
+    await server.start()
+
+    try:
+        while True:
+            await asyncio.sleep(1)
+            inval = await testvar.get_value()
+            print(inval)
+            await testvarout.set_value(inval*2)
+            sys.stdout.flush()
+    finally:
+        await server.stop()
+
+asyncio.run(main())
+
 EOF