PAPI_INSTALL_FLAGS=$(PIP_INSTALL_DONE) $(PIP_PATCH_DONE) $(PAPI_INSTALL_DONE)
$(PIP_INSTALL_DONE):
- @virtualenv $(PYTHON_VENV_PATH)
+ @virtualenv $(PYTHON_VENV_PATH) -p python2.7
@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && pip install $(PYTHON_DEPENDS)"
@touch $@
@rm -f $(PAPI_INSTALL_FLAGS)
doc: verify-python-path
- @virtualenv $(PYTHON_VENV_PATH)
+ @virtualenv $(PYTHON_VENV_PATH) -p python2.7
@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && pip install $(PYTHON_DEPENDS) sphinx"
@bash -c "source $(PYTHON_VENV_PATH)/bin/activate && make -C doc WS_ROOT=$(WS_ROOT) BR=$(BR) NO_VPP_PAPI=1 html"
.. _SkipTest: https://docs.python.org/2/library/unittest.html#unittest.SkipTest
.. _virtualenv: http://docs.python-guide.org/en/latest/dev/virtualenvs/
.. _scapy: http://www.secdev.org/projects/scapy/
+.. _logging: https://docs.python.org/2/library/logging.html
.. |vtf| replace:: VPP Test Framework
3. *test_<name>*:
This is the guts of the test case. It should execute the test scenario
and use the various assert functions from the unittest framework to check
- necessary.
+ necessary. Multiple test_<name> methods can exist in a test case.
4. `tearDown <VppTestCase.tearDown>`:
The tearDown function is called after each test function with the purpose
of doing partial cleanup.
Method called once after running all of the test functions to perform
the final cleanup.
+Logging
+#######
+
+Each test case has a logger automatically created for it, stored in
+'logger' property, based on logging_. Use the logger's standard methods
+debug(), info(), error(), ... to emit log messages to the logger.
+
+All the log messages go always into a log file in temporary directory
+(see below).
+
+To control the messages printed to console, specify the V= parameter.
+
+.. code-block:: shell
+
+ make test # minimum verbosity
+ make test V=1 # moderate verbosity
+ make test V=2 # maximum verbosity
+
Test temporary directory and VPP life cycle
###########################################
Test separation is achieved by separating the test files and vpp instances.
Each test creates a temporary directory and it's name is used to create
a shared memory prefix which is used to run a VPP instance.
+The temporary directory name contains the testcase class name for easy
+reference, so for testcase named 'TestVxlan' the directory could be named
+e.g. vpp-unittest-TestVxlan-UNUP3j.
This way, there is no conflict between any other VPP instances running
on the box and the test VPP. Any temporary files created by the test case
are stored in this temporary test directory.
+The test temporary directory holds the following interesting files:
+
+* log.txt - this contains the logger output on max verbosity
+* pg*_in.pcap - last injected packet stream into VPP, named after the interface,
+ so for pg0, the file will be named pg0_in.pcap
+* pg*_out.pcap - last capture file created by VPP for interface, similarly,
+ named after the interface, so for e.g. pg1, the file will be named
+ pg1_out.pcap
+* history files - whenever the capture is restarted or a new stream is added,
+ the existing files are rotated and renamed, soo all the pcap files
+ are always saved for later debugging if needed
+* core - if vpp dumps a core, it'll be stored in the temporary directory
+* vpp_stdout.txt - file containing output which vpp printed to stdout
+* vpp_stderr.txt - file containing output which vpp printed to stderr
+
+*NOTE*: existing temporary directories named vpp-unittest-* are automatically
+removed when invoking 'make test*' or 'make retest*' to keep the temporary
+directory clean.
+
Virtual environment
###################
So e.g. `remote_mac <VppInterface.remote_mac>` address is the MAC address
assigned to the virtual host connected to the VPP.
+Automatically generated addresses
+#################################
+
+To send packets, one needs to typically provide some addresses, otherwise
+the packets will be dropped. The interface objects in |vtf| automatically
+provide addresses based on (typically) their indexes, which ensures
+there are no conflicts and eases debugging by making the addressing scheme
+consistent.
+
+The developer of a test case typically doesn't need to work with the actual
+numbers, rather using the properties of the objects. The addresses typically
+come in two flavors: '<address>' and '<address>n' - note the 'n' suffix.
+The former address is a Python string, while the latter is translated using
+socket.inet_pton to raw format in network byte order - this format is suitable
+for passing as an argument to VPP APIs.
+
+e.g. for the IPv4 address assigned to the VPP interface:
+
+* local_ip4 - Local IPv4 address on VPP interface (string)
+* local_ip4n - Local IPv4 address - raw, suitable as API parameter.
+
+These addresses need to be configured in VPP to be usable using e.g.
+`config_ip4` API. Please see the documentation to `VppInterface` for more
+details.
+
+By default, there is one remote address of each kind created for L3:
+remote_ip4 and remote_ip6. If the test needs more addresses, because it's
+simulating more remote hosts, they can be generated using
+`generate_remote_hosts` API and the entries for them inserted into the ARP
+table using `configure_ipv4_neighbors` API.
+
Packet flow in the |vtf|
########################
Packets are written into a temporary .pcap file, which is then read by the VPP
and the packets are injected into the VPP world.
+To add a list of packets to an interface, call the `add_stream` method on that
+interface. Once everything is prepared, call `pg_start` method to start
+the packet generator on the VPP side.
+
VPP -> test framework
~~~~~~~~~~~~~~~~~~~~~
capture feature is used to capture and write traffic to a temporary .pcap file,
which is then read and analyzed by the |vtf|.
+The following APIs are available to the test case for reading pcap files.
+
+* `get_capture`: this API is suitable for bulk & batch style of test, where
+ a list of packets is prepared & sent, then the received packets are read
+ and verified. The API needs the number of packets which are expected to
+ be captured (ignoring filtered packets - see below) to know when the pcap
+ file is completely written by the VPP. If using packet infos for verifying
+ packets, then the counts of the packet infos can be automatically used
+ by `get_capture` to get the proper count (in this case the default value
+ None can be supplied as expected_count or ommitted altogether).
+* `wait_for_packet`: this API is suitable for interactive style of test,
+ e.g. when doing session management, three-way handsakes, etc. This API waits
+ for and returns a single packet, keeping the capture file in place
+ and remembering context. Repeated invocations return following packets
+ (or raise Exception if timeout is reached) from the same capture file
+ (= packets arriving on the same interface).
+
+*NOTE*: it is not recommended to mix these APIs unless you understand how they
+work internally. None of these APIs rotate the pcap capture file, so calling
+e.g. `get_capture` after `wait_for_packet` will return already read packets.
+It is safe to switch from one API to another after calling `enable_capture`
+as that API rotates the capture file.
+
+Automatic filtering of packets:
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Both APIs (`get_capture` and `wait_for_packet`) by default filter the packet
+capture, removing known uninteresting packets from it - these are IPv6 Router
+Advertisments and IPv6 Router Alerts. These packets are unsolicitated
+and from the point of |vtf| are random. If a test wants to receive these
+packets, it should specify either None or a custom filtering function
+as the value to the 'filter_out_fn' argument.
+
+Common API flow for sending/receiving packets:
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+We will describe a simple scenario, where packets are sent from pg0 to pg1
+interface, assuming that the interfaces were created using
+`create_pg_interfaces` API.
+
+1. Create a list of packets for pg0::
+
+ packet_count = 10
+ packets = create_packets(src=self.pg0, dst=self.pg1,
+ count=packet_count)
+
+2. Add that list of packets to the source interface::
+
+ self.pg0.add_stream(packets)
+
+3. Enable capture on the destination interface::
+
+ self.pg1.enable_capture()
+
+4. Start the packet generator::
+
+ self.pg_start()
+
+5. Wait for capture file to appear and read it::
+
+ capture = self.pg1.get_capture(expected_count=packet_count)
+
+6. Verify packets match sent packets::
+
+ self.verify_capture(send=packets, captured=capture)
+
Test framework objects
######################
* `VppSubInterface`: VPP sub-interface abstract class, containing common
functionality for e.g. `VppDot1QSubint` and `VppDot1ADSubint` classes
-How VPP API/CLI is called
-#########################
+How VPP APIs/CLIs are called
+############################
Vpp provides python bindings in a python module called vpp-papi, which the test
framework installs in the virtual environment. A shim layer represented by
more readable. E.g. ip_add_del_route API takes ~25 parameters, of which
in the common case, only 3 are needed.
+Utility methods
+###############
+
+Some interesting utility methods are:
+
+* `ppp`: 'Pretty Print Packet' - returns a string containing the same output
+ as Scapy's packet.show() would print
+* `ppc`: 'Pretty Print Capture' - returns a string containing printout of
+ a capture (with configurable limit on the number of packets printed from it)
+ using `ppp`
+
+*NOTE*: Do not use Scapy's packet.show() in the tests, because it prints
+the output to stdout. All output should go to the logger associated with
+the test case.
+
Example: how to add a new test
##############################
-In this example, we will describe how to add a new test case which tests VPP...
+In this example, we will describe how to add a new test case which tests
+basic IPv4 forwarding.
+
+1. Add a new file called test_ip4_fwd.py in the test directory, starting
+ with a few imports::
+
+ from framework import VppTestCase
+ from scapy.layers.l2 import Ether
+ from scapy.packet import Raw
+ from scapy.layers.inet import IP, UDP
+ from random import randint
+
+2. Create a class inherited from the VppTestCase::
+
+ class IP4FwdTestCase(VppTestCase):
+ """ IPv4 simple forwarding test case """
+
+2. Add a setUpClass function containing the setup needed for our test to run::
+
+ @classmethod
+ def setUpClass(self):
+ super(IP4FwdTestCase, self).setUpClass()
+ self.create_pg_interfaces(range(2)) # create pg0 and pg1
+ for i in self.pg_interfaces:
+ i.admin_up() # put the interface up
+ i.config_ip4() # configure IPv4 address on the interface
+ i.resolve_arp() # resolve ARP, so that we know VPP MAC
+
+3. Create a helper method to create the packets to send::
+
+ def create_stream(self, src_if, dst_if, count):
+ packets = []
+ for i in range(count):
+ # create packet info stored in the test case instance
+ info = self.create_packet_info(src_if, dst_if)
+ # convert the info into packet payload
+ payload = self.info_to_payload(info)
+ # create the packet itself
+ p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
+ IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4) /
+ UDP(sport=randint(1000, 2000), dport=5678) /
+ Raw(payload))
+ # store a copy of the packet in the packet info
+ info.data = p.copy()
+ # append the packet to the list
+ packets.append(p)
+
+ # return the created packet list
+ return packets
+
+4. Create a helper method to verify the capture::
+
+ def verify_capture(self, src_if, dst_if, capture):
+ packet_info = None
+ for packet in capture:
+ try:
+ ip = packet[IP]
+ udp = packet[UDP]
+ # convert the payload to packet info object
+ payload_info = self.payload_to_info(str(packet[Raw]))
+ # make sure the indexes match
+ self.assert_equal(payload_info.src, src_if.sw_if_index,
+ "source sw_if_index")
+ self.assert_equal(payload_info.dst, dst_if.sw_if_index,
+ "destination sw_if_index")
+ packet_info = self.get_next_packet_info_for_interface2(
+ src_if.sw_if_index,
+ dst_if.sw_if_index,
+ packet_info)
+ # make sure we didn't run out of saved packets
+ self.assertIsNotNone(packet_info)
+ self.assert_equal(payload_info.index, packet_info.index,
+ "packet info index")
+ saved_packet = packet_info.data # fetch the saved packet
+ # assert the values match
+ self.assert_equal(ip.src, saved_packet[IP].src,
+ "IP source address")
+ # ... more assertions here
+ self.assert_equal(udp.sport, saved_packet[UDP].sport,
+ "UDP source port")
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:",
+ packet))
+ raise
+ remaining_packet = self.get_next_packet_info_for_interface2(
+ src_if.sw_if_index,
+ dst_if.sw_if_index,
+ packet_info)
+ self.assertIsNone(remaining_packet,
+ "Interface %s: Packet expected from interface "
+ "%s didn't arrive" % (dst_if.name, src_if.name))
+
+5. Add the test code to test_basic function::
+
+ def test_basic(self):
+ count = 10
+ # create the packet stream
+ packets = self.create_stream(self.pg0, self.pg1, count)
+ # add the stream to the source interface
+ self.pg0.add_stream(packets)
+ # enable capture on both interfaces
+ self.pg0.enable_capture()
+ self.pg1.enable_capture()
+ # start the packet generator
+ self.pg_start()
+ # get capture - the proper count of packets was saved by
+ # create_packet_info() based on dst_if parameter
+ capture = self.pg1.get_capture()
+ # assert nothing captured on pg0 (always do this last, so that
+ # some time has already passed since pg_start())
+ self.pg0.assert_nothing_captured()
+ # verify capture
+ self.verify_capture(self.pg0, self.pg1, capture)
+
+6. Run the test by issuing 'make test'.
-1. Add a new file called...
-2. Add a setUpClass function containing...
-3. Add the test code in test...
-4. Run the test...
|vtf| module documentation
##########################
-
+
.. toctree::
:maxdepth: 2
:glob:
debug_cli = ""
if cls.step or cls.debug_gdb or cls.debug_gdbserver:
debug_cli = "cli-listen localhost:5002"
- cls.vpp_cmdline = [cls.vpp_bin, "unix", "{", "nodaemon", debug_cli, "}",
+ cls.vpp_cmdline = [cls.vpp_bin,
+ "unix", "{", "nodaemon", debug_cli, "}",
"api-segment", "{", "prefix", cls.shm_prefix, "}"]
if cls.plugin_path is not None:
cls.vpp_cmdline.extend(["plugin_path", cls.plugin_path])
print(double_line_delim)
print("VPP or GDB server is still running")
print(single_line_delim)
- raw_input("When done debugging, press ENTER to kill the process"
- " and finish running the testcase...")
+ raw_input("When done debugging, press ENTER to kill the "
+ "process and finish running the testcase...")
if hasattr(cls, 'vpp'):
if hasattr(cls, 'vapi'):
def __init__(self, stream, descriptions, verbosity):
"""
- :param stream File descriptor to store where to report test results. Set
- to the standard error stream by default.
- :param descriptions Boolean variable to store information if to use test
- case descriptions.
+ :param stream File descriptor to store where to report test results.
+ Set to the standard error stream by default.
+ :param descriptions Boolean variable to store information if to use
+ test case descriptions.
:param verbosity Integer variable to store required verbosity level.
"""
unittest.TestResult.__init__(self, stream, descriptions, verbosity)
unittest.TestResult.stopTest(self, test)
if self.verbosity > 0:
self.stream.writeln(single_line_delim)
- self.stream.writeln("%-60s%s" %
- (self.getDescription(test), self.result_string))
+ self.stream.writeln("%-60s%s" % (self.getDescription(test),
+ self.result_string))
self.stream.writeln(single_line_delim)
else:
- self.stream.writeln("%-60s%s" %
- (self.getDescription(test), self.result_string))
+ self.stream.writeln("%-60s%s" % (self.getDescription(test),
+ self.result_string))
def printErrors(self):
"""
s = signaldict[abs(self.testcase.vpp.returncode)]
else:
s = "unknown"
- msg = "VPP subprocess died unexpectedly with returncode %d [%s]" % (
- self.testcase.vpp.returncode, s)
+ msg = "VPP subprocess died unexpectedly with returncode %d [%s]" %\
+ (self.testcase.vpp.returncode, s)
self.logger.critical(msg)
core_path = self.testcase.tempdir + '/core'
if os.path.isfile(core_path):
self.pg_start()
- # Pick first received frame and check if it's the
- # non-encapsulated frame
+ # Pick first received frame and check if it's the non-encapsulated
+ # frame
out = self.pg1.get_capture(1)
pkt = out[0]
self.assert_eq_pkts(pkt, self.frame_request)
payload_info = self.payload_to_info(str(packet[Raw]))
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)" %
- (dst_if.name, payload_info.src, packet_index))
+ self.logger.debug(
+ "Got packet on port %s: src=%u (id=%u)" %
+ (dst_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])
pkts = self.create_stream(self.pg0, self.pg2, self.pg_if_packet_sizes)
self.pg0.add_stream(pkts)
- self.create_classify_table(
- 'mac', self.build_mac_mask(src_mac='ffffffffffff'), data_offset=-14)
+ self.create_classify_table('mac',
+ self.build_mac_mask(src_mac='ffffffffffff'),
+ data_offset=-14)
self.create_classify_session(
self.pg0, self.acl_tbl_idx.get('mac'),
self.build_mac_match(src_mac=self.pg0.remote_mac))
pkts = self.create_stream(self.pg0, self.pg3, self.pg_if_packet_sizes)
self.pg0.add_stream(pkts)
- self.create_classify_table('pbr', self.build_ip_mask(src_ip='ffffffff'))
+ self.create_classify_table(
+ 'pbr', self.build_ip_mask(
+ src_ip='ffffffff'))
pbr_option = 1
self.create_classify_session(
self.pg0, self.acl_tbl_idx.get('pbr'),
if len(payload) * 2 != len(masked_expected_data):
return False
- # iterate over pairs: raw byte from payload and ASCII code for that byte
- # from masked payload (or XX if masked)
+ # iterate over pairs: raw byte from payload and ASCII code for that
+ # byte from masked payload (or XX if masked)
for i in range(len(payload)):
p = payload[i]
m = masked_expected_data[2 * i:2 * i + 2]
payload_info = self.payload_to_info(str(packet[Raw]))
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)" %
- (dst_if.name, payload_info.src, packet_index))
+ self.logger.debug(
+ "Got packet on port %s: src=%u (id=%u)" %
+ (dst_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])
- add new 1k,
- del 1.5k
- ..note:: Python API is to slow to add many routes, needs C code replacement.
+ ..note:: Python API is too slow to add many routes, needs replacement.
"""
def config_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
:return list: added ips with 32 prefix
"""
added_ips = []
- dest_addr = int(
- socket.inet_pton(socket.AF_INET, start_dest_addr).encode('hex'), 16)
+ dest_addr = int(socket.inet_pton(socket.AF_INET,
+ start_dest_addr).encode('hex'),
+ 16)
dest_addr_len = 32
n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
for _ in range(count):
def unconfig_fib_many_to_one(self, start_dest_addr, next_hop_addr, count):
removed_ips = []
- dest_addr = int(
- socket.inet_pton(socket.AF_INET, start_dest_addr).encode('hex'), 16)
+ dest_addr = int(socket.inet_pton(socket.AF_INET,
+ start_dest_addr).encode('hex'),
+ 16)
dest_addr_len = 32
n_next_hop_addr = socket.inet_pton(socket.AF_INET, next_hop_addr)
for _ in range(count):
try:
# Create pg interfaces
- cls.create_pg_interfaces(range(cls.nr_of_vrfs * cls.pg_ifs_per_vrf))
+ cls.create_pg_interfaces(
+ range(cls.nr_of_vrfs * cls.pg_ifs_per_vrf))
# Packet flows mapping pg0 -> pg1, pg2 etc.
cls.flows = dict()
"""
Create packet streams for all configured l2-pg interfaces, send all
prepared packet streams and verify that:
- - all packets received correctly on all pg-l2 interfaces assigned to
- bridge domains
+ - all packets received correctly on all pg-l2 interfaces assigned
+ to bridge domains
- no packet received on all pg-l2 interfaces not assigned to bridge
domains
:raise RuntimeError: If no packet captured on l2-pg interface assigned
- to the bridge domain or if any packet is captured on l2-pg interface
- not assigned to the bridge domain.
+ to the bridge domain or if any packet is captured on l2-pg
+ interface not assigned to the bridge domain.
"""
# Test
# Create incoming packet streams for packet-generator interfaces
from scapy.packet import Raw
from scapy.layers.l2 import Ether, Dot1Q
-from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, ICMPv6ND_RA, \
- ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation
+from scapy.layers.inet6 import IPv6, UDP, ICMPv6ND_NS, ICMPv6ND_RS, \
+ ICMPv6ND_RA, ICMPv6NDOptSrcLLAddr, getmacbyip6, ICMPv6MRD_Solicitation
from util import ppp
from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ptop, in6_islladdr, \
in6_mactoifaceid
payload_info = self.payload_to_info(str(packet[Raw]))
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)" %
- (dst_if.name, payload_info.src, packet_index))
+ self.logger.debug(
+ "Got packet on port %s: src=%u (id=%u)" %
+ (dst_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])
intf.assert_nothing_captured(remark=remark)
def test_ns(self):
- """ IPv6 Neighbour Soliciatation Exceptions
+ """ IPv6 Neighbour Solicitation Exceptions
- Test sceanrio:
+ Test scenario:
- Send an NS Sourced from an address not covered by the link sub-net
- Send an NS to an mcast address the router has not joined
- Send NS for a target address the router does not onn.
ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
pkts = [p]
- self.send_and_assert_no_replies(self.pg0, pkts,
- "No response to NS source by address not on sub-net")
+ self.send_and_assert_no_replies(
+ self.pg0, pkts,
+ "No response to NS source by address not on sub-net")
#
- # An NS for sent to a solicited mcast group the router is not a member of
- # FAILS
+ # An NS for sent to a solicited mcast group the router is
+ # not a member of FAILS
#
if 0:
nsma = in6_getnsma(inet_pton(socket.AF_INET6, "fd::ffff"))
ICMPv6NDOptSrcLLAddr(lladdr=self.pg0.remote_mac))
pkts = [p]
- self.send_and_assert_no_replies(self.pg0, pkts,
- "No response to NS sent to unjoined mcast address")
+ self.send_and_assert_no_replies(
+ self.pg0, pkts,
+ "No response to NS sent to unjoined mcast address")
#
# An NS whose target address is one the router does not own
in6_ptop(mk_ll_addr(intf.local_mac)))
def test_rs(self):
- """ IPv6 Router Soliciatation Exceptions
+ """ IPv6 Router Solicitation Exceptions
- Test sceanrio:
+ Test scenario:
"""
#
- # Before we begin change the IPv6 RA responses to use the unicast address
- # that way we will not confuse them with the periodic Ras which go to the Mcast
- # address
+ # Before we begin change the IPv6 RA responses to use the unicast
+ # address - that way we will not confuse them with the periodic
+ # RAs which go to the mcast address
#
self.pg0.ip6_ra_config(send_unicast=1)
#
# When we reconfiure the IPv6 RA config, we reset the RA rate limiting,
- # so we need to do this before each test below so as not to drop packets for
- # rate limiting reasons. Test this works here.
+ # so we need to do this before each test below so as not to drop
+ # packets for rate limiting reasons. Test this works here.
#
self.pg0.ip6_ra_config(send_unicast=1)
self.send_and_expect_ra(self.pg0, pkts, "Rate limit reset RS")
self.pg0, pkts, "RS sourced from link-local", src_ip=ll)
#
- # Source from the unspecified address ::. This happens when the RS is sent before
- # the host has a configured address/sub-net, i.e. auto-config.
- # Since the sender has no IP address, the reply comes back mcast - so the
- # capture needs to not filter this.
- # If we happen to pick up the periodic RA at this point then so be it, it's not
- # an error.
+ # Source from the unspecified address ::. This happens when the RS
+ # is sent before the host has a configured address/sub-net,
+ # i.e. auto-config. Since the sender has no IP address, the reply
+ # comes back mcast - so the capture needs to not filter this.
+ # If we happen to pick up the periodic RA at this point then so be it,
+ # it's not an error.
#
self.pg0.ip6_ra_config(send_unicast=1)
p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
@unittest.skip("Unsupported")
def test_mrs(self):
- """ IPv6 Multicast Router Soliciatation Exceptions
+ """ IPv6 Multicast Router Solicitation Exceptions
- Test sceanrio:
+ Test scenario:
"""
#
# Create BD with MAC learning and unknown unicast flooding disabled
# and put interfaces to this BD
- cls.vapi.bridge_domain_add_del(bd_id=cls.bd_id, uu_flood=0, learn=0)
+ cls.vapi.bridge_domain_add_del(
+ bd_id=cls.bd_id, uu_flood=0, learn=0)
for pg_if in cls.pg_interfaces:
cls.vapi.sw_interface_set_l2_bridge(pg_if.sw_if_index,
bd_id=cls.bd_id)
end_nr = start + count / n_int
for j in range(start, end_nr):
host = self.hosts_by_pg_idx[pg_if.sw_if_index][j]
- self.vapi.l2fib_add_del(host.mac, self.bd_id, pg_if.sw_if_index,
- static_mac=1)
+ self.vapi.l2fib_add_del(
+ host.mac, self.bd_id, pg_if.sw_if_index, static_mac=1)
counter += 1
percentage = counter / count * 100
if percentage > percent:
for pg_if in self.pg_interfaces:
for j in range(count / n_int):
host = self.hosts_by_pg_idx[pg_if.sw_if_index][0]
- self.vapi.l2fib_add_del(host.mac, self.bd_id, pg_if.sw_if_index,
- is_add=0)
+ self.vapi.l2fib_add_del(
+ host.mac, self.bd_id, pg_if.sw_if_index, is_add=0)
self.deleted_hosts_by_pg_idx[pg_if.sw_if_index].append(host)
del self.hosts_by_pg_idx[pg_if.sw_if_index][0]
counter += 1
for i in range(0, len(cls.pg_interfaces), 3):
cls.flows[cls.pg_interfaces[i]] = [cls.pg_interfaces[i + 1],
cls.pg_interfaces[i + 2]]
- cls.flows[cls.pg_interfaces[i + 1]] = [cls.pg_interfaces[i],
- cls.pg_interfaces[i + 2]]
- cls.flows[cls.pg_interfaces[i + 2]] = [cls.pg_interfaces[i],
- cls.pg_interfaces[i + 1]]
+ cls.flows[cls.pg_interfaces[i + 1]] = \
+ [cls.pg_interfaces[i], cls.pg_interfaces[i + 2]]
+ cls.flows[cls.pg_interfaces[i + 2]] = \
+ [cls.pg_interfaces[i], cls.pg_interfaces[i + 1]]
# Mapping between packet-generator index and lists of test hosts
cls.hosts_by_pg_idx = dict()
def create_bd_and_mac_learn(self, count, start=1):
"""
- Create required number of bridge domains with MAC learning enabled, put
- 3 l2-pg interfaces to every bridge domain and send MAC learning packets.
+ Create required number of bridge domains with MAC learning enabled,
+ put 3 l2-pg interfaces to every bridge domain and send MAC learning
+ packets.
:param int count: Number of bridge domains to be created.
:param int start: Starting number of the bridge domain ID.
"""L2XC Multi-instance Test Case HLD:
**NOTES:**
- - higher number (more than 15) of pg-l2 interfaces causes problems => only \
- 14 pg-l2 interfaces and 10 cross-connects are tested
- - jumbo packets in configuration with 14 l2-pg interfaces leads to \
- problems too
+ - higher number (more than 15) of pg-l2 interfaces causes problems => only
+ 14 pg-l2 interfaces and 10 cross-connects are tested
+ - jumbo packets in configuration with 14 l2-pg interfaces leads to
+ problems too
**config 1**
- add 14 pg-l2 interfaces
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 1**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 2**
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 2**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 3**
- send L2 MAC frames between all pairs of pg-l2 interfaces
**verify 3**
- - all packets received correctly in case of cross-connected l2-pg interfaces
+ - all packets received correctly in case of cross-connected l2-pg
+ interfaces
- no packet received in case of not cross-connected l2-pg interfaces
**config 4**
cls.flows = dict()
for i in range(len(cls.pg_interfaces)):
delta = 1 if i % 2 == 0 else -1
- cls.flows[cls.pg_interfaces[i]] = [cls.pg_interfaces[i + delta]]
+ cls.flows[cls.pg_interfaces[i]] =\
+ [cls.pg_interfaces[i + delta]]
# Mapping between packet-generator index and lists of test hosts
cls.hosts_by_pg_idx = dict()
if same_port:
self.assertEqual(packet[TCP].sport, self.tcp_port_in)
else:
- self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
+ self.assertNotEqual(
+ packet[TCP].sport, self.tcp_port_in)
self.tcp_port_out = packet[TCP].sport
elif packet.haslayer(UDP):
if same_port:
self.assertEqual(packet[UDP].sport, self.udp_port_in)
else:
- self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
+ self.assertNotEqual(
+ packet[UDP].sport, self.udp_port_in)
self.udp_port_out = packet[UDP].sport
else:
if same_port:
addr_only = 0
l_ip = socket.inet_pton(socket.AF_INET, local_ip)
e_ip = socket.inet_pton(socket.AF_INET, external_ip)
- self.vapi.snat_add_static_mapping(l_ip, e_ip, local_port, external_port,
- addr_only, vrf_id, is_add)
+ self.vapi.snat_add_static_mapping(
+ l_ip,
+ e_ip,
+ local_port,
+ external_port,
+ addr_only,
+ vrf_id,
+ is_add)
def snat_add_address(self, ip, is_add=1):
"""
self.pg3.assert_nothing_captured()
def test_multiple_inside_interfaces(self):
- """SNAT multiple inside interfaces with non-overlapping address space"""
+ """
+ SNAT multiple inside interfaces with non-overlapping address space
+ """
self.snat_add_address(self.snat_addr)
self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
for i in self.interfaces:
last_info[i.sw_if_index] = None
dst_sw_if_index = dst_if.sw_if_index
- if len(capture_pg1) != len(capture_pg2):
- self.logger.error(
- "Different number of outgoing and mirrored packets : %u != %u" %
- (len(capture_pg1), len(capture_pg2)))
- raise
+ self.AssertEqual(
+ len(capture_pg1),
+ len(capture_pg2),
+ "Different number of outgoing and mirrored packets : %u != %u" %
+ (len(capture_pg1),
+ len(capture_pg2)))
for pkt_pg1, pkt_pg2 in zip(capture_pg1, capture_pg2):
try:
ip1 = pkt_pg1[IP]
@property
def remote_mac(self):
- """MAC-address of the remote interface "connected" to this interface."""
+ """MAC-address of the remote interface "connected" to this interface"""
return self._remote_hosts[0].mac
@property
def admin_up(self):
"""Put interface ADMIN-UP."""
- self.test.vapi.sw_interface_set_flags(self.sw_if_index, admin_up_down=1)
+ self.test.vapi.sw_interface_set_flags(self.sw_if_index,
+ admin_up_down=1)
def admin_down(self):
"""Put interface ADMIN-down."""
- self.test.vapi.sw_interface_set_flags(self.sw_if_index, admin_up_down=0)
+ self.test.vapi.sw_interface_set_flags(self.sw_if_index,
+ admin_up_down=0)
def ip6_enable(self):
"""IPv6 Enable interface"""
- self.test.vapi.ip6_sw_interface_enable_disable(self.sw_if_index, enable=1)
+ self.test.vapi.ip6_sw_interface_enable_disable(self.sw_if_index,
+ enable=1)
def ip6_disable(self):
"""Put interface ADMIN-DOWN."""
- self.test.vapi.ip6_sw_interface_enable_disable(self.sw_if_index, enable=0)
+ self.test.vapi.ip6_sw_interface_enable_disable(self.sw_if_index,
+ enable=0)
def add_sub_if(self, sub_if):
"""Register a sub-interface with this interface.
class RoutePath:
- def __init__(self, nh_addr, nh_sw_if_index, nh_table_id=0, labels=[], nh_via_label=MPLS_LABEL_INVALID):
+ def __init__(
+ self,
+ nh_addr,
+ nh_sw_if_index,
+ nh_table_id=0,
+ labels=[],
+ nh_via_label=MPLS_LABEL_INVALID):
self.nh_addr = socket.inet_pton(socket.AF_INET, nh_addr)
self.nh_itf = nh_sw_if_index
self.nh_table_id = nh_table_id
def add_vpp_config(self):
for path in self.paths:
- self._test.vapi.ip_add_del_route(self.dest_addr,
- self.dest_addr_len,
- path.nh_addr,
- path.nh_itf,
- table_id=self.table_id,
- next_hop_out_label_stack=path.nh_labels,
- next_hop_n_out_labels=len(
- path.nh_labels),
- next_hop_via_label=path.nh_via_label)
+ self._test.vapi.ip_add_del_route(
+ self.dest_addr,
+ self.dest_addr_len,
+ path.nh_addr,
+ path.nh_itf,
+ table_id=self.table_id,
+ next_hop_out_label_stack=path.nh_labels,
+ next_hop_n_out_labels=len(
+ path.nh_labels),
+ next_hop_via_label=path.nh_via_label)
def remove_vpp_config(self):
for path in self.paths:
MPLS to IP Binding
"""
- def __init__(self, test, local_label, dest_addr, dest_addr_len):
+ def __init__(self, test, local_label, dest_addr, dest_addr_len):
self._test = test
self.dest_addr = socket.inet_pton(socket.AF_INET, dest_addr)
self.dest_addr_len = dest_addr_len
def add_vpp_config(self):
for path in self.paths:
- self._test.vapi.mpls_route_add_del(self.local_label,
- self.eos_bit,
- 1,
- path.nh_addr,
- path.nh_itf,
- table_id=self.table_id,
- next_hop_out_label_stack=path.nh_labels,
- next_hop_n_out_labels=len(
- path.nh_labels),
- next_hop_via_label=path.nh_via_label,
- next_hop_table_id=path.nh_table_id)
+ self._test.vapi.mpls_route_add_del(
+ self.local_label,
+ self.eos_bit,
+ 1,
+ path.nh_addr,
+ path.nh_itf,
+ table_id=self.table_id,
+ next_hop_out_label_stack=path.nh_labels,
+ next_hop_n_out_labels=len(
+ path.nh_labels),
+ next_hop_via_label=path.nh_via_label,
+ next_hop_table_id=path.nh_table_id)
def remove_vpp_config(self):
for path in self.paths:
from collections import deque
# Sphinx creates auto-generated documentation by importing the python source
-# files and collecting the docstrings from them. The NO_VPP_PAPI flag allows the
-# vpp_papi_provider.py file to be importable without having to build the whole
-# vpp api if the user only wishes to generate the test documentation.
+# files and collecting the docstrings from them. The NO_VPP_PAPI flag allows
+# the vpp_papi_provider.py file to be importable without having to build
+# the whole vpp api if the user only wishes to generate the test documentation.
do_import = True
try:
no_vpp_papi = os.getenv("NO_VPP_PAPI")
send_unicast,):
return self.api(self.papi.sw_interface_ip6nd_ra_config,
{'sw_if_index': sw_if_index,
- 'suppress' : suppress,
- 'send_unicast' : send_unicast})
+ 'suppress': suppress,
+ 'send_unicast': send_unicast})
def ip6_sw_interface_enable_disable(self, sw_if_index, enable):
"""
"""
:param is_add:
:param mask:
- :param match_n_vectors: (Default value = 1):
+ :param match_n_vectors: (Default value = 1)
:param table_index: (Default value = 0xFFFFFFFF)
:param nbuckets: (Default value = 2)
:param memory_size: (Default value = 2097152)
from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_ismaddr
from scapy.utils import inet_pton, inet_ntop
+
def is_ipv6_misc(p):
""" Is packet one of uninteresting IPv6 broadcasts? """
if p.haslayer(ICMPv6ND_RA):
self._capture_cli = "packet-generator capture pg%u pcap %s" % (
self.pg_index, self.out_path)
self._cap_name = "pcap%u" % self.sw_if_index
- self._input_cli = "packet-generator new pcap %s source pg%u name %s" % (
- self.in_path, self.pg_index, self.cap_name)
+ self._input_cli = \
+ "packet-generator new pcap %s source pg%u name %s" % (
+ self.in_path, self.pg_index, self.cap_name)
- def rotate_out_file(self):
+ def enable_capture(self):
+ """ Enable capture on this packet-generator interface"""
try:
if os.path.isfile(self.out_path):
os.rename(self.out_path,
self._out_file))
except:
pass
-
- def enable_capture(self):
- """ Enable capture on this packet-generator interface"""
- self.rotate_out_file()
# FIXME this should be an API, but no such exists atm
self.test.vapi.cli(self.capture_cli)
self._pcap_reader = None
before = len(output.res)
if filter_out_fn:
output.res = [p for p in output.res if not filter_out_fn(p)]
- removed = before - len(output.res)
+ removed = len(output.res) - before
if removed:
self.test.logger.debug(
"Filtered out %s packets from capture (returning %s)" %
based_on = "based on stored packet_infos"
if expected_count == 0:
raise Exception(
- "Internal error, expected packet count for %s is 0!" % name)
+ "Internal error, expected packet count for %s is 0!" %
+ name)
self.test.logger.debug("Expecting to capture %s (%s) packets on %s" % (
expected_count, based_on, name))
+ if expected_count == 0:
+ raise Exception(
+ "Internal error, expected packet count for %s is 0!" % name)
while remaining_time > 0:
before = time.time()
capture = self._get_capture(remaining_time, filter_out_fn)
if len(capture.res) == expected_count:
# bingo, got the packets we expected
return capture
- elif expected_count == 0:
- return None
remaining_time -= elapsed_time
if capture:
raise Exception("Captured packets mismatch, captured %s packets, "
self.test.logger.debug("Waiting for capture file %s to appear, "
"timeout is %ss" % (self.out_path, timeout))
else:
- self.test.logger.debug("Capture file %s already exists" %
- self.out_path)
+ self.test.logger.debug(
+ "Capture file %s already exists" %
+ self.out_path)
return True
while time.time() < limit:
if os.path.isfile(self.out_path):
self._pcap_reader = PcapReader(self.out_path)
break
except:
- self.test.logger.debug("Exception in scapy.PcapReader(%s): "
- "%s" % (self.out_path, format_exc()))
+ self.test.logger.debug(
+ "Exception in scapy.PcapReader(%s): "
+ "%s" %
+ (self.out_path, format_exc()))
if not self._pcap_reader:
raise Exception("Capture file %s did not appear within "
"timeout" % self.out_path)
"Packet received after %ss was filtered out" %
(time.time() - (deadline - timeout)))
else:
- self.test.logger.debug("Packet received after %fs" %
- (time.time() - (deadline - timeout)))
+ self.test.logger.debug(
+ "Packet received after %fs" %
+ (time.time() - (deadline - timeout)))
return p
time.sleep(0) # yield
self.test.logger.debug("Timeout - no packets received")
self.test.logger.info("No ARP received on port %s" %
pg_interface.name)
return
- self.rotate_out_file()
arp_reply = captured_packet.copy() # keep original for exception
# Make Dot1AD packet content recognizable to scapy
if arp_reply.type == 0x88a8:
captured_packet = pg_interface.wait_for_packet(
deadline - now, filter_out_fn=None)
except:
- self.test.logger.error("Timeout while waiting for NDP response")
+ self.test.logger.error(
+ "Timeout while waiting for NDP response")
raise
ndp_reply = captured_packet.copy() # keep original for exception
# Make Dot1AD packet content recognizable to scapy
self._local_mac = opt.lladdr
self.test.logger.debug(self.test.vapi.cli("show trace"))
# we now have the MAC we've been after
- self.rotate_out_file()
return
except:
self.test.logger.info(
- ppp("Unexpected response to NDP request:", captured_packet))
+ ppp("Unexpected response to NDP request:",
+ captured_packet))
now = time.time()
self.test.logger.debug(self.test.vapi.cli("show trace"))
- self.rotate_out_file()
raise Exception("Timeout while waiting for NDP response")
Code Review / vpp.git / commitdiff