.. _unittest: https://docs.python.org/2/library/unittest.html .. _TestCase: https://docs.python.org/2/library/unittest.html#unittest.TestCase .. _AssertionError: https://docs.python.org/2/library/exceptions.html#exceptions.AssertionError .. _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 .. _process: https://docs.python.org/2/library/multiprocessing.html#the-process-class .. _pipes: https://docs.python.org/2/library/multiprocessing.html#multiprocessing.Pipe .. _managed: https://docs.python.org/2/library/multiprocessing.html#managers .. |vtf| replace:: VPP Test Framework |vtf| ===== .. contents:: :local: :depth: 1 Overview ######## The goal of the |vtf| is to ease writing, running and debugging unit tests for the VPP. For this, python was chosen as a high level language allowing rapid development with scapy_ providing the necessary tool for creating and dissecting packets. Anatomy of a test case ###################### Python's unittest_ is used as the base framework upon which the VPP test framework is built. A test suite in the |vtf| consists of multiple classes derived from `VppTestCase`, which is itself derived from TestCase_. The test class defines one or more test functions, which act as test cases. Function flow when running a test case is: 1. `setUpClass `: This function is called once for each test class, allowing a one-time test setup to be executed. If this functions throws an exception, none of the test functions are executed. 2. `setUp `: The setUp function runs before each of the test functions. If this function throws an exception other than AssertionError_ or SkipTest_, then this is considered an error, not a test failure. 3. *test_*: 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. Multiple test_ methods can exist in a test case. 4. `tearDown `: The tearDown function is called after each test function with the purpose of doing partial cleanup. 5. `tearDownClass `: 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 Parallel test execution ####################### |vtf| test suites can be run in parallel. Each test suite is executed in a separate process spawned by Python multiprocessing process_. The results from child test suites are sent to parent through pipes_, which are aggregated and summarized at the end of the run. Stdout, stderr and logs logged in child processes are redirected to individual parent managed_ queues. The data from these queues are then emitted to stdout of the parent process in the order the test suites have finished. In case there are no finished test suites (such as at the beginning of the run), the data from last started test suite are emitted in real time. To enable parallel test run, specify the number of parallel processes: .. code-block:: shell make test TEST_JOBS=n # at most n processes will be spawned make test TEST_JOBS=auto # chosen based on the number of cores # and the size of shared memory 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 f

---
name: Address Resolution Protocol
maintainer: Neale Ranns <nranns@cisco.com>
features:
  - ARP responder

description: "An implementation of the Address resolution protocol (ARP) as described in RFC826"
state: production
properties: [API, CLI, MULTITHREAD]

udp = packet[UDP] # convert the payload to packet info object payload_info = self.payload_to_info(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'.