aboutsummaryrefslogtreecommitdiffstats
path: root/test/Makefile
blob: 06596424e9d715179dcd99c5868bd3c0a3b91b19 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
.PHONY: verify-python-path

VPP_TEST_FAILED_DIR=/tmp/vpp-failed-unittests/

verify-python-path:
ifndef VPP_PYTHON_PREFIX
	$(error VPP_PYTHON_PREFIX is not set)
endif

.PHONY: verify-no-running-vpp

ifdef VPP_ZOMBIE_NOCHECK
VPP_PIDS=
else
VPP_PIDS=$(shell pgrep -d, -x vpp_main)
endif

ifeq ($(DEBUG),gdb)
FORCE_FOREGROUND=1
else ifeq ($(DEBUG),gdbserver)
FORCE_FOREGROUND=1
else ifeq ($(DEBUG),core)
FORCE_FOREGROUND=1
else
FORCE_FOREGROUND=0
endif

ifdef PROFILE_OUTPUT
PROFILE_OUTPUT_OPTS=-o $(PROFILE_OUTPUT)
endif

ifndef PROFILE_SORT_BY
PROFILE_SORT_BY=cumtime
endif

ifeq ($(PROFILE),1)
PYTHON_PROFILE_OPTS=-m cProfile $(PROFILE_OUTPUT_OPTS) -s $(PROFILE_SORT_BY)
FORCE_FOREGROUND=1
endif

verify-no-running-vpp:
	@if [ "$(VPP_PIDS)" != "" ]; then \
		echo; \
		echo "*** Existing vpp processes detected (PID(s): $(VPP_PIDS)). Running tests under these conditions is not supported. ***"; \
		echo; \
		ps -fp $(VPP_PIDS);\
		echo; \
		false; \
	fi

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

ifeq ($(TEST_DEBUG),1)
VPP_PYTHON_PREFIX:=$(VPP_PYTHON_PREFIX)/debug
PYTHON_EXTRA_DEPENDS=objgraph pympler
else
PYTHON_EXTRA_DEPENDS=
endif

PYTHON_VENV_PATH=$(VPP_PYTHON_PREFIX)/virtualenv
PYTHON_DEPENDS=$(PYTHON_EXTRA_DEPENDS) faulthandler six scapy==2.4.0 pexpect cryptography subprocess32 cffi git+https://github.com/vpp-dev/py-lispnetworking
SCAPY_SOURCE=$(shell find $(PYTHON_VENV_PATH) -name site-packages)
BUILD_COV_DIR=$(BR)/test-cov

GET_PIP_SCRIPT=$(VPP_PYTHON_PREFIX)/get-pip.py
PIP_INSTALL_DONE=$(VPP_PYTHON_PREFIX)/pip-install.done
PIP_PATCH_DONE=$(VPP_PYTHON_PREFIX)/pip-patch.done
PAPI_INSTALL_DONE=$(VPP_PYTHON_PREFIX)/papi-install.done

PAPI_INSTALL_FLAGS=$(PIP_INSTALL_DONE) $(PIP_PATCH_DONE) $(PAPI_INSTALL_DONE)

ifeq ($(PYTHON),)
PYTHON_INTERP=python2.7
else
PYTHON_INTERP=$(PYTHON)
endif

$(GET_PIP_SCRIPT):
	@mkdir -p $(VPP_PYTHON_PREFIX)
	@bash -c "cd $(VPP_PYTHON_PREFIX) && curl -O https://bootstrap.pypa.io/get-pip.py"

$(PIP_INSTALL_DONE): $(GET_PIP_SCRIPT)
	@virtualenv $(PYTHON_VENV_PATH) -p $(PYTHON_INTERP)
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && python $(GET_PIP_SCRIPT)"
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && pip install $(PYTHON_DEPENDS)"
	@touch $@

$(PIP_PATCH_DONE): $(PIP_INSTALL_DONE)
	@echo --- patching ---
	@sleep 1 # Ensure python recompiles patched *.py files -> *.pyc
	for f in $(CURDIR)/patches/scapy-2.4/*.patch ; do \
		echo Applying patch: $$(basename $$f) ; \
		patch -p1 -d $(SCAPY_SOURCE) < $$f ; \
	done
	@touch $@

$(PAPI_INSTALL_DONE): $(PIP_PATCH_DONE)
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && cd $(WS_ROOT)/src/vpp-api/python && python setup.py install"
	@touch $@

define retest-func
@env FORCE_FOREGROUND=$(FORCE_FOREGROUND) VPP_TEST_FAILED_DIR=$(VPP_TEST_FAILED_DIR) scripts/setsid_wrapper.sh $(FORCE_FOREGROUND) $(PYTHON_VENV_PATH)/bin/activate python $(PYTHON_PROFILE_OPTS) run_tests.py -d $(TEST_DIR) $(UNITTEST_EXTRA_OPTS) || env VPP_TEST_FAILED_DIR=$(VPP_TEST_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 $(PYTHON_VENV_PATH)/bin/activate && python sanity_import_vpp_papi.py
SANITY_RUN_VPP_CMD=source $(PYTHON_VENV_PATH)/bin/activate && python sanity_run_vpp.py
endif

sanity: verify-no-running-vpp
	@sys_req/dev_shm_size.sh
	@bash -c "$(SANITY_IMPORT_VPP_PAPI_CMD) ||\
		(echo \"*******************************************************************\" &&\
		 echo \"* Sanity check failed, cannot import vpp_papi\" &&\
		 echo \"* to debug: \" &&\
		 echo \"* 1. enter test shell:   make test-shell\" &&\
		 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)"

.PHONY: ext
ext:
	make -C ext

test-dep: verify-python-path $(PAPI_INSTALL_DONE)

test: verify-python-path $(PAPI_INSTALL_DONE) sanity reset
	$(call retest-func)

test-ext: verify-python-path $(PAPI_INSTALL_DONE) ext sanity reset
	$(call retest-func)

retest: verify-python-path sanity reset
	$(call retest-func)

shell: verify-python-path $(PAPI_INSTALL_DONE)
	@echo "source $(PYTHON_VENV_PATH)/bin/activate;\
		echo '***';\
		echo VPP_TEST_BUILD_DIR=$(VPP_TEST_BUILD_DIR);\
		echo VPP_TEST_BIN=$(VPP_TEST_BIN);\
		echo VPP_TEST_PLUGIN_PATH=$(VPP_TEST_PLUGIN_PATH);\
		echo VPP_TEST_INSTALL_PATH=$(VPP_TEST_INSTALL_PATH);\
		echo EXTERN_TESTS=$(EXTERN_TESTS);\
		echo EXTERN_PLUGINS=$(EXTERN_PLUGINS);\
                echo EXTERN_COV_DIR=$(EXTERN_COV_DIR);\
		echo LD_LIBRARY_PATH=$(LD_LIBRARY_PATH);\
		echo '***';\
		exec </dev/tty" | bash -i

.PHONY: wipe doc

reset:
	@rm -f /dev/shm/vpp-unittest-*
	@rm -rf /tmp/vpp-unittest-*
	@rm -rf $(VPP_TEST_FAILED_DIR)
	@mkdir $(VPP_TEST_FAILED_DIR)

wipe: reset
	@make -C ext clean
	@rm -rf $(PYTHON_VENV_PATH)
	@rm -f $(PAPI_INSTALL_FLAGS)

doc: verify-python-path $(PIP_PATCH_DONE)
	@virtualenv $(PYTHON_VENV_PATH) -p $(PYTHON_INTERP)
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && pip install sphinx sphinx-rtd-theme"
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && make -C doc WS_ROOT=$(WS_ROOT) BR=$(BR) NO_VPP_PAPI=1 html"

.PHONY: wipe-doc

wipe-doc:
	@make -C doc wipe BR=$(BR)

cov: wipe-cov reset ext verify-python-path $(PAPI_INSTALL_DONE)
	@lcov --zerocounters --directory $(VPP_TEST_BUILD_DIR)
	@test -z "$(EXTERN_COV_DIR)" || lcov --zerocounters --directory $(EXTERN_COV_DIR)
	$(call retest-func)
	@mkdir $(BUILD_COV_DIR)
	@lcov --capture --directory $(VPP_TEST_BUILD_DIR) --output-file $(BUILD_COV_DIR)/coverage.info
	@test -z "$(EXTERN_COV_DIR)" || lcov --capture --directory $(EXTERN_COV_DIR) --output-file $(BUILD_COV_DIR)/extern-coverage.info
	@genhtml $(BUILD_COV_DIR)/coverage.info --output-directory $(BUILD_COV_DIR)/html
	@test -z "$(EXTERN_COV_DIR)" || genhtml $(BUILD_COV_DIR)/extern-coverage.info --output-directory $(BUILD_COV_DIR)/extern-html
	@echo
	@echo "Build finished. Code coverage report is in $(BUILD_COV_DIR)/html/index.html"
	@test -z "$(EXTERN_COV_DIR)" || echo "Code coverage report for out-of-tree objects is in $(BUILD_COV_DIR)/extern-html/index.html"

.PHONY: wipe-cov

wipe-cov: wipe
	@rm -rf $(BUILD_COV_DIR)

.PHONY: checkstyle
checkstyle: verify-python-path
	@virtualenv $(PYTHON_VENV_PATH) -p $(PYTHON_INTERP)
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && pip install pycodestyle"
	@bash -c "source $(PYTHON_VENV_PATH)/bin/activate &&\
		pycodestyle --show-source --ignore=W504,E126,E241,E226,E305,E704,E741,E722 -v $(WS_ROOT)/test/*.py ||\
		(echo \"*******************************************************************\" &&\
		 echo \"* Test framework PEP8 compliance check FAILED \" &&\
	         echo \"*******************************************************************\" &&\
		 false)"
	@echo "*******************************************************************"
	@echo "* Test framework PEP8 compliance check passed"
	@echo "*******************************************************************"

help:
	@echo "Running tests:"
	@echo ""
	@echo " test                - build and run (basic) functional tests"
	@echo " test-debug          - build and run (basic) functional tests (debug build)"
	@echo " test-all            - build and run (all) functional tests"
	@echo " test-all-debug      - build and run (all) functional tests (debug build)"
	@echo " retest              - run functional tests"
	@echo " retest-debug        - run functional tests (debug build)"
	@echo " test-wipe           - wipe (temporary) files generated by unit tests"
	@echo " test-shell          - enter shell with test environment"
	@echo " test-shell-debug    - enter shell with test environment (debug build)"
	@echo ""
	@echo "Arguments controlling test runs:"
	@echo " V=[0|1|2]            - set test verbosity level"
	@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 " TIMEOUT=<timeout>    - fail test suite if any single test takes longer than <timeout> (in seconds) to finish"
	@echo " RETRIES=<n>          - retry failed tests <n> times"
	@echo " DEBUG=<type>         - set VPP debugging kind"
	@echo "    DEBUG=core        - detect coredump and load it in gdb on crash"
	@echo "    DEBUG=gdb         - allow easy debugging by printing VPP PID "
	@echo "                        and waiting for user input before running "
	@echo "                        and tearing down a testcase"
	@echo "    DEBUG=gdbserver   - run gdb inside a gdb server, otherwise "
	@echo "                        same as above"
	@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] - 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 "    by wildcard       - wildcard filter is <file>.<class>.<test function>, each can be replaced by '*'"
	@echo "                        e.g. TEST='test_bfd.*.*' is equivalent to above example of filter by file-name"
	@echo "                             TEST='bfd.*.*' is equivalent to above example of filter by file-suffix"
	@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 " VPP_ZOMBIE_NOCHECK=1 - skip checking for vpp (zombie) processes (CAUTION)"
	@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 " 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_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 "Creating test documentation"
	@echo " test-doc            - generate documentation for test framework"
	@echo " test-wipe-doc       - wipe documentation for test framework"
	@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 " test-checkstyle     - check PEP8 compliance"
	@echo ""
an> self.logger.info(self.vapi.ppcli("show classify tables verbose")) self.logger.info(self.vapi.ppcli("show bridge-domain %s detail" % self.bd_id)) if self.acl_active_table == 'mac_inout': self.output_acl_set_interface( self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0) self.input_acl_set_interface( self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0) self.acl_active_table = '' elif self.acl_active_table == 'mac_out': self.output_acl_set_interface( self.pg1, self.acl_tbl_idx.get(self.acl_active_table), 0) self.acl_active_table = '' elif self.acl_active_table == 'mac_in': self.input_acl_set_interface( self.pg0, self.acl_tbl_idx.get(self.acl_active_table), 0) self.acl_active_table = '' super(TestClassifyAcl, self).tearDown() @staticmethod def build_mac_mask(dst_mac='', src_mac='', ether_type=''): """Build MAC ACL mask data with hexstring format :param str dst_mac: source MAC address <0-ffffffffffff> :param str src_mac: destination MAC address <0-ffffffffffff> :param str ether_type: ethernet type <0-ffff> """ return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format( dst_mac, src_mac, ether_type)).rstrip('0') @staticmethod def build_mac_match(dst_mac='', src_mac='', ether_type=''): """Build MAC ACL match data with hexstring format :param str dst_mac: source MAC address <x:x:x:x:x:x> :param str src_mac: destination MAC address <x:x:x:x:x:x> :param str ether_type: ethernet type <0-ffff> """ if dst_mac: dst_mac = dst_mac.replace(':', '') if src_mac: src_mac = src_mac.replace(':', '') return ('{!s:0>12}{!s:0>12}{!s:0>4}'.format( dst_mac, src_mac, ether_type)).rstrip('0') def create_classify_table(self, key, mask, data_offset=0, is_add=1): """Create Classify Table :param str key: key for classify table (ex, ACL name). :param str mask: mask value for interested traffic. :param int match_n_vectors: :param int is_add: option to configure classify table. - create(1) or delete(0) """ r = self.vapi.classify_add_del_table( is_add, binascii.unhexlify(mask), match_n_vectors=(len(mask) - 1) // 32 + 1, miss_next_index=0, current_data_flag=1, current_data_offset=data_offset) self.assertIsNotNone(r, 'No response msg for add_del_table') self.acl_tbl_idx[key] = r.new_table_index def create_classify_session(self, intf, table_index, match, hit_next_index=0xffffffff, is_add=1): """Create Classify Session :param VppInterface intf: Interface to apply classify session. :param int table_index: table index to identify classify table. :param str match: matched value for interested traffic. :param int pbr_action: enable/disable PBR feature. :param int vrfid: VRF id. :param int is_add: option to configure classify session. - create(1) or delete(0) """ r = self.vapi.classify_add_del_session( is_add, table_index, binascii.unhexlify(match), hit_next_index=hit_next_index) self.assertIsNotNone(r, 'No response msg for add_del_session') def input_acl_set_interface(self, intf, table_index, is_add=1): """Configure Input ACL interface :param VppInterface intf: Interface to apply Input ACL feature. :param int table_index: table index to identify classify table. :param int is_add: option to configure classify session. - enable(1) or disable(0) """ r = self.vapi.input_acl_set_interface( is_add, intf.sw_if_index, l2_table_index=table_index) self.assertIsNotNone(r, 'No response msg for acl_set_interface') def output_acl_set_interface(self, intf, table_index, is_add=1): """Configure Output ACL interface :param VppInterface intf: Interface to apply Output ACL feature. :param int table_index: table index to identify classify table. :param int is_add: option to configure classify session. - enable(1) or disable(0) """ r = self.vapi.output_acl_set_interface( is_add, intf.sw_if_index, l2_table_index=table_index) self.assertIsNotNone(r, 'No response msg for acl_set_interface') def create_hosts(self, count, start=0): """ Create required number of host MAC addresses and distribute them among interfaces. Create host IPv4 address for every host MAC address. :param int count: Number of hosts to create MAC/IPv4 addresses for. :param int start: Number to start numbering from. """ n_int = len(self.pg_interfaces) macs_per_if = count / n_int i = -1 for pg_if in self.pg_interfaces: i += 1 start_nr = macs_per_if * i + start end_nr = count + start if i == (n_int - 1) \ else macs_per_if * (i + 1) + start hosts = self.hosts_by_pg_idx[pg_if.sw_if_index] for j in range(start_nr, end_nr): host = Host( "00:00:00:ff:%02x:%02x" % (pg_if.sw_if_index, j), "172.17.1%02x.%u" % (pg_if.sw_if_index, j), "2017:dead:%02x::%u" % (pg_if.sw_if_index, j)) hosts.append(host) def create_upper_layer(self, packet_index, proto, ports=0): p = self.proto_map[proto] if p == 'UDP': if ports == 0: return UDP(sport=random.randint(self.udp_sport_from, self.udp_sport_to), dport=random.randint(self.udp_dport_from, self.udp_dport_to)) else: return UDP(sport=ports, dport=ports) elif p == 'TCP': if ports == 0: return TCP(sport=random.randint(self.tcp_sport_from, self.tcp_sport_to), dport=random.randint(self.tcp_dport_from, self.tcp_dport_to)) else: return TCP(sport=ports, dport=ports) return '' def create_stream(self, src_if, packet_sizes, traffic_type=0, ipv6=0, proto=-1, ports=0, fragments=False, pkt_raw=True, etype=-1): """ Create input packet stream for defined interface using hosts or deleted_hosts list. :param object src_if: Interface to create packet stream for. :param list packet_sizes: List of required packet sizes. :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise. :return: Stream of packets. """ pkts = [] if self.flows.__contains__(src_if): src_hosts = self.hosts_by_pg_idx[src_if.sw_if_index] for dst_if in self.flows[src_if]: dst_hosts = self.hosts_by_pg_idx[dst_if.sw_if_index] n_int = len(dst_hosts) * len(src_hosts) for i in range(0, n_int): dst_host = dst_hosts[i / len(src_hosts)] src_host = src_hosts[i % len(src_hosts)] pkt_info = self.create_packet_info(src_if, dst_if) if ipv6 == 1: pkt_info.ip = 1 elif ipv6 == 0: pkt_info.ip = 0 else: pkt_info.ip = random.choice([0, 1]) if proto == -1: pkt_info.proto = random.choice(self.proto[self.IP]) else: pkt_info.proto = proto payload = self.info_to_payload(pkt_info) p = Ether(dst=dst_host.mac, src=src_host.mac) if etype > 0: p = Ether(dst=dst_host.mac, src=src_host.mac, type=etype) if pkt_info.ip: p /= IPv6(dst=dst_host.ip6, src=src_host.ip6) if fragments: p /= IPv6ExtHdrFragment(offset=64, m=1) else: if fragments: p /= IP(src=src_host.ip4, dst=dst_host.ip4, flags=1, frag=64) else: p /= IP(src=src_host.ip4, dst=dst_host.ip4) if traffic_type == self.ICMP: if pkt_info.ip: p /= ICMPv6EchoRequest(type=self.icmp6_type, code=self.icmp6_code) else: p /= ICMP(type=self.icmp4_type, code=self.icmp4_code) else: p /= self.create_upper_layer(i, pkt_info.proto, ports) if pkt_raw: p /= Raw(payload) pkt_info.data = p.copy() if pkt_raw: size = random.choice(packet_sizes) self.extend_packet(p, size) pkts.append(p) return pkts def verify_capture(self, pg_if, capture, traffic_type=0, ip_type=0, etype=-1): """ Verify captured input packet stream for defined interface. :param object pg_if: Interface to verify captured packet stream for. :param list capture: Captured packet stream. :param traffic_type: 1: ICMP packet, 2: IPv6 with EH, 0: otherwise. """ last_info = dict() for i in self.pg_interfaces: last_info[i.sw_if_index] = None dst_sw_if_index = pg_if.sw_if_index for packet in capture: if etype > 0: if packet[Ether].type != etype: self.logger.error(ppp("Unexpected ethertype in packet:", packet)) else: continue try: # Raw data for ICMPv6 are stored in ICMPv6EchoRequest.data if traffic_type == self.ICMP and ip_type == self.IPV6: payload_info = self.payload_to_info( packet[ICMPv6EchoRequest].data) payload = packet[ICMPv6EchoRequest] else: payload_info = self.payload_to_info(packet[Raw]) payload = packet[self.proto_map[payload_info.proto]] except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) raise if ip_type != 0: self.assertEqual(payload_info.ip, ip_type) if traffic_type == self.ICMP: try: if payload_info.ip == 0: self.assertEqual(payload.type, self.icmp4_type) self.assertEqual(payload.code, self.icmp4_code) else: self.assertEqual(payload.type, self.icmp6_type) self.assertEqual(payload.code, self.icmp6_code) except: self.logger.error(ppp("Unexpected or invalid packet " "(outside network):", packet)) raise else: try: ip_version = IPv6 if payload_info.ip == 1 else IP ip = packet[ip_version] packet_index = payload_info.index self.assertEqual(payload_info.dst, dst_sw_if_index) self.logger.debug("Got packet on port %s: src=%u (id=%u)" % (pg_if.name, payload_info.src, packet_index)) next_info = self.get_next_packet_info_for_interface2( payload_info.src, dst_sw_if_index, last_info[payload_info.src]) last_info[payload_info.src] = next_info self.assertTrue(next_info is not None) self.assertEqual(packet_index, next_info.index) saved_packet = next_info.data # Check standard fields self.assertEqual(ip.src, saved_packet[ip_version].src) self.assertEqual(ip.dst, saved_packet[ip_version].dst) p = self.proto_map[payload_info.proto] if p == 'TCP': tcp = packet[TCP] self.assertEqual(tcp.sport, saved_packet[ TCP].sport) self.assertEqual(tcp.dport, saved_packet[ TCP].dport) elif p == 'UDP': udp = packet[UDP] self.assertEqual(udp.sport, saved_packet[ UDP].sport) self.assertEqual(udp.dport, saved_packet[ UDP].dport) except: self.logger.error(ppp("Unexpected or invalid packet:", packet)) raise for i in self.pg_interfaces: remaining_packet = self.get_next_packet_info_for_interface2( i, dst_sw_if_index, last_info[i.sw_if_index]) self.assertTrue( remaining_packet is None, "Port %u: Packet expected from source %u didn't arrive" % (dst_sw_if_index, i.sw_if_index)) def run_traffic_no_check(self): # Test # Create incoming packet streams for packet-generator interfaces for i in self.pg_interfaces: if self.flows.__contains__(i): pkts = self.create_stream(i, self.pg_if_packet_sizes) if len(pkts) > 0: i.add_stream(pkts) # Enable packet capture and start packet sending self.pg_enable_capture(self.pg_interfaces) self.pg_start() def run_verify_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0, frags=False, pkt_raw=True, etype=-1): # Test # Create incoming packet streams for packet-generator interfaces pkts_cnt = 0 for i in self.pg_interfaces: if self.flows.__contains__(i): pkts = self.create_stream(i, self.pg_if_packet_sizes, traffic_type, ip_type, proto, ports, frags, pkt_raw, etype) if len(pkts) > 0: i.add_stream(pkts) pkts_cnt += len(pkts) # Enable packet capture and start packet sendingself.IPV self.pg_enable_capture(self.pg_interfaces) self.pg_start() # Verify # Verify outgoing packet streams per packet-generator interface for src_if in self.pg_interfaces: if self.flows.__contains__(src_if): for dst_if in self.flows[src_if]: capture = dst_if.get_capture(pkts_cnt) self.logger.info("Verifying capture on interface %s" % dst_if.name) self.verify_capture(dst_if, capture, traffic_type, ip_type, etype) def run_verify_negat_test(self, traffic_type=0, ip_type=0, proto=-1, ports=0, frags=False, etype=-1): # Test self.reset_packet_infos() for i in self.pg_interfaces: if self.flows.__contains__(i): pkts = self.create_stream(i, self.pg_if_packet_sizes, traffic_type, ip_type, proto, ports, frags, True, etype) if len(pkts) > 0: i.add_stream(pkts) # Enable packet capture and start packet sending self.pg_enable_capture(self.pg_interfaces) self.pg_start() # Verify # Verify outgoing packet streams per packet-generator interface for src_if in self.pg_interfaces: if self.flows.__contains__(src_if): for dst_if in self.flows[src_if]: self.logger.info("Verifying capture on interface %s" % dst_if.name) capture = dst_if.get_capture(0) self.assertEqual(len(capture), 0) def build_classify_table(self, src_mac='', dst_mac='', ether_type='', etype='', key='mac', hit_next_index=0xffffffff): # Basic ACL testing a_mask = self.build_mac_mask(src_mac=src_mac, dst_mac=dst_mac, ether_type=ether_type) self.create_classify_table(key, a_mask) for host in self.hosts_by_pg_idx[self.pg0.sw_if_index]: s_mac = host.mac if src_mac else '' if dst_mac: for dst_if in self.flows[self.pg0]: for dst_host in self.hosts_by_pg_idx[dst_if.sw_if_index]: self.create_classify_session( self.pg0, self.acl_tbl_idx.get(key), self.build_mac_match(src_mac=s_mac, dst_mac=dst_host.mac, ether_type=etype), hit_next_index=hit_next_index) else: self.create_classify_session( self.pg0, self.acl_tbl_idx.get(key), self.build_mac_match(src_mac=s_mac, dst_mac='', ether_type=etype), hit_next_index=hit_next_index) def test_0000_warmup_test(self): """ Learn the MAC addresses """ self.create_hosts(2) self.run_traffic_no_check() def test_0010_inacl_permit_src_mac(self): """ Input L2 ACL test - permit source MAC Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with source MAC address. - Send and verify received packets on pg1 interface. """ key = 'mac_in' self.build_classify_table(src_mac='ffffffffffff', key=key) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) def test_0011_inacl_permit_dst_mac(self): """ Input L2 ACL test - permit destination MAC Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with destination MAC address. - Send and verify received packets on pg1 interface. """ key = 'mac_in' self.build_classify_table(dst_mac='ffffffffffff', key=key) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) def test_0012_inacl_permit_src_dst_mac(self): """ Input L2 ACL test - permit source and destination MAC Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with source and destination MAC addresses. - Send and verify received packets on pg1 interface. """ key = 'mac_in' self.build_classify_table( src_mac='ffffffffffff', dst_mac='ffffffffffff', key=key) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) def test_0013_inacl_permit_ether_type(self): """ Input L2 ACL test - permit ether_type Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with destination MAC address. - Send and verify received packets on pg1 interface. """ key = 'mac_in' self.build_classify_table( ether_type='ffff', etype=hex(ETH_P_IP)[2:], key=key) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) def test_0015_inacl_deny(self): """ Input L2 ACL test - deny Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with source MAC address. - Send and verify no received packets on pg1 interface. """ key = 'mac_in' self.build_classify_table( src_mac='ffffffffffff', hit_next_index=0, key=key) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_negat_test(self.IP, self.IPV4, -1) def test_0020_outacl_permit(self): """ Output L2 ACL test - permit Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with source MAC address. - Send and verify received packets on pg1 interface. """ key = 'mac_out' self.build_classify_table(src_mac='ffffffffffff', key=key) self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) def test_0025_outacl_deny(self): """ Output L2 ACL test - deny Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACL with source MAC address. - Send and verify no received packets on pg1 interface. """ key = 'mac_out' self.build_classify_table( src_mac='ffffffffffff', hit_next_index=0, key=key) self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_negat_test(self.IP, self.IPV4, -1) def test_0030_inoutacl_permit(self): """ Input+Output L2 ACL test - permit Test scenario for basic IP ACL with source IP - Create IPv4 stream for pg0 -> pg1 interface. - Create ACLs with source MAC address. - Send and verify received packets on pg1 interface. """ key = 'mac_inout' self.build_classify_table(src_mac='ffffffffffff', key=key) self.output_acl_set_interface(self.pg1, self.acl_tbl_idx.get(key)) self.input_acl_set_interface(self.pg0, self.acl_tbl_idx.get(key)) self.acl_active_table = key self.run_verify_test(self.IP, self.IPV4, -1) if __name__ == '__main__': unittest.main(testRunner=VppTestRunner)