Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Runing c and py tests. #270

Draft
wants to merge 1 commit into
base: develop
Choose a base branch
from
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions test/common/unit_test_tcp_client.py
Original file line number Diff line number Diff line change
@@ -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
13 changes: 13 additions & 0 deletions test/common/unit_test_tcp_server.py
Original file line number Diff line number Diff line change
@@ -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()
125 changes: 125 additions & 0 deletions test/conftest.py
Original file line number Diff line number Diff line change
@@ -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))
2 changes: 2 additions & 0 deletions test/pytest.ini
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
[pytest]
python_files = -
1 change: 1 addition & 0 deletions test/test_annotations/test_annotations.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@

38 changes: 38 additions & 0 deletions test/test_annotations/test_annotations_client.py
Original file line number Diff line number Diff line change
@@ -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()
43 changes: 43 additions & 0 deletions test/test_annotations/test_annotations_server.py
Original file line number Diff line number Diff line change
@@ -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()