demos/opc_ua/scripts/test_opcua.py

#
# Copyright 2019,2020 NXP
# SPDX-License-Identifier: Apache-2.0
#
#

import subprocess
import sys
import logging
import time

logging.basicConfig(format='%(message)s', level=logging.DEBUG)
log = logging.getLogger(__name__)

try:
    from subprocess import CREATE_NEW_CONSOLE
except:
    log.info("Test script is supported only on windows")
    sys.exit(1)


def doTest(arguments):
    connection_type = arguments[1]
    connection_param = arguments[2]
    root_ca = arguments[3]


    logFile = open('opcua_results.txt', 'wb')
    opcua_with_rootca = "NOT TESTED"
    opcua_without_rootca = "NOT TESTED"

    log.info("******* CREATE CREDENTIALS *******" )
    create_cert = "python3 createOPCUACredentials_Optional.py "
    log.info("%s" % create_cert)
    p = subprocess.Popen(create_cert, shell=True, stdout=subprocess.PIPE)
    stdout, stderr = p.communicate()

    time.sleep(3)
    log.info("\n")
    log.info("******* OPCUA PROVISION *******" )
    prov_cmd = "python3 provisionOPCUAServer.py " + connection_param + " " + connection_type
    log.info("%s" % prov_cmd)
    p = subprocess.Popen(prov_cmd, shell=True, stdout=subprocess.PIPE)
    stdout, stderr = p.communicate()


    # TEST OPC UA WITH ROOT CA
    time.sleep(3)
    log.info("\n")
    srv_cmd = "python3 open62541Server.py " + connection_type + " " + connection_param + " " + root_ca
    log.info("%s" % srv_cmd)
    ser = subprocess.Popen(("python3 open62541Server.py %s %s %s" % (connection_type, connection_param, root_ca)), creationflags=CREATE_NEW_CONSOLE)

    log.info("Starting Client")
    time.sleep(3)
    log.info("\n")
    output = subprocess.Popen(("python3 open62541Client.py opc.tcp://127.0.0.1:4840"),stdout=subprocess.PIPE).stdout
    time.sleep(3)
    log.info("\n")

    substr_found = 0
    for line in output:
        log.info("%s", line)
        str_res = line.find(str.encode("Connected!"))
        if str_res != -1:
            substr_found = 1

    if substr_found == 1:
        result_msg = "TEST(OPCUA, open62541_with_root_ca) PASS"
        log.info(result_msg)
        opcua_with_rootca = "PASS"
    else:
        result_msg = "TEST(OPCUA, open62541_with_root_ca) FAIL"
        log.info(result_msg)
        opcua_with_rootca = "FAIL"

    logFile.write(str.encode(result_msg))
    logFile.write(str.encode("\n"))

    output.close()
    log.info("Stop server ...")
    log.info("\n\n")
    temp = subprocess.Popen("TASKKILL /F /PID {pid} /T".format(pid=ser.pid), stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            shell=True,
                            )
    #std_out, std_err = temp.communicate()
    #std_out = std_out.strip()
    #std_err = std_err.strip()
    #log.info("INFO:  std_out: %s" % bytes.decode(std_out))
    #log.error("ERROR: std_err: %s" % bytes.decode(std_err))
    time.sleep(3)
    log.info("\n")



    # TEST OPC UA WITHOUT ROOT CA
    time.sleep(3)
    log.info("\n")
    srv_cmd = "python3 open62541Server.py " + connection_type + " " + connection_param
    log.info("%s" % srv_cmd)
    ser = subprocess.Popen(("python3 open62541Server.py %s %s %s" % (connection_type, connection_param, root_ca)), creationflags=CREATE_NEW_CONSOLE)

    log.info("Starting Client")
    time.sleep(3)
    log.info("\n")
    output = subprocess.Popen(("python3 open62541Client.py opc.tcp://127.0.0.1:4840"),stdout=subprocess.PIPE).stdout
    time.sleep(3)
    log.info("\n")

    substr_found = 0
    for line in output:
        log.info("%s", line)
        str_res = line.find(str.encode("Connected!"))
        if str_res != -1:
            substr_found = 1

    if substr_found == 1:
        result_msg = "TEST(OPCUA, open62541_without_root_ca) PASS"
        log.info(result_msg)
        opcua_without_rootca = "PASS"
    else:
        result_msg = "TEST(OPCUA, open62541_without_root_ca) FAIL"
        log.info(result_msg)
        opcua_without_rootca = "FAIL"

    logFile.write(str.encode(result_msg))
    logFile.write(str.encode("\n"))

    output.close()
    log.info("Stop server ...")
    log.info("\n\n")
    temp = subprocess.Popen("TASKKILL /F /PID {pid} /T".format(pid=ser.pid), stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            shell=True,
                            )
    #std_out, std_err = temp.communicate()
    #std_out = std_out.strip()
    #std_err = std_err.strip()
    #log.info("INFO:  std_out: %s" % bytes.decode(std_out))
    #log.error("ERROR: std_err: %s" % bytes.decode(std_err))


    logFile.close()


    log.info("*************** RESULT ****************")
    log.info("OPC UA WITH ROOT CA ------- %s" %opcua_with_rootca)
    log.info("OPC UA WITHOUT ROOT CA ---- %s" %opcua_without_rootca)
    log.info("***************************************")
    return 0


def printUsage():
    print('Invalid input argument')
    print('Run as -  test_opcua.py <jrcpv2|vcom> <ip_address|port_name> <root_ca>')
    print('Example invocation - test_opcua.py jrcpv2 127.0.0.1:8050 ..\\credentials\\open62541_client_cert.der')
    sys.exit()


if len(sys.argv) != 4:
    printUsage()
else:
    if doTest(sys.argv) != 0:
        printUsage()