diff --git a/test/common/unit_test_tcp_client.py b/test/common/unit_test_tcp_client.py new file mode 100644 index 000000000..fd33fb91e --- /dev/null +++ b/test/common/unit_test_tcp_client.py @@ -0,0 +1,5 @@ +def getClientManager(erpc): + transport = erpc.transport.TCPTransport("localhost", 12345, False) + clientManager = erpc.client.ClientManager( + transport, erpc.basic_codec.BasicCodec) + return clientManager diff --git a/test/common/unit_test_tcp_server.py b/test/common/unit_test_tcp_server.py new file mode 100644 index 000000000..07ed711f6 --- /dev/null +++ b/test/common/unit_test_tcp_server.py @@ -0,0 +1,13 @@ +def getServer(erpc, commonInterface): + class CommonServer(commonInterface): + + def quit(self): + exit() + + def getServerAllocated(self): + return 0 + + transport = erpc.transport.TCPTransport("localhost", 12345, True) + server = erpc.simple_server.SimpleServer( + transport, erpc.basic_codec.BasicCodec) + return server,CommonServer() diff --git a/test/conftest.py b/test/conftest.py new file mode 100644 index 000000000..4444def56 --- /dev/null +++ b/test/conftest.py @@ -0,0 +1,125 @@ +#! /usr/bin/python + +# Copyright (c) 2022 NXP +# Copyright (c) 2022 ACRIOS Systems s.r.o. +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause + +import pytest +import py +import subprocess +import os + +PyLang = "py" +CLang = "c" + +TcpTransport = "tcp" + +Langs = [CLang, PyLang] +Transports = [TcpTransport] + + +class TestException(Exception): + pass + + +class TestFile(pytest.File): + def collect(self): + for langClient in Langs: + for langServer in Langs: + for transport in Transports: + case = TestCase(self.fspath.purebasename, + langClient, langServer, transport, self.fspath) + yield TestItem.from_parent(self, name=case.desc, case=case) + + +class TestCase(object): + def __init__(self, name: str, langClient: str, langServer: str, transport: str, caseDir: str): + self._name = name + self._casename = f"{name}_{langClient}_{langServer}_{transport}" + self._langClient = langClient + self._langServer = langServer + self._transport = transport + self._caseDir = caseDir + + @property + def desc(self): + return f"{self._name}_{self._langClient}_{self._langServer}_{self._transport}" + + def run(self): + timeToWaitForTest = 20 + serverProcess = subprocess.Popen(self._getServerCmd(), shell=True) + clientProcess = subprocess.Popen(self._getClientCmd(), shell=True) + serverProcess.wait(timeToWaitForTest) + isServerKilled = False + isClientKilled = False + if serverProcess.poll() == None: + serverProcess.kill() + isServerKilled = True + if clientProcess.poll() == None: + clientProcess.kill() + isClientKilled = True + outsServer, errsServer = serverProcess.communicate() + outsClient, errsClient = clientProcess.communicate() + if isServerKilled: + raise TestException( + f"Server was killed as app didn't finished in {timeToWaitForTest}s: {outsServer} {errsServer}") + if isClientKilled: + raise TestException( + f"Client was killed as app didn't finished in {timeToWaitForTest}s: {outsClient} {errsClient}") + if serverProcess.poll() != 0: + raise TestException( + f"Server execution failed {serverProcess.poll()}: {outsServer} {errsServer}") + if clientProcess.poll() != 0: + raise TestException( + f"Client execution failed {clientProcess.poll()}: {outsClient} {errsClient}") + + def _getClientCmd(self): + return self._getAppCmd("client", self._langClient) + + def _getServerCmd(self): + return self._getAppCmd("server", self._langServer) + + def _getAppCmd(self, clientServerString: str, lang: str): + appCmd = "" + if lang == PyLang: + appCmd = f"{self._name}_{clientServerString}.py" + elif lang == CLang: + appCmd = f"Linux/{self._transport}/gcc/{self._name}_{clientServerString}/Debug/{self._name}_{clientServerString}_{self._transport}_test" + else: + raise TestException(f"Unknown lang: {lang}") + + appCmd = os.path.join(self._caseDir, appCmd) + + if lang == PyLang: + appCmd = f"python3 {appCmd}" + elif lang == CLang: + pass + else: + raise TestException(f"Unknown lang: {lang}") + + return appCmd + + +class TestItem(pytest.Item): + def __init__(self, name, parent, case): + super(TestItem, self).__init__(name, parent) + self.case = case + + def runtest(self): + self.case.run() + + def repr_failure(self, excinfo): + """ called when self.runtest() raises an exception. """ + if isinstance(excinfo.value, TestException): + return excinfo.value.args[0] + return "failure: " + str(excinfo) + + def reportinfo(self): + return self.fspath, 0, "test case: " + self.parent.name.split("/")[0] + '::' + self.case.desc + + +def pytest_collect_file(path: py._path.local.LocalPath, parent: pytest.Collector): + if path.basename.startswith("test") and path.basename.endswith("client.py"): + return TestFile.from_parent(parent, fspath=py._path.local.LocalPath(path.dirname)) diff --git a/test/pytest.ini b/test/pytest.ini new file mode 100644 index 000000000..389dd28dc --- /dev/null +++ b/test/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +python_files = - diff --git a/test/test_annotations/test_annotations.py b/test/test_annotations/test_annotations.py new file mode 100644 index 000000000..8d1c8b69c --- /dev/null +++ b/test/test_annotations/test_annotations.py @@ -0,0 +1 @@ + diff --git a/test/test_annotations/test_annotations_client.py b/test/test_annotations/test_annotations_client.py new file mode 100644 index 000000000..31330a95e --- /dev/null +++ b/test/test_annotations/test_annotations_client.py @@ -0,0 +1,38 @@ +import sys +import os + +# pathToErpc = os.path.abspath(os.path.join( +# '..', 'erpc_python')) +# print(pathToErpc) +pathToErpc = os.path.abspath('erpc_python') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +# pathToErpc = os.path.abspath(os.path.join('..')) +# sys.path.insert(0, pathToErpc) +# print(pathToErpc) +pathToErpc = os.path.abspath('.') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +pathToErpc = os.path.abspath('test') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +from Linux.tcp.gcc.test_annotations_server.Debug import test, test_unit_test_common # nopep8 +from erpc_python import erpc # nopep8 +from common import unit_test_tcp_client # nopep8 + +clientManager = unit_test_tcp_client.getClientManager(erpc) + +annotateTestClient = test.client.AnnotateTestClient(clientManager) +commonClient = test_unit_test_common.client.CommonClient(clientManager) + + +def AnnotationServiceID(): + assert (5 == test.interface.IAnnotateTest.SERVICE_ID) + + +def testIfMyIntAndConstExist(): + assert (5 == annotateTestClient.testIfMyIntAndConstExist(5)) + + +def quit(): + commonClient.quit() diff --git a/test/test_annotations/test_annotations_server.py b/test/test_annotations/test_annotations_server.py new file mode 100644 index 000000000..ea55a9d52 --- /dev/null +++ b/test/test_annotations/test_annotations_server.py @@ -0,0 +1,43 @@ +import sys +import os + +# To access folders outside of current folder +# pathToErpc = os.path.abspath(os.path.join( +# '..', 'erpc_python')) +# print(pathToErpc) +# sys.path.insert(0, pathToErpc) +pathToErpc = os.path.abspath('erpc_python') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +# pathToErpc = os.path.abspath(os.path.join('..')) +# sys.path.insert(0, pathToErpc) +# print(pathToErpc) +pathToErpc = os.path.abspath('.') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +pathToErpc = os.path.abspath('test') +print(pathToErpc) +sys.path.insert(0, pathToErpc) +from erpc_python import erpc # nopep8 +from Linux.tcp.gcc.test_annotations_server.Debug import test, test_unit_test_common # nopep8 +from common import unit_test_tcp_server # nopep8 + + +class AnnotateTestServer(test.interface.IAnnotateTest): + def add(self, a, b): + return a+b + + def testIfFooStructExist(self, a): + pass + + def testIfMyEnumExist(self, a): + pass + + def testIfMyIntAndConstExist(self, a): + return a + +server,commonServer = unit_test_tcp_server.getServer(erpc, test_unit_test_common.interface.ICommon) +server.add_service(test.server.AnnotateTestService(AnnotateTestServer( +))) +server.add_service(test_unit_test_common.server.CommonService(commonServer)) +server.run()