@echo " startup.conf file is present"
@echo " GDB=<path> - gdb binary to use for debugging"
@echo " PLATFORM=<name> - target platform. default is vpp"
- @echo " TEST=<filter> - apply filter to test set, see test-help"
@echo " DPDK_CONFIG=<conf> - add specified dpdk config commands to"
@echo " autogenerated startup.conf"
@echo " (e.g. \"no-pci\" )"
rebuild-release: wipe-release build-release
export TEST_DIR ?= $(WS_ROOT)/test
-export RND_SEED ?= $(shell python3 -c 'import time; print(time.time())')
define test
$(if $(filter-out $(2),retest),make -C $(BR) PLATFORM=vpp TAG=$(1) vpp-install,)
OS_ID=$(OS_ID) \
RND_SEED=$(RND_SEED) \
CACHE_OUTPUT=$(CACHE_OUTPUT) \
+ TAG=$(1) \
$(2)
endef
.PHONY: test-all
test-all:
- $(eval EXTENDED_TESTS=yes)
+ $(eval EXTENDED_TESTS=1)
$(call test,vpp,test)
.PHONY: test-all-debug
test-all-debug:
- $(eval EXTENDED_TESTS=yes)
+ $(eval EXTENDED_TESTS=1)
$(call test,vpp_debug,test)
.PHONY: papi-wipe
.PHONY: test-cov
test-cov:
- $(eval EXTENDED_TESTS=yes)
+ $(eval EXTENDED_TESTS=1)
$(call test,vpp_gcov,cov)
.PHONY: test-wipe-cov
.PHONY: retest-all
retest-all:
- $(eval EXTENDED_TESTS=yes)
+ $(eval EXTENDED_TESTS=1)
$(call test,vpp,retest)
.PHONY: retest-all-debug
retest-all-debug:
- $(eval EXTENDED_TESTS=yes)
+ $(eval EXTENDED_TESTS=1)
$(call test,vpp_debug,retest)
.PHONY: test-start-vpp-in-gdb
::
- $ make test-help
- test - build and run (basic) functional tests
- test-debug - build and run (basic) functional tests (debug build)
- test-all - build and run functional and extended tests
- test-all-debug - build and run functional and extended tests (debug build)
- retest - run functional tests
- retest-debug - run functional tests (debug build)
- retest-all - run functional and extended tests
- retest-all-debug - run functional and extended tests (debug build)
- test-cov - generate code coverage report for test framework
- test-gcov - build and run functional tests (gcov build)
- test-wipe - wipe (temporary) files generated by unit tests
- test-wipe-cov - wipe code coverage report for test framework
- test-wipe-doc - wipe documentation for test framework
- test-wipe-papi - rebuild vpp_papi sources
- test-wipe-all - wipe (temporary) files generated by unit tests, docs, and coverage
- test-shell - enter shell with test environment
- test-shell-debug - enter shell with test environment (debug build)
- test-checkstyle - check PEP8 compliance for test framework
- test-refresh-deps - refresh the Python dependencies for the tests
-
- Arguments controlling test runs:
- V=[0|1|2] - set test verbosity level
- 0=ERROR, 1=INFO, 2=DEBUG
- TEST_JOBS=[<n>|auto] - use at most <n> parallel python processes for test execution, if auto, set to number of available cpus (default: 1)
- MAX_VPP_CPUS=[<n>|auto]- use at most <n> cpus for running vpp main and worker threads, if auto, set to number of available cpus (default: auto)
- CACHE_OUTPUT=[0|1] - cache VPP stdout/stderr and log as one block after test finishes (default: 1)
- FAILFAST=[0|1] - fail fast if 1, complete all tests if 0
- TIMEOUT=<timeout> - fail test suite if any single test takes longer than <timeout> (in seconds) to finish (default: 600)
- RETRIES=<n> - retry failed tests <n> times
- DEBUG=<type> - set VPP debugging kind
- DEBUG=core - detect coredump and load it in gdb on crash
- DEBUG=gdb - allow easy debugging by printing VPP PID
- and waiting for user input before running
- and tearing down a testcase
- DEBUG=gdbserver - run gdb inside a gdb server, otherwise
- same as above
- DEBUG=attach - attach test case to already running vpp in gdb (see test-start-vpp-in-gdb)
-
- STEP=[yes|no] - ease debugging by stepping through a testcase
- SANITY=[yes|no] - perform sanity import of vpp-api/sanity vpp run before running tests (default: yes)
- EXTENDED_TESTS=[1|y] - used by '[re]test-all' & '[re]test-all-debug' to run extended tests
- TEST=<filter> - filter the set of tests:
- by file-name - only run tests from specified file, e.g. TEST=test_bfd selects all tests from test_bfd.py
- by file-suffix - same as file-name, but 'test_' is omitted e.g. TEST=bfd selects all tests from test_bfd.py
- by wildcard - wildcard filter is <file>.<class>.<test function>, each can be replaced by '*'
- e.g. TEST='test_bfd.*.*' is equivalent to above example of filter by file-name
- TEST='bfd.*.*' is equivalent to above example of filter by file-suffix
- TEST='bfd.BFDAPITestCase.*' selects all tests from test_bfd.py which are part of BFDAPITestCase class
- TEST='bfd.BFDAPITestCase.test_add_bfd' selects a single test named test_add_bfd from test_bfd.py/BFDAPITestCase
- TEST='*.*.test_add_bfd' selects all test functions named test_add_bfd from all files/classes
-
- VARIANT=<variant> - specify which march node variant to unit test
- e.g. VARIANT=skx test the skx march variants
- e.g. VARIANT=icl test the icl march variants
-
- COREDUMP_SIZE=<size> - pass <size> as unix { coredump-size <size> } argument to vpp
- e.g. COREDUMP_SIZE=4g
- COREDUMP_SIZE=unlimited
- COREDUMP_COMPRESS=1 - compress core files if not debugging them
- EXTERN_TESTS=<path> - path to out-of-tree test_<name>.py files containing test cases
- EXTERN_PLUGINS=<path> - path to out-of-tree plugins to be loaded by vpp under test
- EXTERN_COV_DIR=<path> - path to out-of-tree prefix, where source, object and .gcda files can be found for coverage report
-
- PROFILE=1 - enable profiling of test framework via cProfile module
- PROFILE_SORT_BY=opt - sort profiling report by opt - consult cProfile documentation for possible values (default: cumtime)
- PROFILE_OUTPUT=file - output profiling info to file - use absolute path (default: stdout)
-
- TEST_DEBUG=1 - turn on debugging of the test framework itself (expert)
-
- SKIP_AARCH64=1 - skip tests that are failing on the ARM platorm in FD.io CI
-
- RND_SEED=seed - Seed RND with given seed
-
- Starting VPP in GDB for use with DEBUG=attach:
-
- test-start-vpp-in-gdb - start VPP in gdb (release)
- test-start-vpp-debug-in-gdb - start VPP in gdb (debug)
-
- Arguments controlling VPP in GDB runs:
-
- VPP_IN_GDB_TMP_DIR - specify directory to run VPP IN (default: /tmp/unittest-attach-gdb)
- VPP_IN_GDB_NO_RMDIR=0 - don't remove existing tmp dir but fail instead
- VPP_IN_GDB_CMDLINE=1 - add 'interactive' to VPP arguments to run with command line
-
- Creating test documentation
- test-doc - generate documentation for test framework
- test-wipe-doc - wipe documentation for test framework
-
- Creating test code coverage report
- test-cov - generate code coverage report for test framework
- test-wipe-cov - wipe code coverage report for test framework
-
- Verifying code-style
- test-checkstyle - check PEP8 compliance
+ $ make test-help
+ Running tests:
+
+ test - build and run (basic) functional tests
+ test-debug - build and run (basic) functional tests (debug build)
+ test-all - build and run functional and extended tests
+ test-all-debug - build and run functional and extended tests (debug build)
+ retest - run functional tests
+ retest-debug - run functional tests (debug build)
+ retest-all - run functional and extended tests
+ retest-all-debug - run functional and extended tests (debug build)
+ test-cov - generate code coverage report for test framework
+ test-gcov - build and run functional tests (gcov build)
+ test-wipe - wipe (temporary) files generated by unit tests
+ test-wipe-cov - wipe code coverage report for test framework
+ test-wipe-papi - rebuild vpp_papi sources
+ test-wipe-all - wipe (temporary) files generated by unit tests, and coverage
+ test-shell - enter shell with test environment
+ test-shell-debug - enter shell with test environment (debug build)
+ test-checkstyle - check PEP8 compliance for test framework
+ test-refresh-deps - refresh the Python dependencies for the tests
+
+ Arguments controlling test runs:
+
+ V=[0|1|2] - set test verbosity level
+ 0=ERROR, 1=INFO, 2=DEBUG
+ TEST_JOBS=[<n>|auto] - use at most <n> parallel python processes for test execution, if auto, set to number of available cpus (default: 1)
+ MAX_VPP_CPUS=[<n>|auto]- use at most <n> cpus for running vpp main and worker threads, if auto, set to number of available cpus (default: auto)
+ CACHE_OUTPUT=[0|n|no] - disable cache VPP stdout/stderr and log as one block after test finishes (default: yes)
+ FAILFAST=[1|y|yes] - fail fast if 1, otherwise complete all tests
+ TIMEOUT=<timeout> - fail test suite if any single test takes longer than <timeout> (in seconds) to finish (default: 600)
+ RETRIES=<n> - retry failed tests <n> times
+ DEBUG=<type> - set VPP debugging kind
+ DEBUG=core - detect coredump and load it in gdb on crash
+ DEBUG=gdb - allow easy debugging by printing VPP PID
+ and waiting for user input before running
+ and tearing down a testcase
+ DEBUG=gdbserver - run gdb inside a gdb server, otherwise
+ same as above
+ DEBUG=attach - attach test case to already running vpp in gdb (see test-start-vpp-in-gdb)
+ STEP=[1|y|yes] - enable stepping through a testcase (for testcase debugging)
+ SANITY=[0|n|no] - disable sanity import of vpp-api/sanity vpp run before running tests
+ EXTENDED_TESTS=[1|y|yes] - run extended tests
+ TEST=<filter> - filter the set of tests:
+ by file-name - only run tests from specified file, e.g. TEST=test_bfd selects all tests from test_bfd.py
+ by file-suffix - same as file-name, but 'test_' is omitted e.g. TEST=bfd selects all tests from test_bfd.py
+ by wildcard - wildcard filter is <file>.<class>.<test function>, each can be replaced by '*'
+ e.g. TEST='test_bfd.*.*' is equivalent to above example of filter by file-name
+ TEST='bfd.*.*' is equivalent to above example of filter by file-suffix
+ TEST='bfd.BFDAPITestCase.*' selects all tests from test_bfd.py which are part of BFDAPITestCase class
+ TEST='bfd.BFDAPITestCase.test_add_bfd' selects a single test named test_add_bfd from test_bfd.py/BFDAPITestCase
+ TEST='*.*.test_add_bfd' selects all test functions named test_add_bfd from all files/classes
+ VARIANT=<variant> - specify which march node variant to unit test
+ e.g. VARIANT=skx test the skx march variants
+ e.g. VARIANT=icl test the icl march variants
+ COREDUMP_SIZE=<size> - pass <size> as unix { coredump-size <size> } argument to vpp
+ e.g. COREDUMP_SIZE=4g
+ COREDUMP_SIZE=unlimited
+ COREDUMP_COMPRESS=[1|y|yes] - compress core files if not debugging them
+ EXTERN_TESTS=<path> - path to out-of-tree test_<name>.py files containing test cases
+ EXTERN_PLUGINS=<path> - path to out-of-tree plugins to be loaded by vpp under test
+ EXTERN_COV_DIR=<path> - path to out-of-tree prefix, where source, object and .gcda files can be found for coverage report
+ PROFILE=[1|y|yes] - enable profiling of test framework via cProfile module
+ PROFILE_SORT_BY=opt - sort profiling report by opt - consult cProfile documentation for possible values (default: cumtime)
+ PROFILE_OUTPUT=file - output profiling info to file - use absolute path (default: stdout)
+ TEST_DEBUG=[1|y|yes] - enable debugging of the test framework itself (expert)
+ API_FUZZ=[1|y|yes] - enable VPP api fuzz testing
+ RND_SEED=<seed> - Seed RND with given seed
+
+ Starting VPP in GDB for use with DEBUG=attach:
+
+ test-start-vpp-in-gdb - start VPP in gdb (release)
+ test-start-vpp-debug-in-gdb - start VPP in gdb (debug)
+
+ Creating test code coverage report:
+
+ test-cov - generate code coverage report for test framework
+ test-wipe-cov - wipe code coverage report for test framework
+
+ Verifying code-style:
+
+ test-checkstyle - check PEP8 compliance
endif
ifeq ($(PROFILE),1)
-PYTHON_PROFILE_OPTS=-m cProfile $(PROFILE_OUTPUT_OPTS) -s $(PROFILE_SORT_BY)
+PYTHON_OPTS="-m cProfile $(PROFILE_OUTPUT_OPTS) -s $(PROFILE_SORT_BY)"
FORCE_FOREGROUND=1
endif
-UNITTEST_EXTRA_OPTS=
-UNITTEST_FAILFAST_OPTS=
-
-ifeq ($(FAILFAST),1)
-UNITTEST_EXTRA_OPTS=-f
-endif
-
-ifneq ($(EXTERN_TESTS),)
-UNITTEST_EXTRA_OPTS=$(UNITTEST_FAILFAST_OPTS) -d $(EXTERN_TESTS)
-endif
-
VENV_PATH=$(TEST_DIR)/venv
ifeq ($(TEST_DEBUG),1)
PYTHON_INTERP=$(PYTHON)
endif
+ifeq ($(V),)
+V=0
+endif
+
PYTHON_VERSION=$(shell $(PYTHON_INTERP) -c 'import sys; print(sys.version_info.major)')
PIP_VERSION=22.0.3
# Keep in sync with requirements.txt
PLUGIN_SRC_DIR=$(INTERN_PLUGIN_SRC_DIR)
endif
-define retest-func
-@env VPP_IN_GDB=$(VPP_IN_GDB) FORCE_FOREGROUND=$(FORCE_FOREGROUND) FAILED_DIR=$(FAILED_DIR) VENV_PATH=$(VENV_PATH) scripts/setsid_wrapper.sh $(FORCE_FOREGROUND) $(VENV_PATH)/bin/activate $(PYTHON_INTERP) $(PYTHON_PROFILE_OPTS) run_tests.py -d $(TEST_DIR) $(UNITTEST_EXTRA_OPTS) || env FAILED_DIR=$(FAILED_DIR) COMPRESS_FAILED_TEST_LOGS=$(COMPRESS_FAILED_TEST_LOGS) scripts/compress_failed.sh
-endef
-
.PHONY: sanity
-ifeq ($(SANITY),no)
-SANITY_IMPORT_VPP_PAPI_CMD=true
-SANITY_RUN_VPP_CMD=true
-else
-SANITY_IMPORT_VPP_PAPI_CMD=source $(VENV_PATH)/bin/activate && python3 sanity_import_vpp_papi.py
-SANITY_RUN_VPP_CMD=source $(VENV_PATH)/bin/activate && python3 sanity_run_vpp.py
-endif
-
ifndef TEST_JOBS
PARALLEL_ILLEGAL=0
+TEST_JOBS=1
else ifeq ($(FORCE_FOREGROUND),0)
PARALLEL_ILLEGAL=0
-else ifeq ($(TEST_JOBS),auto)
-PARALLEL_ILLEGAL=0
-else ifeq ($(TEST_JOBS),1)
+else ifneq ($(findstring $(TEST_JOBS),1 auto),)
PARALLEL_ILLEGAL=0
else
PARALLEL_ILLEGAL=1
endif
+ifneq ($(DEBUG),)
+SANITY=no
+endif
+
+ifneq ($(findstring $(SANITY),0 n no),)
+SANITY_IMPORT_VPP_PAPI_CMD=true
+ARG0=
+else
+SANITY_IMPORT_VPP_PAPI_CMD=source $(VENV_PATH)/bin/activate && $(PYTHON_INTERP) sanity_import_vpp_papi.py
+ARG0=--sanity
+endif
+
+ARG1=
+ifneq ($(findstring $(FAILFAST),1 y yes),)
+ARG1=--failfast
+endif
+
+ARG2=
+ifneq ($(findstring $(EXTENDED_TESTS),1 y yes),)
+ARG2=--extended
+endif
+
+ARG3=
+ifneq ($(EXTERN_TESTS),)
+ARG3=--test-src-dir $(EXTERN_TESTS)
+endif
+
+ARG4=
+ifneq ($(findstring $(FORCE_FOREGROUND),1 y yes),)
+ARG4=--force-foreground
+endif
+
+ARG5=
+ifneq ($(findstring $(COREDUMP_COMPRESS),1 y yes),)
+ARG5=--compress-core
+endif
+
+ARG6=
+ifneq ($(findstring $(STEP),1 y yes),)
+ARG6=--step
+endif
+
+ARG7=
+ifneq ($(findstring $(TESTS_GCOV),1 y yes),)
+ARG7=--gcov
+endif
+
+ARG8=
+ifneq ($(EXTERN_PLUGINS),)
+ARG8=--extern-plugin-dir=$(EXTERN_PLUGINS)
+endif
+
+ARG9=
+ifneq ($(DEBUG),)
+ARG9=--debug=$(DEBUG)
+endif
+
+ARG10=
+ifneq ($(COREDUMP_SIZE),)
+ARG10=--coredump-size=$(COREDUMP_SIZE)
+endif
+
+ARG11=
+ifneq ($(VARIANT),)
+ARG11=--variant=$(VARIANT)
+endif
+
+ARG12=--cache-vpp-output
+ifneq ($(findstring $(CACHE_OUTPUT),0 n no),)
+ARG12=
+endif
+
+ARG13=
+ifneq ($(MAX_VPP_CPUS),)
+ARG13=--max-vpp-cpus=$(MAX_VPP_CPUS)
+endif
+
+ARG14=
+ifneq ($(TIMEOUT),)
+ARG14=--timeout=$(TIMEOUT)
+endif
+
+ARG15=
+ifneq ($(findstring $(TEST_DEBUG),1 y yes),)
+ARG15=--debug-framework
+endif
+
+ARG16=
+ifneq ($(findstring $(API_FUZZ),1 y yes),)
+ARG16=--api-fuzz=on
+endif
+
+EXTRA_ARGS=$(ARG0) $(ARG1) $(ARG2) $(ARG3) $(ARG4) $(ARG5) $(ARG6) $(ARG7) $(ARG8) $(ARG9) $(ARG10) $(ARG11) $(ARG12) $(ARG13) $(ARG14) $(ARG15) $(ARG16)
+
+RUN_TESTS_ARGS=--failed-dir=$(FAILED_DIR) --verbose=$(V) --jobs=$(TEST_JOBS) --filter=$(TEST) --retries=$(RETRIES) --venv-dir=$(VENV_PATH) --vpp-ws-dir=$(WS_ROOT) --vpp-tag=$(TAG) --rnd-seed=$(RND_SEED) --vpp-worker-count="$(VPP_WORKER_COUNT)" --keep-pcaps --python-opts=$(PYTHON_OPTS) $(PLUGIN_PATH_ARGS) $(TEST_PLUGIN_PATH_ARGS) $(EXTRA_ARGS)
+
+define retest-func
+@scripts/run.sh $(RUN_TESTS_ARGS) || env FAILED_DIR=$(FAILED_DIR) COMPRESS_FAILED_TEST_LOGS=$(COMPRESS_FAILED_TEST_LOGS) scripts/compress_failed.sh
+endef
+
sanity: test-dep
@bash -c "test $(PARALLEL_ILLEGAL) -eq 0 ||\
(echo \"*******************************************************************\" &&\
echo \"* 2. execute debugger: gdb python -ex 'run sanity_import_vpp_papi.py'\" &&\
echo \"*******************************************************************\" &&\
false)"
- @bash -c "$(SANITY_RUN_VPP_CMD) ||\
- (echo \"*******************************************************************\" &&\
- echo \"* Sanity check failed, cannot run vpp\" &&\
- echo \"*******************************************************************\" &&\
- false)"
$(FAILED_DIR): reset
@mkdir -p $@
echo PYTHONPATH=$(PYTHONPATH);\
echo RND_SEED=$(RND_SEED);\
echo VPP_BUILD_DIR=$(VPP_BUILD_DIR);\
- echo VPP_BIN=$(VPP_BIN);\
echo VPP_PLUGIN_PATH=$(VPP_PLUGIN_PATH);\
echo VPP_TEST_PLUGIN_PATH=$(VPP_TEST_PLUGIN_PATH);\
echo VPP_INSTALL_PATH=$(VPP_INSTALL_PATH);\
echo EXTERN_TESTS=$(EXTERN_TESTS);\
echo EXTERN_PLUGINS=$(EXTERN_PLUGINS);\
- echo EXTERN_COV_DIR=$(EXTERN_COV_DIR);\
+ echo EXTERN_COV_DIR=$(EXTERN_COV_DIR);\
echo LD_LIBRARY_PATH=$(LD_LIBRARY_PATH);\
echo '***';\
exec </dev/tty" | bash -i
.PHONY: start-gdb
start-gdb: sanity
- $(eval VPP_IN_GDB=1)
- $(eval FORCE_FOREGROUND=1)
- $(call retest-func)
+ @bash -c "source $(VENV_PATH)/bin/activate && python3 -c 'from debug import start_vpp_in_gdb; start_vpp_in_gdb()' $(RUN_TESTS_ARGS)"
.PHONY: checkstyle
checkstyle: $(PIP_INSTALL_DONE)
@echo " test-refresh-deps - refresh the Python dependencies for the tests"
@echo ""
@echo "Arguments controlling test runs:"
+ @echo ""
@echo " V=[0|1|2] - set test verbosity level"
@echo " 0=ERROR, 1=INFO, 2=DEBUG"
@echo " TEST_JOBS=[<n>|auto] - use at most <n> parallel python processes for test execution, if auto, set to number of available cpus (default: 1)"
@echo " MAX_VPP_CPUS=[<n>|auto]- use at most <n> cpus for running vpp main and worker threads, if auto, set to number of available cpus (default: auto)"
- @echo " CACHE_OUTPUT=[0|1] - cache VPP stdout/stderr and log as one block after test finishes (default: 1)"
- @echo " FAILFAST=[0|1] - fail fast if 1, complete all tests if 0"
+ @echo " CACHE_OUTPUT=[0|n|no] - disable cache VPP stdout/stderr and log as one block after test finishes (default: yes)"
+ @echo " FAILFAST=[1|y|yes] - fail fast if 1, otherwise complete all tests"
@echo " TIMEOUT=<timeout> - fail test suite if any single test takes longer than <timeout> (in seconds) to finish (default: 600)"
@echo " RETRIES=<n> - retry failed tests <n> times"
@echo " DEBUG=<type> - set VPP debugging kind"
@echo " DEBUG=gdbserver - run gdb inside a gdb server, otherwise"
@echo " same as above"
@echo " DEBUG=attach - attach test case to already running vpp in gdb (see test-start-vpp-in-gdb)"
- @echo ""
- @echo " STEP=[yes|no] - ease debugging by stepping through a testcase"
- @echo " SANITY=[yes|no] - perform sanity import of vpp-api/sanity vpp run before running tests (default: yes)"
- @echo " EXTENDED_TESTS=[1|y] - used by '[re]test-all' & '[re]test-all-debug' to run extended tests"
+ @echo " STEP=[1|y|yes] - enable stepping through a testcase (for testcase debugging)"
+ @echo " SANITY=[0|n|no] - disable sanity import of vpp-api/sanity vpp run before running tests"
+ @echo " EXTENDED_TESTS=[1|y|yes] - run extended tests"
@echo " TEST=<filter> - filter the set of tests:"
@echo " by file-name - only run tests from specified file, e.g. TEST=test_bfd selects all tests from test_bfd.py"
@echo " by file-suffix - same as file-name, but 'test_' is omitted e.g. TEST=bfd selects all tests from test_bfd.py"
@echo " TEST='bfd.BFDAPITestCase.*' selects all tests from test_bfd.py which are part of BFDAPITestCase class"
@echo " TEST='bfd.BFDAPITestCase.test_add_bfd' selects a single test named test_add_bfd from test_bfd.py/BFDAPITestCase"
@echo " TEST='*.*.test_add_bfd' selects all test functions named test_add_bfd from all files/classes"
- @echo ""
@echo " VARIANT=<variant> - specify which march node variant to unit test"
@echo " e.g. VARIANT=skx test the skx march variants"
@echo " e.g. VARIANT=icl test the icl march variants"
- @echo ""
@echo " COREDUMP_SIZE=<size> - pass <size> as unix { coredump-size <size> } argument to vpp"
@echo " e.g. COREDUMP_SIZE=4g"
@echo " COREDUMP_SIZE=unlimited"
- @echo " COREDUMP_COMPRESS=1 - compress core files if not debugging them"
+ @echo " COREDUMP_COMPRESS=[1|y|yes] - compress core files if not debugging them"
@echo " EXTERN_TESTS=<path> - path to out-of-tree test_<name>.py files containing test cases"
@echo " EXTERN_PLUGINS=<path> - path to out-of-tree plugins to be loaded by vpp under test"
@echo " EXTERN_COV_DIR=<path> - path to out-of-tree prefix, where source, object and .gcda files can be found for coverage report"
- @echo ""
- @echo " PROFILE=1 - enable profiling of test framework via cProfile module"
+ @echo " PROFILE=[1|y|yes] - enable profiling of test framework via cProfile module"
@echo " PROFILE_SORT_BY=opt - sort profiling report by opt - consult cProfile documentation for possible values (default: cumtime)"
@echo " PROFILE_OUTPUT=file - output profiling info to file - use absolute path (default: stdout)"
- @echo ""
- @echo " TEST_DEBUG=1 - turn on debugging of the test framework itself (expert)"
- @echo ""
- @echo " SKIP_AARCH64=1 - skip tests that are failing on the ARM platorm in FD.io CI"
- @echo ""
- @echo " RND_SEED=seed - Seed RND with given seed"
+ @echo " TEST_DEBUG=[1|y|yes] - enable debugging of the test framework itself (expert)"
+ @echo " API_FUZZ=[1|y|yes] - enable VPP api fuzz testing"
+ @echo " RND_SEED=<seed> - Seed RND with given seed"
@echo ""
@echo "Starting VPP in GDB for use with DEBUG=attach:"
@echo ""
@echo " test-start-vpp-in-gdb - start VPP in gdb (release)"
@echo " test-start-vpp-debug-in-gdb - start VPP in gdb (debug)"
@echo ""
- @echo "Arguments controlling VPP in GDB runs:"
- @echo " "
- @echo " VPP_IN_GDB_TMP_DIR - specify directory to run VPP IN (default: /tmp/unittest-attach-gdb)"
- @echo " VPP_IN_GDB_NO_RMDIR=0 - don't remove existing tmp dir but fail instead"
- @echo " VPP_IN_GDB_CMDLINE=1 - add 'interactive' to VPP arguments to run with command line"
+ @echo "Creating test code coverage report:"
@echo ""
- @echo "Creating test code coverage report"
@echo " test-cov - generate code coverage report for test framework"
@echo " test-wipe-cov - wipe code coverage report for test framework"
@echo ""
- @echo "Verifying code-style"
+ @echo "Verifying code-style:"
+ @echo ""
@echo " test-checkstyle - check PEP8 compliance"
@echo ""
--- /dev/null
+import argparse
+import os
+import psutil
+import textwrap
+import time
+
+
+def positive_int_or_default(default):
+ def positive_integer(v):
+ if v is None or v == "":
+ return default
+ return int(v)
+ return positive_integer
+
+
+def positive_int_or_auto(v):
+ if v is None or v in ("", "auto"):
+ return "auto"
+ if int(v) <= 0:
+ raise ValueError("value must be positive or auto")
+ return int(v)
+
+
+def int_or_auto(v):
+ if v is None or v in ("", "auto"):
+ return "auto"
+ if int(v) < 0:
+ raise ValueError("value must be positive or auto")
+ return int(v)
+
+
+def int_choice_or_default(options, default):
+ assert default in options
+
+ def choice(v):
+ if v is None or v == "":
+ return default
+ if int(v) in options:
+ return int(v)
+ raise ValueError("invalid choice")
+ return choice
+
+
+def worker_config(v):
+ if v is None or v == "":
+ return 0
+ if v.startswith("workers "):
+ return(int(v.split(" ")[1]))
+ return int(v)
+
+
+def directory(v):
+ if not os.path.isdir(v):
+ raise ValueError(f"provided path '{v}' doesn't exist "
+ "or is not a directory")
+ return v
+
+
+def directory_verify_or_create(v):
+ if not os.path.isdir(v):
+ os.mkdir(v)
+ return v
+
+
+parser = argparse.ArgumentParser(description="VPP unit tests",
+ formatter_class=argparse.RawTextHelpFormatter)
+
+parser.add_argument("--failfast", action="store_true",
+ help="stop running tests on first failure")
+
+parser.add_argument("--test-src-dir", action="append", type=directory,
+ help="directory containing test files "
+ "(may be specified multiple times) "
+ "(VPP_WS_DIR/test is added automatically to the set)")
+
+default_verbose = 0
+
+parser.add_argument("--verbose", action="store", default=default_verbose,
+ type=int_choice_or_default((0, 1, 2), default_verbose),
+ help="verbosity setting - 0 - least verbose, "
+ "2 - most verbose (default: 0)")
+
+default_test_run_timeout = 600
+
+parser.add_argument("--timeout", action="store",
+ type=positive_int_or_default(default_test_run_timeout),
+ default=default_test_run_timeout,
+ metavar="TEST_RUN_TIMEOUT",
+ help="test run timeout in seconds - per test "
+ f"(default: {default_test_run_timeout})")
+
+parser.add_argument("--failed-dir", action="store", type=directory,
+ help="directory containing failed tests")
+
+filter_help_string = """\
+expression consists of 3 string selectors separated by '.' separators:
+
+ <file>.<class>.<function>
+
+- selectors restrict which files/classes/functions are run
+- selector can be replaced with '*' or omitted entirely if not needed
+- <file> selector is automatically prepended with 'test_' if required
+- '.' separators are required only if selector(s) follow(s)
+
+examples:
+
+1. all of the following expressions are equivalent and will select
+ all test classes and functions from test_bfd.py:
+ 'test_bfd' 'bfd' 'test_bfd..' 'bfd.' 'bfd.*.*' 'test_bfd.*.*'
+2. 'bfd.BFDAPITestCase' selects all tests from test_bfd.py,
+ which are part of BFDAPITestCase class
+3. 'bfd.BFDAPITestCase.test_add_bfd' selects a single test named
+ test_add_bfd from test_bfd.py/BFDAPITestCase
+4. '.*.test_add_bfd' selects all test functions named test_add_bfd
+ from all files/classes
+"""
+parser.add_argument("--filter", action="store",
+ metavar="FILTER_EXPRESSION", help=filter_help_string)
+
+default_retries = 0
+
+parser.add_argument("--retries", action="store", default=default_retries,
+ type=positive_int_or_default(default_retries),
+ help="retry failed tests RETRIES times")
+
+parser.add_argument("--step", action="store_true", default=False,
+ help="enable stepping through tests")
+
+debug_help_string = """\
+attach - attach to already running vpp
+core - detect coredump and load core in gdb on crash
+gdb - print VPP PID and pause allowing attaching gdb
+gdbserver - same as above, but run gdb in gdbserver
+"""
+
+parser.add_argument("--debug", action="store",
+ choices=["attach", "core", "gdb", "gdbserver"],
+ help=debug_help_string)
+
+parser.add_argument("--debug-framework", action="store_true",
+ help="enable internal test framework debugging")
+
+parser.add_argument("--compress-core", action="store_true",
+ help="compress core files if not debugging them")
+
+parser.add_argument("--extended", action="store_true",
+ help="run extended tests")
+
+parser.add_argument("--sanity", action="store_true",
+ help="perform sanity vpp run before running tests")
+
+parser.add_argument("--force-foreground", action="store_true",
+ help="force running in foreground - don't fork")
+
+parser.add_argument("--jobs", action="store", type=positive_int_or_auto,
+ default="auto", help="maximum concurrent test jobs")
+
+parser.add_argument("--venv-dir", action="store",
+ type=directory, help="path to virtual environment")
+
+default_rnd_seed = time.time()
+parser.add_argument("--rnd-seed", action="store", default=default_rnd_seed,
+ type=positive_int_or_default(default_rnd_seed),
+ help="random generator seed (default: current time)")
+
+parser.add_argument("--vpp-worker-count", action="store", type=worker_config,
+ default=0, help="number of vpp workers")
+
+parser.add_argument("--gcov", action="store_true",
+ default=False, help="running gcov tests")
+
+parser.add_argument("--cache-vpp-output", action="store_true", default=False,
+ help="cache VPP stdout/stderr and log as one block "
+ "after test finishes")
+
+parser.add_argument("--vpp-ws-dir", action="store", required=True,
+ type=directory, help="vpp workspace directory")
+
+parser.add_argument("--vpp-tag", action="store", default="vpp_debug",
+ metavar="VPP_TAG", required=True,
+ help="vpp tag (e.g. vpp, vpp_debug, vpp_gcov)")
+
+parser.add_argument("--vpp", action="store", help="path to vpp binary "
+ "(default: derive from VPP_WS_DIR and VPP_TAG)")
+
+parser.add_argument("--vpp-install-dir", type=directory,
+ action="store", help="path to vpp install directory"
+ "(default: derive from VPP_WS_DIR and VPP_TAG)")
+
+parser.add_argument("--vpp-build-dir", action="store", type=directory,
+ help="vpp build directory"
+ "(default: derive from VPP_WS_DIR and VPP_TAG)")
+
+parser.add_argument("--vpp-plugin-dir", action="append", type=directory,
+ help="directory containing vpp plugins"
+ "(default: derive from VPP_WS_DIR and VPP_TAG)")
+
+parser.add_argument("--vpp-test-plugin-dir", action="append", type=directory,
+ help="directory containing vpp api test plugins"
+ "(default: derive from VPP_WS_DIR and VPP_TAG)")
+
+parser.add_argument("--extern-plugin-dir", action="append", type=directory,
+ default=[], help="directory containing external plugins")
+
+parser.add_argument("--coredump-size", action="store", default="unlimited",
+ help="specify vpp coredump size")
+
+parser.add_argument("--max-vpp-cpus", action="store", type=int_or_auto,
+ default=0, help="max cpus used by vpp")
+
+variant_help_string = """\
+specify which march node variant to unit test
+ e.g. --variant=skx - test the skx march variants
+ e.g. --variant=icl - test the icl march variants
+"""
+
+parser.add_argument("--variant", action="store", help=variant_help_string)
+
+parser.add_argument("--api-fuzz", action="store", default=None,
+ help="specify api fuzzing parameters")
+
+parser.add_argument("--wipe-tmp-dir", action="store_true", default=True,
+ help="remove test tmp directory before running test")
+
+parser.add_argument("--tmp-dir", action="store", default="/tmp",
+ type=directory_verify_or_create,
+ help="directory where to store test temporary directories")
+
+parser.add_argument("--log-dir", action="store",
+ type=directory_verify_or_create,
+ help="directory where to store directories "
+ "containing log files (default: --tmp-dir)")
+
+default_keep_pcaps = False
+parser.add_argument("--keep-pcaps", action="store_true",
+ default=default_keep_pcaps,
+ help="if set, keep all pcap files from a test run"
+ f" (default: {default_keep_pcaps})")
+
+config = parser.parse_args()
+
+ws = config.vpp_ws_dir
+br = f"{ws}/build-root"
+tag = config.vpp_tag
+
+if config.vpp_install_dir is None:
+ config.vpp_install_dir = f"{br}/install-{tag}-native"
+
+if config.vpp is None:
+ config.vpp = f"{config.vpp_install_dir}/vpp/bin/vpp"
+
+if config.vpp_build_dir is None:
+ config.vpp_build_dir = f"{br}/build-{tag}-native"
+
+libs = ["lib", "lib64"]
+
+if config.vpp_plugin_dir is None:
+ config.vpp_plugin_dir = [
+ f"{config.vpp_install_dir}/vpp/{lib}/vpp_plugins" for lib in libs]
+
+if config.vpp_test_plugin_dir is None:
+ config.vpp_test_plugin_dir = [
+ f"{config.vpp_install_dir}/vpp/{lib}/vpp_api_test_plugins"
+ for lib in libs]
+
+test_dirs = [f"{ws}/test"]
+
+if config.test_src_dir is not None:
+ test_dirs.extend(config.test_src_dir)
+
+config.test_src_dir = test_dirs
+
+
+if config.venv_dir is None:
+ config.venv_dir = f"{ws}/test/venv"
+
+available_cpus = psutil.Process().cpu_affinity()
+num_cpus = len(available_cpus)
+
+if config.max_vpp_cpus == 'auto':
+ max_vpp_cpus = num_cpus
+elif config.max_vpp_cpus > 0:
+ max_vpp_cpus = min(config.max_vpp_cpus, num_cpus)
+else:
+ max_vpp_cpus = num_cpus
+
+if __name__ == "__main__":
+ print("Provided arguments:")
+ for i in config.__dict__:
+ print(f" {i} is {config.__dict__[i]}")
+++ /dev/null
-import os
-import psutil
-
-available_cpus = psutil.Process().cpu_affinity()
-num_cpus = len(available_cpus)
-
-max_vpp_cpus = os.getenv("MAX_VPP_CPUS", "auto").lower()
-
-if max_vpp_cpus == "auto":
- max_vpp_cpus = num_cpus
-else:
- try:
- max_vpp_cpus = int(max_vpp_cpus)
- except ValueError as e:
- raise ValueError("Invalid MAX_VPP_CPUS value specified, valid "
- "values are a positive integer or 'auto'") from e
- if max_vpp_cpus <= 0:
- raise ValueError("Invalid MAX_VPP_CPUS value specified, valid "
- "values are a positive integer or 'auto'")
- if max_vpp_cpus > num_cpus:
- max_vpp_cpus = num_cpus
from sanity_run_vpp import SanityTestCase
from shutil import rmtree
-from cpu_config import available_cpus
+from config import available_cpus
gdb_path = '/usr/bin/gdb'
# but any test case class could be used ...
SanityTestCase.set_debug_flags("attach")
SanityTestCase.tempdir = SanityTestCase.get_tempdir()
- if os.path.exists(SanityTestCase.tempdir):
- if os.getenv("VPP_IN_GDB_NO_RMDIR", "0") in ["1", "y", "yes"]:
- raise FileExistsError(
- "Temporary directory exists and removal denied.")
- print("Removing existing temp dir '%s'." % SanityTestCase.tempdir)
- rmtree(SanityTestCase.tempdir)
- print("Creating temp dir '%s'." % SanityTestCase.tempdir)
- os.mkdir(SanityTestCase.tempdir)
SanityTestCase.assign_cpus(
available_cpus[:SanityTestCase.get_cpus_required()])
SanityTestCase.setUpConstants()
vpp_cmdline = SanityTestCase.vpp_cmdline
- if os.getenv("VPP_IN_GDB_CMDLINE", "y").lower() in ["1", "y", "yes"]:
- print("Hacking cmdline to make VPP interactive.")
- vpp_cmdline.insert(vpp_cmdline.index("nodaemon"), "interactive")
+ print("Hacking cmdline to make VPP interactive.")
+ vpp_cmdline.insert(vpp_cmdline.index("nodaemon"), "interactive")
print("VPP cmdline is %s" % " ".join(vpp_cmdline))
print("Running GDB.")
def print_callback(file_name, cls, method):
print("%s.%s.%s" % (file_name, cls.__name__, method))
-
-
-if __name__ == '__main__':
- parser = argparse.ArgumentParser(description="Discover VPP unit tests")
- parser.add_argument("-d", "--dir", action='append', type=str,
- help="directory containing test files "
- "(may be specified multiple times)")
- args = parser.parse_args()
- if args.dir is None:
- args.dir = "."
-
- ignore_path = os.getenv("VENV_PATH", "")
- suite = unittest.TestSuite()
- for d in args.dir:
- discover_tests(d, print_callback, ignore_path)
import signal
import subprocess
import unittest
-import tempfile
+import re
import time
import faulthandler
import random
import copy
-import psutil
import platform
+import shutil
from collections import deque
from threading import Thread, Event
from inspect import getdoc, isclass
import scapy.compat
from scapy.packet import Raw, Packet
+from config import config, available_cpus, num_cpus, max_vpp_cpus
import hook as hookmodule
from vpp_pg_interface import VppPGInterface
from vpp_sub_interface import VppSubInterface
from scapy.layers.inet6 import ICMPv6DestUnreach, ICMPv6EchoRequest
from scapy.layers.inet6 import ICMPv6EchoReply
-from cpu_config import available_cpus, num_cpus, max_vpp_cpus
logger = logging.getLogger(__name__)
SKIP_CPU_SHORTAGE = 5
-class BoolEnvironmentVariable(object):
-
- def __init__(self, env_var_name, default='n', true_values=None):
- self.name = env_var_name
- self.default = default
- self.true_values = true_values if true_values is not None else \
- ("y", "yes", "1")
-
- def __bool__(self):
- return os.getenv(self.name, self.default).lower() in self.true_values
-
- if sys.version_info[0] == 2:
- __nonzero__ = __bool__
-
- def __repr__(self):
- return 'BoolEnvironmentVariable(%r, default=%r, true_values=%r)' % \
- (self.name, self.default, self.true_values)
-
-
-debug_framework = BoolEnvironmentVariable('TEST_DEBUG')
-if debug_framework:
+if config.debug_framework:
import debug_internal
"""
limit = -1
stdout_fragment = split[-1]
testclass.vpp_stdout_deque.extend(split[:limit])
- if not testclass.cache_vpp_output:
+ if not config.cache_vpp_output:
for line in split[:limit]:
testclass.logger.info(
"VPP STDOUT: %s" % line.rstrip("\n"))
stderr_fragment = split[-1]
testclass.vpp_stderr_deque.extend(split[:limit])
- if not testclass.cache_vpp_output:
+ if not config.cache_vpp_output:
for line in split[:limit]:
testclass.logger.error(
"VPP STDERR: %s" % line.rstrip("\n"))
# flag will take care of properly terminating the loop
-def _is_skip_aarch64_set():
- return BoolEnvironmentVariable('SKIP_AARCH64')
-
-
-is_skip_aarch64_set = _is_skip_aarch64_set()
-
-
def _is_platform_aarch64():
return platform.machine() == 'aarch64'
is_platform_aarch64 = _is_platform_aarch64()
-def _running_extended_tests():
- return BoolEnvironmentVariable("EXTENDED_TESTS")
-
-
-running_extended_tests = _running_extended_tests()
-
-
-def _running_gcov_tests():
- return BoolEnvironmentVariable("GCOV_TESTS")
-
-
-running_gcov_tests = _running_gcov_tests()
-
-
-def get_environ_vpp_worker_count():
- worker_config = os.getenv("VPP_WORKER_CONFIG", None)
- if worker_config:
- elems = worker_config.split(" ")
- if elems[0] != "workers" or len(elems) != 2:
- raise ValueError("Wrong VPP_WORKER_CONFIG == '%s' value." %
- worker_config)
- return int(elems[1])
- else:
- return 0
-
-
-environ_vpp_worker_count = get_environ_vpp_worker_count()
-
-
class KeepAliveReporter(object):
"""
Singleton object which reports test start to parent process
else:
desc = test.id()
- self.pipe.send((desc, test.vpp_bin, test.tempdir, test.vpp.pid))
+ self.pipe.send((desc, config.vpp, test.tempdir, test.vpp.pid))
class TestCaseTag(Enum):
if cls.has_tag(TestCaseTag.FIXME_VPP_WORKERS):
cls.vpp_worker_count = 0
else:
- cls.vpp_worker_count = environ_vpp_worker_count
+ cls.vpp_worker_count = config.vpp_worker_count
return cls.vpp_worker_count
@classmethod
@classmethod
def setUpConstants(cls):
""" Set-up the test case class based on environment variables """
- cls.step = BoolEnvironmentVariable('STEP')
- # inverted case to handle '' == True
- c = os.getenv("CACHE_OUTPUT", "1")
- cls.cache_vpp_output = False if c.lower() in ("n", "no", "0") else True
- cls.vpp_bin = os.getenv('VPP_BIN', "vpp")
- extern_plugin_path = os.getenv('EXTERN_PLUGINS')
+ cls.step = config.step
+ cls.plugin_path = ":".join(config.vpp_plugin_dir)
+ cls.test_plugin_path = ":".join(config.vpp_test_plugin_dir)
+ cls.extern_plugin_path = ":".join(config.extern_plugin_dir)
debug_cli = ""
if cls.step or cls.debug_gdb or cls.debug_gdbserver:
debug_cli = "cli-listen localhost:5002"
- coredump_size = None
- size = os.getenv("COREDUMP_SIZE")
- if size is not None:
- coredump_size = "coredump-size %s" % size
- if coredump_size is None:
+ size = re.search(r"\d+[gG]", config.coredump_size)
+ if size:
+ coredump_size = f"coredump-size {config.coredump_size}".lower()
+ else:
coredump_size = "coredump-size unlimited"
-
- default_variant = os.getenv("VARIANT")
+ default_variant = config.variant
if default_variant is not None:
default_variant = "defaults { %s 100 }" % default_variant
else:
default_variant = ""
- api_fuzzing = os.getenv("API_FUZZ")
+ api_fuzzing = config.api_fuzz
if api_fuzzing is None:
api_fuzzing = 'off'
cls.vpp_cmdline = [
- cls.vpp_bin,
+ config.vpp,
"unix", "{", "nodaemon", debug_cli, "full-coredump",
coredump_size, "runtime-dir", cls.tempdir, "}",
"api-trace", "{", "on", "}",
"api-segment", "{", "prefix", cls.get_api_segment_prefix(), "}",
"cpu", "{", "main-core", str(cls.cpus[0]), ]
- if extern_plugin_path is not None:
+ if cls.extern_plugin_path not in (None, ""):
cls.extra_vpp_plugin_config.append(
- "add-path %s" % extern_plugin_path)
+ "add-path %s" % cls.extern_plugin_path)
if cls.get_vpp_worker_count():
cls.vpp_cmdline.extend([
"corelist-workers", ",".join([str(x) for x in cls.cpus[1:]])])
print(single_line_delim)
print("You can debug VPP using:")
if cls.debug_gdbserver:
- print("sudo gdb " + cls.vpp_bin +
- " -ex 'target remote localhost:{port}'"
- .format(port=cls.gdbserver_port))
+ print(f"sudo gdb {config.vpp} "
+ f"-ex 'target remote localhost:{cls.gdbserver_port}'")
print("Now is the time to attach gdb by running the above "
"command, set up breakpoints etc., then resume VPP from "
"within gdb by issuing the 'continue' command")
cls.gdbserver_port += 1
elif cls.debug_gdb:
- print("sudo gdb " + cls.vpp_bin + " -ex 'attach %s'" % cls.vpp.pid)
+ print(f"sudo gdb {config.vpp} -ex 'attach {cls.vpp.pid}'")
print("Now is the time to attach gdb by running the above "
"command and set up breakpoints etc., then resume VPP from"
" within gdb by issuing the 'continue' command")
@classmethod
def get_tempdir(cls):
- if cls.debug_attach:
- return os.getenv("VPP_IN_GDB_TMP_DIR",
- "/tmp/unittest-attach-gdb")
- else:
- return tempfile.mkdtemp(prefix='vpp-unittest-%s-' % cls.__name__)
+ tmpdir = f"{config.tmp_dir}/vpp-unittest-{cls.__name__}"
+ if config.wipe_tmp_dir:
+ shutil.rmtree(tmpdir, ignore_errors=True)
+ os.mkdir(tmpdir)
+ return tmpdir
+
+ @classmethod
+ def create_file_handler(cls):
+ if config.log_dir is None:
+ cls.file_handler = FileHandler(f"{cls.tempdir}/log.txt")
+ return
+
+ logdir = f"{config.log_dir}/vpp-unittest-{cls.__name__}"
+ if config.wipe_tmp_dir:
+ shutil.rmtree(logdir, ignore_errors=True)
+ os.mkdir(logdir)
+ cls.file_handler = FileHandler(f"{logdir}/log.txt")
@classmethod
def setUpClass(cls):
"""
super(VppTestCase, cls).setUpClass()
cls.logger = get_logger(cls.__name__)
- seed = os.environ["RND_SEED"]
- random.seed(seed)
+ random.seed(config.rnd_seed)
if hasattr(cls, 'parallel_handler'):
cls.logger.addHandler(cls.parallel_handler)
cls.logger.propagate = False
- d = os.getenv("DEBUG", None)
- cls.set_debug_flags(d)
+ cls.set_debug_flags(config.debug)
cls.tempdir = cls.get_tempdir()
- cls.file_handler = FileHandler("%s/log.txt" % cls.tempdir)
+ cls.create_file_handler()
cls.file_handler.setFormatter(
Formatter(fmt='%(asctime)s,%(msecs)03d %(message)s',
datefmt="%H:%M:%S"))
os.chdir(cls.tempdir)
cls.logger.info("Temporary dir is %s, api socket is %s",
cls.tempdir, cls.get_api_sock_path())
- cls.logger.debug("Random seed is %s", seed)
+ cls.logger.debug("Random seed is %s", config.rnd_seed)
cls.setUpConstants()
cls.reset_packet_infos()
cls._pcaps = []
cls.run_vpp()
cls.reporter.send_keep_alive(cls, 'setUpClass')
VppTestResult.current_test_case_info = TestCaseInfo(
- cls.logger, cls.tempdir, cls.vpp.pid, cls.vpp_bin)
+ cls.logger, cls.tempdir, cls.vpp.pid, config.vpp)
cls.vpp_stdout_deque = deque()
cls.vpp_stderr_deque = deque()
if not cls.debug_attach:
cls.quit()
cls.file_handler.close()
cls.reset_packet_infos()
- if debug_framework:
+ if config.debug_framework:
debug_internal.on_tear_down_class(cls)
def show_commands_at_teardown(self):
def pg_start(cls, trace=True):
""" Enable the PG, wait till it is done, then clean up """
for (intf, worker) in cls._old_pcaps:
- intf.rename_old_pcap_file(intf.get_in_path(worker),
+ intf.handle_old_pcap_file(intf.get_in_path(worker),
intf.in_history_counter)
cls._old_pcaps = []
if trace:
def symlink_failed(self):
if self.current_test_case_info:
try:
- failed_dir = os.getenv('FAILED_DIR')
+ failed_dir = config.failed_dir
link_path = os.path.join(
failed_dir,
'%s-FAILED' %
import os
import logging
+from config import config
+
""" @var formatting delimiter consisting of '=' characters """
double_line_delim = '=' * 78
""" @var formatting delimiter consisting of '-' characters """
if hasattr(record, 'color'):
message = colorize(message, record.color)
return message
-try:
- verbose = int(os.getenv("V", 0))
-except:
- verbose = 0
+
# 40 = ERROR, 30 = WARNING, 20 = INFO, 10 = DEBUG, 0 = NOTSET (all messages)
-if verbose >= 2:
+if config.verbose >= 2:
log_level = 10
-elif verbose == 1:
+elif config.verbose == 1:
log_level = 20
else:
log_level = 40
logger.addHandler(handler)
return logger
+
# Static variables to store color formatting strings.
#
# These variables (RED, GREEN, YELLOW and LPURPLE) are used to configure
import argparse
import time
import threading
+import traceback
import signal
import re
from multiprocessing import Process, Pipe, get_context
from multiprocessing.queues import Queue
from multiprocessing.managers import BaseManager
import framework
-from framework import VppTestRunner, running_extended_tests, VppTestCase, \
+from config import config, num_cpus, available_cpus, max_vpp_cpus
+from framework import VppTestRunner, VppTestCase, \
get_testcase_doc_name, get_test_description, PASS, FAIL, ERROR, SKIP, \
TEST_RUN, SKIP_CPU_SHORTAGE
from debug import spawn_gdb, start_vpp_in_gdb
from log import get_parallel_logger, double_line_delim, RED, YELLOW, GREEN, \
colorize, single_line_delim
from discover_tests import discover_tests
+import sanity_run_vpp
from subprocess import check_output, CalledProcessError
from util import check_core_path, get_core_path, is_core_present
-from cpu_config import num_cpus, max_vpp_cpus, available_cpus
# timeout which controls how long the child has to finish after seeing
# a core dump in test temporary directory. If this is exceeded, parent assumes
VppTestCase.parallel_handler = logger.handlers[0]
result = VppTestRunner(keep_alive_pipe=keep_alive_pipe,
descriptions=descriptions,
- verbosity=verbose,
+ verbosity=config.verbose,
result_pipe=result_pipe,
- failfast=failfast,
+ failfast=config.failfast,
print_summary=False).run(suite)
finished_pipe.send(result.wasSuccessful())
finished_pipe.close()
if last_test_temp_dir:
# Need to create link in case of a timeout or core dump without failure
lttd = os.path.basename(last_test_temp_dir)
- failed_dir = os.getenv('FAILED_DIR')
- link_path = '%s%s-FAILED' % (failed_dir, lttd)
+ link_path = '%s%s-FAILED' % (config.failed_dir, lttd)
if not os.path.exists(link_path):
os.symlink(last_test_temp_dir, link_path)
logger.error("Symlink to failed testcase directory: %s -> %s"
except Exception as e:
logger.exception("Unexpected error running `file' utility "
"on core-file")
- logger.error("gdb %s %s" %
- (os.getenv('VPP_BIN', 'vpp'), core_path))
+ logger.error(f"gdb {config.vpp_bin} {core_path}")
if vpp_pid:
# Copy api post mortem
print(single_line_delim)
spawn_gdb(vpp_binary, get_core_path(tempdir))
print(single_line_delim)
- elif compress_core:
+ elif config.compress_core:
print("Compressing core-file in test directory `%s'" % tempdir)
os.system("gzip %s" % get_core_path(tempdir))
results.append(wrapped_testcase_suite.result)
finished_testcase_suites.add(wrapped_testcase_suite)
stop_run = False
- if failfast and not wrapped_testcase_suite.was_successful():
+ if config.failfast and not wrapped_testcase_suite.was_successful():
stop_run = True
if not wrapped_testcase_suite.was_successful():
continue
fail = False
- if wrapped_testcase_suite.last_heard + test_timeout < \
+ if wrapped_testcase_suite.last_heard + config.timeout < \
time.time():
fail = True
wrapped_testcase_suite.logger.critical(
raise
finally:
read_from_testcases.clear()
- stdouterr_thread.join(test_timeout)
+ stdouterr_thread.join(config.timeout)
manager.shutdown()
handle_cores(failed_wrapped_testcases)
self.filtered.addTest(test_method)
-test_option = "TEST"
-
-
-def parse_test_option():
- f = os.getenv(test_option, None)
+def parse_test_filter(test_filter):
+ f = test_filter
filter_file_name = None
filter_class_name = None
filter_func_name = None
return return_code, results_per_suite.rerun
-def parse_digit_env(env_var, default):
- value = os.getenv(env_var, default)
- if value != default:
- if value.isdigit():
- value = int(value)
- else:
- print('WARNING: unsupported value "%s" for env var "%s",'
- 'defaulting to %s' % (value, env_var, default))
- value = default
- return value
-
-
if __name__ == '__main__':
- verbose = parse_digit_env("V", 0)
+ print(f"Config is: {config}")
- test_timeout = parse_digit_env("TIMEOUT", 600) # default = 10 minutes
+ if config.sanity:
+ print("Running sanity test case.")
+ try:
+ rc = sanity_run_vpp.main()
+ if rc != 0:
+ sys.exit(rc)
+ except Exception as e:
+ print(traceback.format_exc())
+ print("Couldn't run sanity test case.")
+ sys.exit(-1)
test_finished_join_timeout = 15
- retries = parse_digit_env("RETRIES", 0)
+ debug_gdb = config.debug in ["gdb", "gdbserver", "attach"]
+ debug_core = config.debug == "core"
- debug = os.getenv("DEBUG", "n").lower() in ["gdb", "gdbserver", "attach"]
-
- debug_core = os.getenv("DEBUG", "").lower() == "core"
- compress_core = framework.BoolEnvironmentVariable("CORE_COMPRESS")
-
- if os.getenv("VPP_IN_GDB", "n").lower() in ["1", "y", "yes"]:
- start_vpp_in_gdb()
- exit()
-
- step = framework.BoolEnvironmentVariable("STEP")
- force_foreground = framework.BoolEnvironmentVariable("FORCE_FOREGROUND")
-
- run_interactive = debug or step or force_foreground
+ run_interactive = debug_gdb or config.step or config.force_foreground
max_concurrent_tests = 0
print(f"OS reports {num_cpus} available cpu(s).")
- test_jobs = os.getenv("TEST_JOBS", "1").lower() # default = 1 process
+ test_jobs = config.jobs
if test_jobs == 'auto':
if run_interactive:
max_concurrent_tests = 1
print(f"Running at most {max_concurrent_tests} python test "
"processes concurrently.")
else:
- try:
- test_jobs = int(test_jobs)
- except ValueError as e:
- raise ValueError("Invalid TEST_JOBS value specified, valid "
- "values are a positive integer or 'auto'") from e
- if test_jobs <= 0:
- raise ValueError("Invalid TEST_JOBS value specified, valid "
- "values are a positive integer or 'auto'")
- max_concurrent_tests = int(test_jobs)
+ max_concurrent_tests = test_jobs
print(f"Running at most {max_concurrent_tests} python test processes "
"concurrently as set by 'TEST_JOBS'.")
'STEP is set) in parallel (TEST_JOBS is more than 1) is not '
'supported')
- parser = argparse.ArgumentParser(description="VPP unit tests")
- parser.add_argument("-f", "--failfast", action='store_true',
- help="fast failure flag")
- parser.add_argument("-d", "--dir", action='append', type=str,
- help="directory containing test files "
- "(may be specified multiple times)")
- args = parser.parse_args()
- failfast = args.failfast
descriptions = True
print("Running tests using custom test runner.")
- filter_file, filter_class, filter_func = parse_test_option()
+ filter_file, filter_class, filter_func = \
+ parse_test_filter(config.filter)
- print("Active filters: file=%s, class=%s, function=%s" % (
+ print("Selected filters: file=%s, class=%s, function=%s" % (
filter_file, filter_class, filter_func))
filter_cb = FilterByTestOption(filter_file, filter_class, filter_func)
- ignore_path = os.getenv("VENV_PATH", None)
+ ignore_path = config.venv_dir
cb = SplitToSuitesCallback(filter_cb)
- for d in args.dir:
+ for d in config.test_src_dir:
print("Adding tests from directory tree %s" % d)
discover_tests(d, cb, ignore_path)
print("%s out of %s tests match specified filters" % (
tests_amount, tests_amount + cb.filtered.countTestCases()))
- if not running_extended_tests:
+ if not config.extended:
print("Not running extended tests (some tests will be skipped)")
- attempts = retries + 1
+ attempts = config.retries + 1
if attempts > 1:
print("Perform %s attempts to pass the suite..." % attempts)
suite.assign_cpus([])
cpu_shortage = True
full_suite.addTests(suites)
- result = VppTestRunner(verbosity=verbose,
- failfast=failfast,
+ result = VppTestRunner(verbosity=config.verbose,
+ failfast=config.failfast,
print_summary=True).run(full_suite)
was_successful = result.wasSuccessful()
if not was_successful:
pass
-if __name__ == '__main__':
+def main():
rc = 0
tc = SanityTestCase
x, y = Pipe()
print('Sanity test case passed.')
else:
print('Sanity test case failed.')
+ return rc
+
- sys.exit(rc)
+if __name__ == '__main__':
+ sys.exit(main())
--- /dev/null
+#!/bin/bash
+
+ff="0"
+items=
+for i in "$@"
+do
+case $i in
+ --venv-dir=*)
+ venv_dir="${i#*=}"
+ if [ -d $venv_dir ]
+ then
+ venv_dir=$(cd $venv_dir; pwd)
+ else
+ echo "ERROR: '$venv_dir' is not a directory"
+ exit 1
+ fi
+ items="$items --venv-dir=\"$venv_dir\""
+ ;;
+ --vpp-ws-dir=*)
+ ws_dir="${i#*=}"
+ if [ -d $ws_dir ]
+ then
+ ws_dir=$(cd $ws_dir; pwd)
+ else
+ echo "ERROR: '$ws_dir' is not a directory"
+ exit 1
+ fi
+ items="$items --vpp-ws-dir=\"$ws_dir\""
+ ;;
+ --force-foreground)
+ ff="1"
+ items="$items \"$i\""
+ ;;
+ --vpp-tag=*)
+ tag="${i#*=}"
+ items="$items \"$i\""
+ ;;
+ --python-opts=*)
+ python_opts="${i#*=}"
+ ;;
+ *)
+ # unknown option - skip
+ items="$items \"$i\""
+ ;;
+esac
+done
+
+extra_args=""
+if [ -z "$ws_dir" ]
+then
+ ws_dir=$(pwd)
+ echo "Argument --vpp-ws-dir not specified, defaulting to '$ws_dir'"
+ extra_args="$extra_args --vpp-ws-dir=$ws_dir"
+fi
+
+if [ -z "$venv_dir" ]
+then
+ venv_dir="$ws_dir/test/venv"
+ echo "Argument --venv-path not specified, defaulting to '$venv_dir'"
+ extra_args="$extra_args --venv-dir=$venv_dir"
+fi
+
+if [ -z "$tag" ]
+then
+ tag="vpp_debug"
+ echo "Argument --vpp-tag not specified, defaulting to '$tag'"
+ extra_args="$extra_args --vpp-tag=$tag"
+fi
+
+eval set -- $items
+$ws_dir/test/scripts/setsid_wrapper.sh $ws_dir/test/scripts/run_in_venv_with_cleanup.sh $ff $venv_dir/bin/activate python3 $python_opts $ws_dir/test/run_tests.py $extra_args $*
#!/bin/bash
-if [[ "$1" == "1" ]]
+cmd=$1
+force_foreground=$2
+shift
+shift
+
+if [[ "$force_foreground" == "1" ]]
then
- setsid scripts/run_in_venv_with_cleanup.sh $*
+ setsid $cmd $force_foreground $*
else
- setsid scripts/run_in_venv_with_cleanup.sh $* &
+ setsid $cmd $force_foreground $* &
pid=$!
trap "echo setsid_wrapper.sh: got signal, killing child pid ${pid}; kill ${pid}; sleep .1;" SIGINT SIGTERM
wait ${pid}
""" ACL plugin extended stateful tests """
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, TCP
return new_rule
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class ACLPluginConnTestCase(VppTestCase):
""" ACL plugin connection-oriented extended testcases """
from scapy.layers.inet import IP, UDP
from scapy.layers.inet6 import IPv6
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from vpp_lo_interface import VppLoInterface
from vpp_l2 import L2_PORT_TYPE
from vpp_sub_interface import L2_VTR_OP, VppSubInterface, VppDot1QSubint, \
import unittest
-from framework import VppTestCase, VppTestRunner, running_gcov_tests
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
from scapy.layers.l2 import Ether, GRE
from scapy.packet import Raw
+from config import config
from bfd import VppBFDAuthKey, BFD, BFDAuthType, VppBFDUDPSession, \
BFDDiagCode, BFDState, BFD_vpp_echo
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from util import ppp
from vpp_ip import DpoProto
"Poll bit set in BFD packet")
# returning inconsistent results requiring retries in per-patch tests
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_poll_response(self):
""" test correct response to control frame with poll bit set """
bfd_session_up(self)
packet[IPv6].dst)
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class BFDTunTestCase(VppTestCase):
""" BFD over GRE tunnel """
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, \
VppMplsTable, VppIpMRoute, VppMRoutePath, VppIpTable, \
for nhr in nh_routes:
nhr.remove_vpp_config()
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_midpoint_1024(self):
"""BIER midpoint BSL:1024"""
self.bier_midpoint(BIERLength.BIER_LEN_1024, 128, 1024)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_midpoint_512(self):
"""BIER midpoint BSL:512"""
self.bier_midpoint(BIERLength.BIER_LEN_512, 64, 512)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_midpoint_256(self):
"""BIER midpoint BSL:256"""
self.bier_midpoint(BIERLength.BIER_LEN_256, 32, 256)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_midpoint_128(self):
"""BIER midpoint BSL:128"""
self.bier_midpoint(BIERLength.BIER_LEN_128, 16, 128)
self.assertEqual(rx[0][IP].src, "1.1.1.1")
self.assertEqual(rx[0][IP].dst, "232.1.1.2")
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_e2e_1024(self):
""" BIER end-to-end BSL:1024"""
self.bier_e2e(BIERLength.BIER_LEN_1024, 128, 1024)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_e2e_512(self):
""" BIER end-to-end BSL:512"""
self.bier_e2e(BIERLength.BIER_LEN_512, 64, 512)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_e2e_256(self):
""" BIER end-to-end BSL:256"""
self.bier_e2e(BIERLength.BIER_LEN_256, 32, 256)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_bier_e2e_128(self):
""" BIER end-to-end BSL:128"""
self.bier_e2e(BIERLength.BIER_LEN_128, 16, 128)
import unittest
-from framework import VppTestCase, VppTestRunner, running_gcov_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
self.logger.critical(error)
self.assertNotIn('failed', error)
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
+ @unittest.skipUnless(config.gcov, "part of code coverage tests")
def test_bihash_coverage(self):
""" Improve Code Coverage """
""" Container integration tests """
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, TCP
pass
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class ContainerIntegrationTestCase(VppTestCase):
""" Container integration extended testcases """
import struct
import unittest
import scapy.compat
-from framework import VppTestCase, running_extended_tests
+from time import sleep
+from config import config
+from framework import VppTestCase
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, UDPerror
self.assertEqual(0, dms[0].ses_num)
# TODO: ipfix needs to be separated from NAT base plugin
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_session_limit_per_user(self):
""" Deterministic NAT maximum sessions per user limit """
self.vapi.det44_add_del_map(is_add=1, in_addr=self.pg0.remote_ip4,
import struct
import six
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from vpp_neighbor import VppNeighbor
from vpp_ip_route import find_route, VppIpTable
import random
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
import scapy.compat
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IP, TCP, UDP
from scapy.layers.inet6 import IPv6
+from config import config
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from vpp_object import VppObject
from vpp_pg_interface import CaptureTimeoutError
self.logger.info("FFP_TEST_FINISH_0002")
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class DisableIPFIX(MethodHolder):
"""Disable IPFIX"""
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class ReenableIPFIX(MethodHolder):
"""Re-enable IPFIX"""
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class DisableFP(MethodHolder):
"""Disable Flowprobe feature"""
self.logger.info("FFP_TEST_FINISH_0001")
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class ReenableFP(MethodHolder):
"""Re-enable Flowprobe feature"""
from scapy.contrib.igmpv3 import IGMPv3, IGMPv3gr, IGMPv3mq, IGMPv3mr
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from vpp_igmp import find_igmp_state, IGMP_FILTER, IgmpRecord, IGMP_MODE, \
IgmpSG, VppHostState, wait_for_igmp_event
from vpp_ip_route import find_mroute, VppIpTable
)
from ipaddress import IPv4Address, IPv6Address, ip_address
import unittest
+from config import config
from scapy.layers.ipsec import ESP
from scapy.layers.inet import IP, UDP, Ether
from scapy.layers.inet6 import IPv6
if 'auth' in params and params['auth'] == 'rsa-sig':
auth_method = 'rsa-sig'
- work_dir = os.getenv('BR') + '/../src/plugins/ikev2/test/certs/'
+ work_dir = f"{config.vpp_ws_dir}/src/plugins/ikev2/test/certs/"
self.vapi.ikev2_set_local_key(
key_file=work_dir + params['server-key'])
import time
import re
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase
from vpp_object import VppObject
from vpp_pg_interface import CaptureTimeoutError
from vpp_ip_route import VppIpRoute, VppRoutePath
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from util import Host, ppp
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from util import Host, ppp
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from util import Host, ppp
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from util import Host, ppp
import unittest
-from framework import VppTestCase, VppTestRunner, running_gcov_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
self.logger.critical(error)
self.assertNotIn('FAILED', error)
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
+ @unittest.skipUnless(config.gcov, "part of code coverage tests")
def test_mactime_unittest(self):
""" Mactime Plugin Code Coverage Test """
cmds = ["loopback create",
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, ICMP
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from framework import tag_run_solo
from remote_test import RemoteClass, RemoteVppTestCase
from vpp_memif import remove_all_memif_vpp_config, \
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
import os
from io import BytesIO
import scapy.compat
+from config import config
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.data import IP_PROTOS
from scapy.layers.inet import IP, TCP, UDP, ICMP
addresses = self.vapi.nat64_pool_addr_dump()
self.assertEqual(0, len(addresses))
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_ipfix_max_bibs_sessions(self):
""" IPFIX logging maximum session and BIB entries exceeded """
max_bibs = 1280
from io import BytesIO
import scapy.compat
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
import unittest
-from framework import VppTestCase, VppTestRunner, running_gcov_tests
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
import os
import unittest
-from framework import VppTestCase, VppTestRunner, running_gcov_tests
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
import os
import subprocess
import signal
+from config import config
from framework import tag_fixme_vpp_workers
-from framework import VppTestCase, VppTestRunner, running_extended_tests, \
- Worker
+from framework import VppTestCase, VppTestRunner, Worker
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
""" QUIC Test Application Worker """
process = None
- def __init__(self, build_dir, appname, executable_args, logger, role,
+ def __init__(self, appname, executable_args, logger, role,
testcase, env=None, *args, **kwargs):
if env is None:
env = {}
- app = "%s/bin/%s" % (build_dir, appname)
+ app = f"{config.vpp_build_dir}/vpp/bin/{appname}"
self.args = [app] + executable_args
self.role = role
self.wait_for_gdb = 'wait-for-gdb'
def setUp(self):
super(QUICTestCase, self).setUp()
- var = "VPP_BUILD_DIR"
- self.build_dir = os.getenv(var, None)
- if self.build_dir is None:
- raise Exception("Environment variable `%s' not set" % var)
- self.vppDebug = 'vpp_debug' in self.build_dir
+ self.vppDebug = 'vpp_debug' in config.vpp_build_dir
self.create_loopback_interfaces(2)
self.uri = "quic://%s/1234" % self.loop0.local_ip4
def server(self, *args):
_args = self.server_echo_test_args + list(args)
self.worker_server = QUICAppWorker(
- self.build_dir,
self.app,
_args,
self.logger,
def client(self, *args):
_args = self.client_echo_test_args + list(args)
self.worker_client = QUICAppWorker(
- self.build_dir,
self.app,
_args,
self.logger,
test_bytes = ''
timeout = 60
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_quic_ext_transfer_big(self):
"""QUIC external transfer, big stream"""
self.server("TX=0", "RX=2G")
class QUICEchoExtQcloseRxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Qclose Rx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_qclose_rx(self):
"""QUIC external transfer, rx close"""
class QUICEchoExtQcloseTxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Qclose Tx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_qclose_tx(self):
"""QUIC external transfer, tx close"""
class QUICEchoExtEarlyQcloseRxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Early Qclose Rx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_early_qclose_rx(self):
"""QUIC external transfer, early rx close"""
class QUICEchoExtEarlyQcloseTxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Early Qclose Tx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_early_qclose_tx(self):
"""QUIC external transfer, early tx close"""
class QUICEchoExtScloseRxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Sclose Rx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_sclose_rx(self):
"""QUIC external transfer, rx stream close"""
class QUICEchoExtScloseTxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Sclose Tx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_sclose_tx(self):
"""QUIC external transfer, tx stream close"""
class QUICEchoExtEarlyScloseRxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Early Sclose Rx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_early_sclose_rx(self):
"""QUIC external transfer, early rx stream close"""
class QUICEchoExtEarlyScloseTxTestCase(QUICEchoExtTestCase):
"""QUIC Echo External Transfer Early Sclose Tx Test Case"""
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_early_sclose_tx(self):
"""QUIC external transfer, early tx stream close"""
test_bytes = ''
timeout = 60
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_quic_ext_transfer_server_stream_big(self):
"""QUIC external server transfer, big stream"""
self.server("TX=2G", "RX=0")
"""QUIC Echo External Transfer Server Stream Qclose Rx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_qclose_rx(self):
"""QUIC external server transfer, rx close"""
"""QUIC Echo External Transfer Server Stream Qclose Tx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_qclose_tx(self):
"""QUIC external server transfer, tx close"""
"""QUIC Echo External Transfer Server Stream Early Qclose Rx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_early_qclose_rx(self):
"""QUIC external server transfer, early rx close"""
"""QUIC Echo External Transfer Server Stream Early Qclose Tx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_early_qclose_tx(self):
"""QUIC external server transfer, early tx close"""
"""QUIC Echo External Transfer Server Stream Sclose Rx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_sclose_rx(self):
"""QUIC external server transfer, rx stream close"""
"""QUIC Echo External Transfer Server Stream Sclose Tx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_sclose_tx(self):
"""QUIC external server transfer, tx stream close"""
"""QUIC Echo External Transfer Server Stream Early Sclose Rx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_early_sclose_rx(self):
"""QUIC external server transfer, early rx stream close"""
"""QUIC Echo Ext Transfer Server Stream Early Sclose Tx Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_server_stream_early_sclose_tx(self):
"""QUIC external server transfer, early tx stream close"""
"""QUIC Echo External Transfer Server Stream MultiWorker Test Case"""
quic_setup = "serverstream"
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("testcase under development")
def test_quic_ext_transfer_server_stream_multi_workers(self):
"""QUIC external server transfer, multi-worker"""
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
from scapy.contrib.geneve import GENEVE
import unittest
import os
import signal
+from config import config
from framework import VppTestCase, VppTestRunner, Worker
def test_vapi_c(self):
""" run C VAPI tests """
- var = "VPP_BUILD_DIR"
- build_dir = os.getenv(var, None)
- self.assertIsNotNone(build_dir,
- "Environment variable `%s' not set" % var)
- executable = f"{build_dir}/bin/vapi_c_test"
+ executable = f"{config.vpp_build_dir}/vpp/bin/vapi_c_test"
worker = Worker([executable, "vapi client",
self.get_api_segment_prefix()], self.logger)
worker.start()
def test_vapi_cpp(self):
""" run C++ VAPI tests """
- var = "VPP_BUILD_DIR"
- build_dir = os.getenv(var, None)
- self.assertIsNotNone(build_dir,
- "Environment variable `%s' not set" % var)
- executable = f"{build_dir}/bin/vapi_cpp_test"
+ executable = f"{config.vpp_build_dir}/vpp/bin/vapi_cpp_test"
worker = Worker([executable, "vapi client",
self.get_api_segment_prefix()], self.logger)
worker.start()
import subprocess
import signal
import glob
-from framework import VppTestCase, VppTestRunner, running_extended_tests, \
- Worker
+from config import config
+from framework import VppTestCase, VppTestRunner, Worker
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath, FibPathProto
iperf3 = '/usr/bin/iperf3'
class LibraryNotFound(Exception):
pass
- def __init__(self, build_dir, appname, executable_args, logger, env=None,
+ def __init__(self, appname, executable_args, logger, env=None,
role=None, *args, **kwargs):
self.role = role
- vpp_install_path = os.getenv('VPP_INSTALL_PATH')
-
- vcl_ldpreload_glob = "{}/**/{}".format(vpp_install_path, self.libname)
+ vcl_ldpreload_glob = f"{config.vpp_install_dir}/**/{self.libname}"
vcl_ldpreload_so = glob.glob(vcl_ldpreload_glob, recursive=True)
if len(vcl_ldpreload_so) < 1:
app = appname
env.update({'LD_PRELOAD': vcl_ldpreload_so})
elif "sock" in appname:
- app = "%s/bin/%s" % (build_dir, appname)
+ app = f"{config.vpp_build_dir}/vpp/bin/{appname}"
env.update({'LD_PRELOAD': vcl_ldpreload_so})
else:
- app = "%s/bin/%s" % (build_dir, appname)
+ app = f"{config.vpp_build_dir}/vpp/bin/{appname}"
self.args = [app] + executable_args
super(VCLAppWorker, self).__init__(self.args, logger, env,
*args, **kwargs)
super(VCLTestCase, cls).tearDownClass()
def setUp(self):
- var = "VPP_BUILD_DIR"
- self.build_dir = os.getenv(var, None)
- if self.build_dir is None:
- raise EnvironmentError("Environment variable `%s' not set" % var)
- self.vppDebug = 'vpp_debug' in self.build_dir
+ self.vppDebug = 'vpp_debug' in config.vpp_install_dir
self.server_addr = "127.0.0.1"
self.server_port = "22000"
self.server_args = [self.server_port]
self.vcl_app_env = {'VCL_APP_SCOPE_LOCAL': "true"}
self.update_vcl_app_env("", "", self.sapi_server_sock)
- worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
+ worker_server = VCLAppWorker(server_app, server_args,
self.logger, self.vcl_app_env, "server")
worker_server.start()
self.sleep(self.pre_test_sleep)
self.update_vcl_app_env("", "", self.sapi_client_sock)
- worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
+ worker_client = VCLAppWorker(client_app, client_args,
self.logger, self.vcl_app_env, "client")
worker_client.start()
worker_client.join(self.timeout)
self.vcl_app_env = {'VCL_APP_SCOPE_GLOBAL': "true"}
self.update_vcl_app_env("1", "1234", self.sapi_server_sock)
- worker_server = VCLAppWorker(self.build_dir, server_app, server_args,
+ worker_server = VCLAppWorker(server_app, server_args,
self.logger, self.vcl_app_env, "server")
worker_server.start()
self.sleep(self.pre_test_sleep)
self.update_vcl_app_env("2", "5678", self.sapi_client_sock)
- worker_client = VCLAppWorker(self.build_dir, client_app, client_args,
+ worker_client = VCLAppWorker(client_app, client_args,
self.logger, self.vcl_app_env, "client")
worker_client.start()
worker_client.join(self.timeout)
self.logger.debug(self.vapi.cli("show session verbose 2"))
self.logger.debug(self.vapi.cli("show app mq"))
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_ldp_cut_thru_echo(self):
""" run LDP cut thru echo test """
self.cut_thru_test("sock_test_server", self.server_args,
"sock_test_client", self.client_echo_test_args)
- @unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
def test_ldp_cut_thru_iperf3(self):
""" run LDP cut thru iperf3 test """
self.cut_thru_test(iperf3, self.server_iperf3_args,
iperf3, self.client_iperf3_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_ldp_cut_thru_uni_dir_nsock(self):
""" run LDP cut thru uni-directional (multiple sockets) test """
"sock_test_client",
self.client_uni_dir_nsock_test_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("sock test apps need to be improved")
def test_ldp_cut_thru_bi_dir_nsock(self):
""" run LDP cut thru bi-directional (multiple sockets) test """
self.loop0.local_ip4,
self.server_port]
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_vcl_thru_host_stack_quic_uni_dir(self):
""" run VCL thru host stack uni-directional QUIC test """
super(LDPIpv6CutThruTestCase, self).tearDown()
self.cut_thru_tear_down()
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_ldp_ipv6_cut_thru_echo(self):
""" run LDP IPv6 cut thru echo test """
self.client_ipv6_echo_test_args)
@unittest.skipUnless(_have_iperf3, "'%s' not found, Skipping.")
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
def test_ldp_ipv6_cut_thru_iperf3(self):
""" run LDP IPv6 cut thru iperf3 test """
self.cut_thru_test(iperf3, self.server_ipv6_iperf3_args,
iperf3, self.client_ipv6_iperf3_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_ldp_ipv6_cut_thru_uni_dir_nsock(self):
""" run LDP IPv6 cut thru uni-directional (multiple sockets) test """
"sock_test_client",
self.client_ipv6_uni_dir_nsock_test_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
@unittest.skip("sock test apps need to be improved")
def test_ldp_ipv6_cut_thru_bi_dir_nsock(self):
""" run LDP IPv6 cut thru bi-directional (multiple sockets) test """
"vcl_test_client",
self.client_ipv6_echo_test_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_vcl_ipv6_cut_thru_uni_dir_nsock(self):
""" run VCL IPv6 cut thru uni-directional (multiple sockets) test """
"vcl_test_client",
self.client_ipv6_uni_dir_nsock_test_args)
- @unittest.skipUnless(running_extended_tests, "part of extended tests")
+ @unittest.skipUnless(config.extended, "part of extended tests")
def test_vcl_ipv6_cut_thru_bi_dir_nsock(self):
""" run VCL IPv6 cut thru bi-directional (multiple sockets) test """
import pexpect
import time
import signal
-from framework import VppTestCase, VppTestRunner, running_extended_tests
-from framework import running_gcov_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath
+@unittest.skipUnless(config.gcov, "part of code coverage tests")
class TestVlib(VppTestCase):
""" Vlib Unit Test Cases """
vpp_worker_count = 1
def tearDown(self):
super(TestVlib, self).tearDown()
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
def test_vlib_main_unittest(self):
""" Vlib main.c Code Coverage Test """
else:
self.logger.info(cmd + " FAIL retval " + str(r.retval))
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
def test_vlib_node_cli_unittest(self):
""" Vlib node_cli.c Code Coverage Test """
else:
self.logger.info(cmd + " FAIL retval " + str(r.retval))
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
def test_vlib_buffer_c_unittest(self):
""" Vlib buffer.c Code Coverage Test """
else:
self.logger.info(cmd + " FAIL retval " + str(r.retval))
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
def test_vlib_format_unittest(self):
""" Vlib format.c Code Coverage Test """
else:
self.logger.info(cmd + " FAIL retval " + str(r.retval))
- @unittest.skipUnless(running_gcov_tests, "part of code coverage tests")
def test_vlib_main_unittest(self):
""" Private Binary API Segment Test (takes 70 seconds) """
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
-from framework import running_gcov_tests
+from framework import VppTestCase, VppTestRunner
class TestVppinfra(VppTestCase):
from scapy.contrib.igmpv3 import IGMPv3, IGMPv3mr, IGMPv3gr
from scapy.layers.vrrp import IPPROTO_VRRP, VRRPv3
from scapy.utils6 import in6_getnsma, in6_getnsmac
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from util import ip6_normalize
VRRP_VR_FLAG_PREEMPT = 1
return pkt
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class TestVRRP4(VppTestCase):
""" IPv4 VRRP Test Case """
self.assertEqual(rx[VRRPv3].addrlist, [vip])
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class TestVRRP6(VppTestCase):
""" IPv6 VRRP Test Case """
import socket
from util import ip4_range
import unittest
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+from config import config
+from framework import VppTestCase, VppTestRunner
from template_bd import BridgeDomain
from scapy.layers.l2 import Ether
from vpp_ip import INVALID_INDEX
-@unittest.skipUnless(running_extended_tests, "part of extended tests")
+@unittest.skipUnless(config.extended, "part of extended tests")
class TestVxlanGpe(BridgeDomain, VppTestCase):
""" VXLAN-GPE Test Case """
from collections import deque
import queue
from six import moves, iteritems
+from config import config
from vpp_papi import VPPApiClient, mac_pton
from hook import Hook
from vpp_ip_route import MPLS_IETF_MAX_LABEL, MPLS_LABEL_INVALID
# install_dir is a class attribute. We need to set it before
# calling the constructor.
- VPPApiClient.apidir = os.getenv('VPP_INSTALL_PATH')
+ VPPApiClient.apidir = config.vpp_install_dir
self.vpp = VPPApiClient(logger=test_class.logger,
read_timeout=read_timeout,
import time
from traceback import format_exc, format_stack
+from config import config
import scapy.compat
from scapy.utils import wrpcap, rdpcap, PcapReader
from scapy.plist import PacketList
self._cap_name = "pcap%u-sw_if_index-%s" % (
self.pg_index, self.sw_if_index)
- def rename_old_pcap_file(self, path, counter):
+ def handle_old_pcap_file(self, path, counter):
filename = os.path.basename(path)
+
+ if not config.keep_pcaps:
+ try:
+ self.test.logger.debug(f"Removing {path}")
+ os.remove(path)
+ except OSError:
+ self.test.logger.debug(f"OSError: Could not remove {path}")
+ return
+
+ # keep
try:
+
if os.path.isfile(path):
name = "%s/history.[timestamp:%f].[%s-counter:%04d].%s" % \
(self.test.tempdir,
self.name,
counter,
filename)
- self.test.logger.debug("Renaming %s->%s" %
- (path, name))
+ self.test.logger.debug("Renaming %s->%s" % (path, name))
os.rename(path, name)
except OSError:
self.test.logger.debug("OSError: Could not rename %s %s" %
"""
# disable the capture to flush the capture
self.disable_capture()
- self.rename_old_pcap_file(self.out_path, self.out_history_counter)
+ self.handle_old_pcap_file(self.out_path, self.out_history_counter)
# FIXME this should be an API, but no such exists atm
self.test.vapi.cli(self.capture_cli)
self._pcap_reader = None
Code Review / vpp.git / commitdiff