#!/usr/bin/env python """ Vpp VCL tests """ import unittest import os import subprocess import signal from framework import VppTestCase, VppTestRunner, running_extended_tests, \ Worker from vpp_ip_route import VppIpTable, VppIpRoute, VppRoutePath, FibPathProto class VCLAppWorker(Worker): """ VCL Test Application Worker """ def __init__(self, build_dir, appname, args, logger, env={}): vcl_lib_dir = "%s/vpp/lib" % build_dir if "iperf" in appname: app = appname env.update({'LD_PRELOAD': "%s/libvcl_ldpreload.so" % vcl_lib_dir}) elif "sock" in appname: app = "%s/vpp/bin/%s" % (build_dir, appname) env.update({'LD_PRELOAD': "%s/libvcl_ldpreload.so" % vcl_lib_dir}) else: app = "%s/vpp/bin/%s" % (build_dir, appname) self.args = [app] + args super(VCLAppWorker, self).__init__(self.args, logger, env) class VCLTestCase(VppTestCase): """ VCL Test Class """ @classmethod def setUpClass(cls): super(VCLTestCase, cls).setUpClass() @classmethod def tearDownClass(cls): 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 Exception("Environment variable `%s' not set" % var) self.vppDebug = 'vpp_debug' in self.build_dir self.server_addr = "127.0.0.1" self.server_port = "22000" self.server_args = [self.server_port] self.server_ipv6_addr = "::1" self.server_ipv6_args = ["-6", self.server_port] self.timeout = 20 self.echo_phrase = "Hello, world! Jenny is a friend of mine." self.pre_test_sleep = 0.3 self.post_test_sleep = 0.2 if os.path.isfile("/tmp/ldp_server_af_unix_socket"): os.remove("/tmp/ldp_server_af_unix_socket") super(VCLTestCase, self).setUp() def cut_thru_setup(self): self.vapi.session_enable_disable(is_enabled=1) def cut_thru_tear_down(self): self.vapi.session_enable_disable(is_enabled=0) def cut_thru_test(self, server_app, server_args, client_app, client_args): self.env = {'VCL_API_PREFIX': self.shm_prefix, 'VCL_APP_SCOPE_LOCAL': "true"} worker_server = VCLAppWorker(self.build_dir, server_app, server_args, self.logger, self.env) worker_server.start() self.sleep(self.pre_test_sleep) worker_client = VCLAppWorker(self.build_dir, client_app, client_args, self.logger, self.env) worker_client.start() worker_client.join(self.timeout) try: self.validateResults(worker_client, worker_server, self.timeout) except Exception as error: self.fail("Failed with %s" % error) self.sleep(self.post_test_sleep) def thru_host_stack_setup(self): self.vapi.session_enable_disable(is_enabled=1) self.create_loopback_interfaces(2) table_id = 1 for i in self.lo_interfaces: i.admin_up() if table_id != 0: tbl = VppIpTable(self, table_id) tbl.add_vpp_config() i.set_table_ip4(table_id) i.config_ip4() table_id += 1 # Configure namespaces self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234, sw_if_index=self.loop0.sw_if_index) self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678, sw_if_index=self.loop1.sw_if_index) # Add inter-table routes ip_t01 = VppIpRoute(self, self.loop1.local_ip4, 32, [VppRoutePath("0.0.0.0", 0xffffffff, nh_table_id=2)], table_id=1) ip_t10 = VppIpRoute(self, self.loop0.local_ip4, 32, [VppRoutePath("0.0.0.0", 0xffffffff, nh_table_id=1)], table_id=2) ip_t01.add_vpp_config() ip_t10.add_vpp_config() self.logger.debug(self.vapi.cli("show ip fib")) def thru_host_stack_tear_down(self): for i in self.lo_interfaces: i.unconfig_ip4() i.set_table_ip4(0) i.admin_down() def thru_host_stack_ipv6_setup(self): self.vapi.session_enable_disable(is_enabled=1) self.create_loopback_interfaces(2) table_id = 1 for i in self.lo_interfaces: i.admin_up() tbl = VppIpTable(self, table_id, is_ip6=1) tbl.add_vpp_config() i.set_table_ip6(table_id) i.config_ip6() table_id += 1 # Configure namespaces self.vapi.app_namespace_add_del(namespace_id=b"1", secret=1234, sw_if_index=self.loop0.sw_if_index) self.vapi.app_namespace_add_del(namespace_id=b"2", secret=5678, sw_if_index=self.loop1.sw_if_index) # Add inter-table routes ip_t01 = VppIpRoute(self, self.loop1.local_ip6, 128, [VppRoutePath("::0", 0xffffffff, nh_table_id=2)], table_id=1) ip_t10 = VppIpRoute(self, self.loop0.local_ip6, 128, [VppRoutePath("::0", 0xffffffff, nh_table_id=1)], table_id=2) ip_t01.add_vpp_config() ip_t10.add_vpp_config() self.logger.debug(self.vapi.cli("show interface addr")) self.logger.debug(self.vapi.cli("show ip6 fib")) def thru_host_stack_ipv6_tear_down(self): for i in self.lo_interfaces: i.unconfig_ip6() i.set_table_ip6(0) i.admin_down() self.vapi.session_enable_disable(is_enabled=0) def thru_host_stack_test(self, server_app, server_args, client_app, client_args): self.env = {'VCL_API_PREFIX': self.shm_prefix, 'VCL_APP_SCOPE_GLOBAL': "true", 'VCL_APP_NAMESPACE_ID': "1", 'VCL_APP_NAMESPACE_SECRET': "1234"} worker_server = VCLAppWorker(self.build_dir, server_app, server_args, self.logger, self.env) worker_server.start() self.sleep(self.pre_test_sleep) self.env.update({'VCL_APP_NAMESPACE_ID': "2", 'VCL_APP_NAMESPACE_SECRET': "5678"}) worker_client = VCLAppWorker(self.build_dir, client_app, client_args, self.logger, self.env) worker_client.start() worker_client.join(self.timeout) try: self.validateResults(worker_client, worker_server, self.timeout) except Exception as error: self.fail("Failed with %s" % error) self.sleep(self.post_test_sleep) def validateResults(self, worker_client, worker_server, timeout): if os.path.isdir('/proc/{}'.format(worker_server.process.pid)): self.logger.info("Killing server worker process (pid %d)" % worker_server.process.pid) os.killpg(os.getpgid(worker_server.process.pid), signal.SIGTERM) worker_server.join() self.logger.info("Client worker result is `%s'" % worker_client.result) error = False if worker_client.result is None: try: error = True self.logger.error( "Timeout: %ss! Killing client worker process (pid %d)" % (timeout, worker_client.process.pid)) os.killpg(os.getpgid(worker_client.process.pid), signal.SIGKILL) worker_client.join() except OSError: self.logger.debug( "Couldn't kill client worker process") raise if error: raise Exception( "Timeout! Client worker did not finish in %ss" % timeout) self.assert_equal(worker_client.result, 0, "Binary test return code") class LDPCutThruTestCase(VCLTestCase): """ LDP Cut Thru Tests """ @classmethod def setUpClass(cls): super(LDPCutThruTestCase, cls).setUpClass() @classmethod def tearDownClass(cls): super(LDPCutThruTestCase, cls).tearDownClass() def setUp(self): super(LDPCutThruTestCase, self).setUp() self.cut_thru_setup() self.client_echo_test_args = ["-E", self.echo_phrase, "-X", self.server_addr, self.server_port] self.client_iperf3_timeout = 20 self.client_iperf3_args = ["-V4d", "-t 2", "-c", self.server_addr] self.server_iperf3_args = ["-V4d", "-s"] self.client_uni_dir_nsock_timeout = 20 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X", "-I", "2", self.server_addr, self.server_port] self.client_bi_dir_nsock_timeout = 20 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X", "-I", "2", self.server_addr, self.server_port] def tearDown(self): super(LDPCutThruTestCase, self).tearDown() self.cut_thru_tear_down() def show_commands_at_teardown(self): self.logger.debug(self.vapi.cli("show session verbose 2")) @unittest.skipUnless(running_extended_tests, "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) def test_ldp_cut_thru_iperf3(self): """ run LDP cut thru iperf3 test """ try: subprocess.check_output(['iperf3', '-v']) except subprocess.CalledProcessError: self.logger.error( "WARNING: Subprocess returned non-0 running 'iperf3 -v") self.logger.error(" 'test_ldp_cut_thru_iperf3' not run!") return except OSError as e: self.logger.error( "WARNING: Subprocess returned with OS error (%s) %s\n" " 'iperf3' is likely not installed,", e.errno, e.strerror) self.logger.error(" 'test_ldp_cut_thru_iperf3' not run!") return except Exception: self.logger.exception( "Subprocess returned non-0 running 'iperf3 -v") self.timeout = self.client_iperf3_timeout self.cut_thru_test("iperf3", self.server_iperf3_args, "iperf3", self.client_iperf3_args) @unittest.skipUnless(running_extended_tests, "part of extended tests") def test_ldp_cut_thru_uni_dir_nsock(self): """ run LDP cut thru uni-directional (multiple sockets) test """ self.timeout = self.client_uni_dir_nsock_timeout self.cut_thru_test("sock_test_server", self.server_args, "sock_test_client", self.client_uni_dir_nsock_test_args) @unittest.skipUnless(running_extended_tests, "part of extended tests") def test_ldp_cut_thru_bi_dir_nsock(self): """ run LDP cut thru bi-directional (multiple sockets) test """ self.timeout = self.client_bi_dir_nsock_timeout self.cut_thru_test("sock_test_server", self.server_args, "sock_test_client", self.client_bi_dir_nsock_test_args) class VCLCutThruTestCase(VCLTestCase): """ VCL Cut Thru Tests """ @classmethod def setUpClass(cls): super(VCLCutThruTestCase, cls).setUpClass() @classmethod def tearDownClass(cls): super(VCLCutThruTestCase, cls).tearDownClass() def setUp(self): super(VCLCutThruTestCase, self).setUp() self.cut_thru_setup() self.client_echo_test_args = ["-E", self.echo_phrase, "-X", self.server_addr, self.server_port] self.client_uni_dir_nsock_timeout = 20 self.client_uni_dir_nsock_test_args = ["-N", "1000", "-U", "-X", "-I", "2", self.server_addr, self.server_port] self.client_bi_dir_nsock_timeout = 20 self.client_bi_dir_nsock_test_args = ["-N", "1000", "-B", "-X", "-I", "2", self.server_addr, self.server_port] def tearDown(self): super(VCLCutThruTestCase, self).tearDown() def show_commands_at_teardown(self): self.logger.debug(self.vapi.cli("show session verbose 2")) def test_vcl_cut_thru_echo(self): """ run VCL cut thru echo test """ self.cut_thru_test("vcl_test_server", self.server_args, "vcl_test_client", self.client_echo_test_args) def test_vcl_cut_thru_uni_dir_nsock(self): """ run VCL cut thru uni-directional (multiple sockets) test """ self.timeout = self.client_uni_dir_nsock_ti
# Copyright (c) 2017 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

*** Variables ***
# Interface to run tests on.
| ${interface}= | ${node['interfaces']['port1']['name']}
| &{proxyarp_settings_ipv4}= | vrf-id=${0}
| ... | low-addr=192.168.1.2 | high-addr=192.168.1.10
| ${tg_to_dut_ip}= | 192.168.1.100
| ${dut_to_tg_ip}= | 192.168.1.1
| ${prefix_length}= | ${24}
| ${test_ip}= | 192.168.1.5

*** Settings ***
| Resource | resources/libraries/robot/default.robot
| Resource | resources/libraries/robot/honeycomb/honeycomb.robot
| Resource | resources/libraries/robot/honeycomb/interfaces.robot
| Resource | resources/libraries/robot/honeycomb/proxyarp.robot
| Resource | resources/libraries/robot/testing_path.robot
| Resource | resources/libraries/robot/ipv4.robot
| Resource | resources/libraries/robot/traffic.robot
| Library | resources.libraries.python.Trace
| Suite Teardown
| ... | Run Keyword If Any Tests Failed
| ... | Restart Honeycomb And VPP | ${node}
| Force Tags | HC_FUNC
| Documentation | *Honeycomb proxyARP management test suite.*

*** Test Cases ***
# TODO: Add operational data and VAT dump verification if/when avaliable
| TC01: Honeycomb can configure ipv4 proxyARP
| | [Documentation] | Check if Honeycomb can configure the proxyARP feature.
| | [Teardown] | Honeycomb removes proxyARP configuration | ${node}
| | Honeycomb configures proxyARP | ${node} | ${proxyarp_settings_ipv4}

| TC02: Honeycomb can enable proxyarp on an interface
| | [Documentation] | Check if Honeycomb can enable the proxyARP feature\
| | ... | on an interface.
| | [Teardown] | Honeycomb disables proxyARP on interface
| | ... | ${node} | ${interface}
| | Honeycomb enables proxyARP on interface | ${node} | ${interface}

| TC03: DUT sends ARP reply on behalf of another machine from the IP range
| | [Documentation]
| | ... | [Top] TG-DUT1.
| | ... | [Ref] RFC1027.
| | ... | [Cfg] On DUT1 configure interface IPv4 address and proxy ARP
| | ... | for IP range, using Honeycomb API.
| | ... | [Ver] Make TG send ARP request to DUT1 interface,
| | ... | verify if DUT1 sends correct ARP reply on behalf of machine whose
| | ... | IP is in the configured range.
| | [Teardown] | Run Keywords
| | ... | Honeycomb removes proxyARP configuration | ${node}
| | ... | AND | Honeycomb sets interface state
| | ... | ${dut_node} | ${dut_to_tg_if1} | down
| | ... | AND | Honeycomb removes interface ipv4 addresses
| | ... | ${node} | ${interface}
| | Given Path for 2-node testing is set
| | ... | ${nodes['TG']} | ${nodes['DUT1']} | ${nodes['TG']}
| | ${tg_to_dut_if1_name}= | Get interface name | ${tg_node} | ${tg_to_dut_if1}
| | And Honeycomb sets interface state | ${dut_node} | ${dut_to_tg_if1} | up
| | And Honeycomb sets interface ipv4 address with prefix | ${dut_node}