Commit 95c88d5e authored by Ivan Tyagov's avatar Ivan Tyagov

Add e2e test for testing username / password authentication against a coupler.

parent 2c95b72b
import time import time
import string
import random
from opcua import Client from opcua import Client
import slapos.testing.e2e as e2e import slapos.testing.e2e as e2e
# the OPC UA node of relay0 at I2C0 slave
OPC_UA_IDENTIFIER = "ns=1;s=i2c0.relay0"
class BeremizTest(e2e.EndToEndTestCase): class BeremizTest(e2e.EndToEndTestCase):
""" """
This tests check proper functioning with real instances (inside SlapOS cloud) This tests check proper functioning with real instances (inside SlapOS cloud)
...@@ -34,8 +39,8 @@ class BeremizTest(e2e.EndToEndTestCase): ...@@ -34,8 +39,8 @@ class BeremizTest(e2e.EndToEndTestCase):
time.sleep(5*3600) time.sleep(5*3600)
print("Request beremiz-runtime and OSIE coupler instances.") print("Request beremiz-runtime and OSIE coupler instances.")
# supply / request coupler # supply / request coupler in anonymous mode (no user / pass)
instance_name = time.strftime('e2e-test-coupler-%Y-%B-%d-%H:%M:%S') instance_name = time.strftime('e2e-test-coupler-anonymous-%Y-%B-%d-%H:%M:%S')
parameter_dict = {"mode":1} parameter_dict = {"mode":1}
cls.request(cls.coupler_release, cls.request(cls.coupler_release,
instance_name, instance_name,
...@@ -45,6 +50,22 @@ class BeremizTest(e2e.EndToEndTestCase): ...@@ -45,6 +50,22 @@ class BeremizTest(e2e.EndToEndTestCase):
print(connection_dict) print(connection_dict)
cls.coupler_url_ipv6 = connection_dict.get("url-ipv6") cls.coupler_url_ipv6 = connection_dict.get("url-ipv6")
# supply / request coupler in authenticated mode (no user / pass)
cls.username = "admin"
cls.password = ''.join(random.choices(string.ascii_uppercase + string.digits, k=8))
instance_name = time.strftime('e2e-test-coupler-authenticated-%Y-%B-%d-%H:%M:%S')
parameter_dict = {"mode":1,
"username": cls.username,
"password": cls.password,
"opc_ua_port": 5840}
cls.request(cls.coupler_release,
instance_name,
partition_parameter_kw=parameter_dict)
cls.waitUntilGreen(instance_name, 180)
connection_dict = cls.getInstanceInfos(instance_name).connection_dict
print(connection_dict)
cls.authenticated_coupler_url_ipv6 = connection_dict.get("url-ipv6")
# supply / request beremiz-runtime # supply / request beremiz-runtime
instance_name = time.strftime('e2e-test-beremiz-runtime-%Y-%B-%d-%H:%M:%S') instance_name = time.strftime('e2e-test-beremiz-runtime-%Y-%B-%d-%H:%M:%S')
parameter_dict = {"runtime_plc_url": "https://lab.nexedi.com/nexedi/osie/raw/master/Beremiz/beremiz_test_opc_ua/bin/beremiz_test_opc_ua.tgz"} parameter_dict = {"runtime_plc_url": "https://lab.nexedi.com/nexedi/osie/raw/master/Beremiz/beremiz_test_opc_ua/bin/beremiz_test_opc_ua.tgz"}
...@@ -55,10 +76,12 @@ class BeremizTest(e2e.EndToEndTestCase): ...@@ -55,10 +76,12 @@ class BeremizTest(e2e.EndToEndTestCase):
connection_dict = cls.getInstanceInfos(instance_name).connection_dict connection_dict = cls.getInstanceInfos(instance_name).connection_dict
print(connection_dict) print(connection_dict)
def test_plc_increment_run(self): def test_01_plc_increment_run(self):
"""
Test that beremiz-runtime can control over OPC UA a slave coupler by incrementing a counter.
"""
NUMBER_OF_CHECKS = 100 NUMBER_OF_CHECKS = 100
TIMEOUT = 3 TIMEOUT = 3
OPC_UA_IDENTIFIER = "ns=1;s=i2c0.relay0"
# give it some time for services(runtime & coupler) to warm up, connect and run # give it some time for services(runtime & coupler) to warm up, connect and run
time.sleep(30) time.sleep(30)
...@@ -90,3 +113,22 @@ class BeremizTest(e2e.EndToEndTestCase): ...@@ -90,3 +113,22 @@ class BeremizTest(e2e.EndToEndTestCase):
client.disconnect() client.disconnect()
# no failures # no failures
self.assertEqual(test_failures, 0) self.assertEqual(test_failures, 0)
def test_02_username_password_authentication_coupler(self):
"""
Test username / password authentication for coupler.
"""
try:
# connect to a session at OPC-UA server
client = Client(self.authenticated_coupler_url_ipv6)
client.set_user(self.username)
client.set_password(self.password)
client.connect()
root = client.get_root_node()
children_list = root.get_children()
var = client.get_node(OPC_UA_IDENTIFIER)
# by default it is a 0
self.assertEqual(var.get_value(), 0)
print(var.get_value())
finally:
client.disconnect()
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