[dts] [PATCH V1 2/2] memory_register: upstream automation script
yufengx.mo at intel.com
yufengx.mo at intel.com
Wed Jun 6 07:32:38 CEST 2018
From: yufengmx <yufengx.mo at intel.com>
DPDK has no custom example or unit test binary to test memory register and
unregister. The idea behind the testing process is to run dpdk unit test related
with memory. It is the only method to test this feature.
task is related with all dpdk memory management mechanism, including
malloc/mbuf/memory/mempool/memcpy/memzone. Run Compound test based on
driver/port number/socket type.
Signed-off-by: yufengmx <yufengx.mo at intel.com>
---
tests/TestSuite_memory_register.py | 423 +++++++++++++++++++++++++++++++++++++
1 file changed, 423 insertions(+)
create mode 100644 tests/TestSuite_memory_register.py
diff --git a/tests/TestSuite_memory_register.py b/tests/TestSuite_memory_register.py
new file mode 100644
index 0000000..e5bacf6
--- /dev/null
+++ b/tests/TestSuite_memory_register.py
@@ -0,0 +1,423 @@
+# BSD LICENSE
+#
+# Copyright(c) 2010-2018 Intel Corporation. All rights reserved.
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#
+# * Redistributions of source code must retain the above copyright
+# notice, this list of conditions and the following disclaimer.
+# * Redistributions in binary form must reproduce the above copyright
+# notice, this list of conditions and the following disclaimer in
+# the documentation and/or other materials provided with the
+# distribution.
+# * Neither the name of Intel Corporation nor the names of its
+# contributors may be used to endorse or promote products derived
+# from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+import os
+import time
+import re
+import random
+
+from datetime import datetime
+from socket import htons, htonl
+
+from packet import Packet, NVGRE, IPPROTO_NVGRE
+from scapy.sendrecv import sendp
+from scapy.utils import wrpcap, rdpcap, hexstr
+
+import utils
+from test_case import TestCase
+from exception import TimeoutException, VerifyFailure
+from settings import TIMEOUT
+from pmd_output import PmdOutput
+from settings import HEADER_SIZE
+from serializer import Serializer
+
+# used for debug
+from functools import wraps
+from pprint import pprint, pformat
+import inspect
+import traceback
+
+############################################################
+## used for binary process
+############################################################
+class ExecBinProcess(object):
+
+ def __init__(self, **kwargs):
+ # initialize process parameter
+ self.f_session = kwargs.get('fore_session')
+ self.d_session = kwargs.get('daemon_session')
+ self.logger = kwargs.get('logger')
+ self.name = kwargs.get('name')
+ self.output = kwargs.get('output')
+ self.target_code = kwargs.get('target_src')
+ # initialize process
+ self.process_pid = None
+ self.output_log = os.sep.join([self.output,
+ self.name.replace('/', '@') + '.log'])
+ # compile and get executable binary file
+ self.bin = self.compile(self.target_code, self.name)
+ self.bin_name = os.path.basename(self.bin)
+ # set process status
+ self.status = 'close'
+ self.create_test_temp_dir()
+
+ def create_test_temp_dir(self):
+ temp_dir = '/test/test'
+ if not self.is_existed(temp_dir):
+ cmd = "mkdir -p {0}".format(temp_dir)
+ self.d_console(cmd)
+
+ def f_console(self, cmds):
+ return self.execute_cmds(cmds, con_name='fore')
+
+ def d_console(self, cmds):
+ return self.execute_cmds(cmds, con_name='daemon')
+
+ def get_console(self, name):
+ if name == 'fore':
+ console = self.f_session.send_expect
+ msg_pipe = self.f_session.session.get_output_all
+ elif name == 'daemon':
+ console = self.d_session.send_expect
+ msg_pipe = self.d_session.session.get_output_all
+ else:
+ msg = "session has not set"
+ raise VerifyFailure(msg)
+
+ return console, msg_pipe
+
+ def execute_cmds(self, cmds, con_name='fore'):
+ console, msg_pipe = self.get_console(con_name)
+ if len(cmds) == 0:
+ return
+ if isinstance(cmds, (str, unicode)):
+ cmds = [cmds, '# ', 15]
+ if not isinstance(cmds[0], list):
+ cmds = [cmds]
+ if len(cmds) > 1:
+ outputs = []
+ else:
+ outputs = ''
+ #----------------------------------
+ for item in cmds:
+ expected_items = item[1]
+ if expected_items and isinstance(expected_items, (list, tuple)):
+ check_output = True
+ expected_str = expected_items[0] or '# '
+ else:
+ check_output = False
+ expected_str = expected_items or '# '
+
+ try:
+ if len(item) == 3:
+ timeout = int(item[2])
+ output = console(item[0], expected_str, timeout)
+ else:
+ #timeout = 5
+ output = console(item[0], expected_str)
+ except TimeoutException:
+ output = msg_pipe()
+ self.logger.info(output)
+ msg = "execute '{0}' timeout".format(item[0])
+ raise TimeoutException(msg)
+ finally:
+ pass
+ #----------------------------------
+ time.sleep(1)
+ if len(cmds) > 1:
+ outputs.append(output)
+ else:
+ outputs = output
+ if check_output and len(expected_items) >= 2:
+ self.logger.info(output)
+ expected_output = expected_items[1]
+ check_type = True if len(expected_items) == 2 \
+ else expected_items[2]
+ if check_type and expected_output in output:
+ msg = "expected '{0}' is in output".format(expected_output)
+ self.logger.info(msg)
+ elif not check_type and expected_output not in output:
+ msg = "unexpected '{0}' is not in output".format(
+ expected_output)
+ self.logger.info(msg)
+ else:
+ status = "isn't in" if check_type else "is in"
+ msg = "[{0}] {1} output".format(expected_output, status)
+ self.logger.error(msg)
+ raise VerifyFailure(msg)
+
+ return outputs
+
+ def compile(self, target_code, name):
+ compile_path = os.sep.join([target_code, name])
+ # using Dexamples options
+ cmds = ['make -C {0}'.format(compile_path), '# ', 180]
+ self.f_console(cmds)
+ # check executable binary file
+ tool_path = compile_path
+ key_words = [tool_path, 'build', 'app', 'test']
+ exec_bin = os.sep.join(key_words)
+ if not exec_bin or not self.is_existed(exec_bin):
+ msg = 'expected tool <{0}> does not exist'.format(exec_bin)
+ self.logger.error(msg)
+ raise VerifyFailure(msg)
+ return exec_bin
+
+ def is_existed(self, path, console_name='daemon'):
+ ''' check if file existed '''
+ cmd = "ls {0}".format(path)
+ self.execute_cmds(cmd, con_name=console_name)
+ cmd = "echo $?"
+ output = self.execute_cmds(cmd, con_name=console_name)
+ ret = True if output == "0" else False
+ return ret
+
+ def is_alive(self, process_name, check_status):
+ # check subprocess in task space
+ cmd = "ps aux | grep -i '%s' | grep -v grep | awk {'print $2'}"%(
+ process_name)
+ out = self.d_console([cmd, '# ', 5])
+ if out != "":
+ self.process_pid = out.splitlines()[0]
+ self.logger.info("{0}'s pid is {1}".format(self.bin_name,
+ self.process_pid))
+ status = True
+ else:
+ status = False
+
+ if check_status == 'start' and not status:
+ raise_flg = True
+ elif check_status == 'close' and status:
+ raise_flg = True
+ else:
+ raise_flg = False
+
+ if raise_flg:
+ raise VerifyFailure("{0} {1} failed".format(process_name,
+ check_status))
+ else:
+ self.logger.info("{0} {1} success".format(process_name,
+ check_status))
+
+ return status
+
+ def start(self, eal_option='', prompt='# ',
+ ps_type="block", timeout=None):
+ if self.status == 'running':
+ return
+ cmd =['{0} {1} 2>&1'.format(self.bin, eal_option), prompt, 15]
+ output = self.f_console(cmd)
+ # save output in log file
+ log = self.output_log
+ with open(log, 'wb') as fp:
+ fp.write(output)
+ #--------------------------------
+ # if example is block processs
+ # check if process has bootep up
+ if ps_type == "block":
+ time.sleep(10)
+ self.is_alive(self.bin, "start")
+ self.status = 'running'
+
+ def close(self, quit=None, log=None):
+ output = self.f_session.session.get_output_all()
+ log = self.output_log
+ with open(log, 'a') as fp:
+ fp.write(output)
+ if not quit:
+ return output
+ if self.status == 'close':
+ return output
+ if quit.startswith("SIG"):
+ cmds =['kill -TERM {0}'.format(self.process_pid), '# ']
+ output = self.d_console(cmds)
+ else:
+ cmds =[quit, '# ']
+ output = self.f_console(cmds)
+ max_try = 3
+ exec_info = None
+ # check close action is done
+ while max_try:
+ try:
+ self.is_alive(self.bin, 'close')
+ except Exception as e:
+ exec_info = e
+ time.sleep(2)
+ max_try -= 1
+ continue
+ else:
+ break
+ else:
+ raise exec_info
+ self.status = 'close'
+
+ return output
+#############
+
+#############
+class TestMemoryRegister(TestCase):
+ #
+ # framework related methods/parameter
+ #
+ @property
+ def dut_alt_session(self):
+ return self.dut.alt_session
+
+ @property
+ def dut_session(self):
+ return self.dut.session
+
+ @property
+ def ports(self):
+ return self.dut.ports_info
+ #---------------------------------------------------------
+ # testing methods
+ #---------------------------------------------------------
+ def get_options(self, name):
+ options = {
+ #----------------------------
+ 'malloc_autotest': {
+ 'option': ' -n 1 -c f',
+ 'expected': 'test_malloc_bad_params() passed',
+ 'timeout': 30},
+ #----------------------------
+ 'mbuf_autotest': {
+ 'option': ' -n 1 -c f',
+ 'timeout': 180},
+ #----------------------------
+ 'memcpy_autotest': {
+ 'option': ' -n 1 -c f',
+ 'timeout': 180},
+ #----------------------------
+ 'memcpy_perf_autotest': {
+ 'option': ' -n 1 -c f',
+ 'timeout': 3000},
+ #----------------------------
+ 'memory_autotest': {
+ 'option': ' -n 1 -c f',
+ 'timeout': 1200},
+ #----------------------------
+ 'mempool_autotest': {
+ 'option': ' -n 1 -c f',
+ 'timeout': 300},
+ #----------------------------
+ 'mempool_perf_autotest': {
+ 'option': ' -n 1 -c e',
+ 'timeout': 1200},
+ #----------------------------
+ 'memzone_autotest': {
+ 'option': ' -n 1 -c f',
+ 'prompt': 'RTE>>',
+ 'timeout': 180},}
+
+ return options[name]
+
+ def unit_test(self, name):
+ config = self.get_options(name)
+ option = config.get('option')
+ timeout = config.get('timeout') or 180
+ prompt = config.get('prompt') or "R.*T.*E.*>.*>"
+ expected = config.get('expected') or 'Test OK'
+ quit = 'quit'
+ #--------------------------------------------
+ try:
+ self.unit_obj.start(option, prompt=prompt, timeout=timeout)
+ self.unit_obj.f_console([name, [prompt, expected, True], timeout])
+ except Exception as e:
+ self.unit_obj.close(quit=quit)
+ msg = "run {0} failed".format(name)
+ raise VerifyFailure(msg)
+ finally:
+ self.unit_obj.close(quit=quit)
+
+ def create_unit_test(self):
+ psInfo = {
+ 'fore_session': self.dut_session,
+ 'daemon_session': self.dut_alt_session,
+ 'logger': self.logger,
+ 'name': 'test',
+ 'target_src': self.target_source,
+ 'output': self.output_path,}
+ self.unit_obj = ExecBinProcess(**psInfo)
+ #
+ # Test cases.
+ #
+ def set_up_all(self):
+ """
+ Run before each test suite
+ """#------------------------------------------------------------------
+ # initialize ports topology
+ self.dut_ports = self.dut.get_ports()
+ self.verify(len(self.dut_ports) >= 1, "Insufficient ports")
+ #------------------------------------------------------------------
+ self.target_source = self.dut.base_dir
+ #------------------------------------------------------------------
+ # get output path
+ if self.logger.log_path.startswith(os.sep):
+ self.output_path = self.logger.log_path
+ else:
+ cur_path = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
+ self.output_path = cur_path + os.sep + self.logger.log_path
+ #------------------------------------------------------------------
+ # create unit test tool ``test``
+ self.create_unit_test()
+
+ def set_up(self):
+ """
+ Run before each test case.
+ """
+ pass
+
+ def tear_down(self):
+ """
+ Run after each test case.
+ """
+ pass
+
+ def tear_down_all(self):
+ """
+ Run after each test suite.
+ """
+ self.unit_obj.close()
+
+ def test_dpdk_malloc_autotest(self):
+ self.unit_test('malloc_autotest')
+
+ def test_dpdk_mbuf_autotest(self):
+ self.unit_test('mbuf_autotest')
+
+ def test_dpdk_memcpy_autotest(self):
+ self.unit_test('memcpy_autotest')
+
+ def test_dpdk_memcpy_perf_autotest(self):
+ self.unit_test('memcpy_perf_autotest')
+
+ def test_dpdk_memory_autotest(self):
+ self.unit_test('memory_autotest')
+
+ def test_dpdk_mempool_autotest(self):
+ self.unit_test('mempool_autotest')
+
+ def test_dpdk_mempool_perf_autotest(self):
+ self.unit_test('mempool_perf_autotest')
+
+ def test_dpdk_memzone_autotest(self):
+ self.unit_test('memzone_autotest')
--
1.9.3
More information about the dts
mailing list