resources/libraries/python/TestConfig.py

   1 # Copyright (c) 2023 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Special test configurations library."""
  15
  16 from ipaddress import ip_address, AddressValueError
  17 from robot.api import logger
  18
  19 from resources.libraries.python.Constants import Constants
  20 from resources.libraries.python.InterfaceUtil import InterfaceUtil, \
  21     InterfaceStatusFlags
  22 from resources.libraries.python.IPAddress import IPAddress
  23 from resources.libraries.python.IPUtil import IPUtil
  24 from resources.libraries.python.PapiExecutor import PapiSocketExecutor
  25 from resources.libraries.python.topology import Topology
  26
  27
  28 class TestConfig:
  29     """Contains special test configurations implemented in python for faster
  30     execution."""
  31
  32     @staticmethod
  33     def vpp_create_multiple_vxlan_ipv4_tunnels(
  34             node, node_vxlan_if, node_vlan_if, op_node, op_node_if,
  35             n_tunnels, vni_start, src_ip_start, dst_ip_start, ip_step,
  36             bd_id_start):
  37         """Create multiple VXLAN tunnel interfaces and VLAN sub-interfaces on
  38         VPP node.
  39
  40         Put each pair of VXLAN tunnel interface and VLAN sub-interface to
  41         separate bridge-domain.
  42
  43         :param node: VPP node to create VXLAN tunnel interfaces.
  44         :param node_vxlan_if: VPP node interface key to create VXLAN tunnel
  45             interfaces.
  46         :param node_vlan_if: VPP node interface key to create VLAN
  47             sub-interface.
  48         :param op_node: Opposite VPP node for VXLAN tunnel interfaces.
  49         :param op_node_if: Opposite VPP node interface key for VXLAN tunnel
  50             interfaces.
  51         :param n_tunnels: Number of tunnel interfaces to create.
  52         :param vni_start: VNI start ID.
  53         :param src_ip_start: VXLAN tunnel source IP address start.
  54         :param dst_ip_start: VXLAN tunnel destination IP address start.
  55         :param ip_step: IP address incremental step.
  56         :param bd_id_start: Bridge-domain ID start.
  57         :type node: dict
  58         :type node_vxlan_if: str
  59         :type node_vlan_if: str
  60         :type op_node: dict
  61         :type op_node_if: str
  62         :type n_tunnels: int
  63         :type vni_start: int
  64         :type src_ip_start: str
  65         :type dst_ip_start: str
  66         :type ip_step: int
  67         :type bd_id_start: int
  68         """
  69         # configure IPs, create VXLAN interfaces and VLAN sub-interfaces
  70         vxlan_count = TestConfig.vpp_create_vxlan_and_vlan_interfaces(
  71             node, node_vxlan_if, node_vlan_if, n_tunnels, vni_start,
  72             src_ip_start, dst_ip_start, ip_step
  73         )
  74
  75         # update topology with VXLAN interfaces and VLAN sub-interfaces data
  76         # and put interfaces up
  77         TestConfig.vpp_put_vxlan_and_vlan_interfaces_up(
  78             node, vxlan_count, node_vlan_if
  79         )
  80
  81         # configure bridge domains, ARPs and routes
  82         TestConfig.vpp_put_vxlan_and_vlan_interfaces_to_bridge_domain(
  83             node, node_vxlan_if, vxlan_count, op_node, op_node_if, dst_ip_start,
  84             ip_step, bd_id_start
  85         )
  86
  87     @staticmethod
  88     def vpp_create_vxlan_and_vlan_interfaces(
  89             node, node_vxlan_if, node_vlan_if, vxlan_count, vni_start,
  90             src_ip_start, dst_ip_start, ip_step):
  91         """
  92         Configure IPs, create VXLAN interfaces and VLAN sub-interfaces on VPP
  93         node.
  94
  95         :param node: VPP node.
  96         :param node_vxlan_if: VPP node interface key to create VXLAN tunnel
  97             interfaces.
  98         :param node_vlan_if: VPP node interface key to create VLAN
  99             sub-interface.
 100         :param vxlan_count: Number of tunnel interfaces to create.
 101         :param vni_start: VNI start ID.
 102         :param src_ip_start: VXLAN tunnel source IP address start.
 103         :param dst_ip_start: VXLAN tunnel destination IP address start.
 104         :param ip_step: IP address incremental step.
 105         :type node: dict
 106         :type node_vxlan_if: str
 107         :type node_vlan_if: str
 108         :type vxlan_count: int
 109         :type vni_start: int
 110         :type src_ip_start: str
 111         :type dst_ip_start: str
 112         :type ip_step: int
 113         :returns: Number of created VXLAN interfaces.
 114         :rtype: int
 115         """
 116         src_ip_start = ip_address(src_ip_start)
 117         dst_ip_start = ip_address(dst_ip_start)
 118
 119         cmd1 = u"sw_interface_add_del_address"
 120         args1 = dict(
 121             sw_if_index=InterfaceUtil.get_interface_index(node, node_vxlan_if),
 122             is_add=True,
 123             del_all=False,
 124             prefix=None
 125         )
 126         cmd2 = u"vxlan_add_del_tunnel_v3"
 127         args2 = dict(
 128             is_add=True,
 129             instance=Constants.BITWISE_NON_ZERO,
 130             src_address=None,
 131             dst_address=None,
 132             mcast_sw_if_index=Constants.BITWISE_NON_ZERO,
 133             encap_vrf_id=0,
 134             decap_next_index=Constants.BITWISE_NON_ZERO,
 135             vni=None
 136         )
 137         cmd3 = u"create_vlan_subif"
 138         args3 = dict(
 139             sw_if_index=InterfaceUtil.get_interface_index(
 140                 node, node_vlan_if),
 141             vlan_id=None
 142         )
 143
 144         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 145             for i in range(0, vxlan_count):
 146                 try:
 147                     src_ip = src_ip_start + i * ip_step
 148                     dst_ip = dst_ip_start + i * ip_step
 149                 except AddressValueError:
 150                     logger.warn(
 151                         u"Can't do more iterations - IP address limit "
 152                         u"has been reached."
 153                     )
 154                     vxlan_count = i
 155                     break
 156                 args1[u"prefix"] = IPUtil.create_prefix_object(
 157                     src_ip, 128 if src_ip_start.version == 6 else 32
 158                 )
 159                 args2[u"src_address"] = IPAddress.create_ip_address_object(
 160                     src_ip
 161                 )
 162                 args2[u"dst_address"] = IPAddress.create_ip_address_object(
 163                     dst_ip
 164                 )
 165                 args2[u"vni"] = int(vni_start) + i
 166                 args3[u"vlan_id"] = i + 1
 167                 history = bool(not 1 < i < vxlan_count - 1)
 168                 papi_exec.add(cmd1, history=history, **args1)
 169                 papi_exec.add(cmd2, history=history, **args2)
 170                 papi_exec.add(cmd3, history=history, **args3)
 171             papi_exec.get_replies()
 172
 173         return vxlan_count
 174
 175     @staticmethod
 176     def vpp_put_vxlan_and_vlan_interfaces_up(node, vxlan_count, node_vlan_if):
 177         """
 178         Update topology with VXLAN interfaces and VLAN sub-interfaces data
 179         and put interfaces up.
 180
 181         :param node: VPP node.
 182         :param vxlan_count: Number of tunnel interfaces.
 183         :param node_vlan_if: VPP node interface key where VLAN sub-interfaces
 184             have been created.
 185         :type node: dict
 186         :type vxlan_count: int
 187         :type node_vlan_if: str
 188         """
 189         if_data = InterfaceUtil.vpp_get_interface_data(node)
 190         cmd = u"sw_interface_set_flags"
 191         args1 = dict(
 192             sw_if_index=None,
 193             flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
 194         )
 195         args2 = dict(
 196             sw_if_index=None,
 197             flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
 198         )
 199
 200         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 201             for i in range(0, vxlan_count):
 202                 vxlan_subif_key = Topology.add_new_port(node, u"vxlan_tunnel")
 203                 vxlan_subif_name = f"vxlan_tunnel{i}"
 204                 founds = dict(vxlan=False, vlan=False)
 205                 vxlan_subif_idx = None
 206                 vlan_subif_key = Topology.add_new_port(node, u"vlan_subif")
 207                 vlan_subif_name = \
 208                     f"{Topology.get_interface_name(node, node_vlan_if)}.{i+1}"
 209                 vlan_idx = None
 210                 for data in if_data:
 211                     if not founds[u"vxlan"] \
 212                             and data[u"interface_name"] == vxlan_subif_name:
 213                         vxlan_subif_idx = data[u"sw_if_index"]
 214                         founds[u"vxlan"] = True
 215                     elif not founds[u"vlan"] \
 216                             and data[u"interface_name"] == vlan_subif_name:
 217                         vlan_idx = data[u"sw_if_index"]
 218                         founds[u"vlan"] = True
 219                     if founds[u"vxlan"] and founds[u"vlan"]:
 220                         break
 221                 Topology.update_interface_sw_if_index(
 222                     node, vxlan_subif_key, vxlan_subif_idx
 223                 )
 224                 Topology.update_interface_name(
 225                     node, vxlan_subif_key, vxlan_subif_name
 226                 )
 227                 args1[u"sw_if_index"] = vxlan_subif_idx
 228                 Topology.update_interface_sw_if_index(
 229                     node, vlan_subif_key, vlan_idx
 230                 )
 231                 Topology.update_interface_name(
 232                     node, vlan_subif_key, vlan_subif_name
 233                 )
 234                 args2[u"sw_if_index"] = vlan_idx
 235                 history = bool(not 1 < i < vxlan_count - 1)
 236                 papi_exec.add(cmd, history=history, **args1)
 237                 papi_exec.add(cmd, history=history, **args2)
 238             papi_exec.get_replies()
 239
 240     @staticmethod
 241     def vpp_put_vxlan_and_vlan_interfaces_to_bridge_domain(
 242             node, node_vxlan_if, vxlan_count, op_node, op_node_if, dst_ip_start,
 243             ip_step, bd_id_start):
 244         """
 245         Configure ARPs and routes for VXLAN interfaces and put each pair of
 246         VXLAN tunnel interface and VLAN sub-interface to separate bridge-domain.
 247
 248         :param node: VPP node.
 249         :param node_vxlan_if: VPP node interface key where VXLAN tunnel
 250             interfaces have been created.
 251         :param vxlan_count: Number of tunnel interfaces.
 252         :param op_node: Opposite VPP node for VXLAN tunnel interfaces.
 253         :param op_node_if: Opposite VPP node interface key for VXLAN tunnel
 254             interfaces.
 255         :param dst_ip_start: VXLAN tunnel destination IP address start.
 256         :param ip_step: IP address incremental step.
 257         :param bd_id_start: Bridge-domain ID start.
 258         :type node: dict
 259         :type node_vxlan_if: str
 260         :type vxlan_count: int
 261         :type op_node: dict
 262         :type op_node_if:
 263         :type dst_ip_start: str
 264         :type ip_step: int
 265         :type bd_id_start: int
 266         """
 267         dst_ip_start = ip_address(dst_ip_start)
 268
 269         cmd1 = u"ip_neighbor_add_del"
 270         neighbor = dict(
 271             sw_if_index=Topology.get_interface_sw_index(node, node_vxlan_if),
 272             flags=0,
 273             mac_address=Topology.get_interface_mac(op_node, op_node_if),
 274             ip_address=u""
 275         )
 276         args1 = dict(
 277             is_add=1,
 278             neighbor=neighbor
 279         )
 280         cmd2 = u"ip_route_add_del"
 281         kwargs = dict(
 282             interface=node_vxlan_if,
 283             gateway=str(dst_ip_start)
 284         )
 285         route = IPUtil.compose_vpp_route_structure(
 286             node, str(dst_ip_start),
 287             128 if dst_ip_start.version == 6 else 32, **kwargs
 288         )
 289         args2 = dict(
 290             is_add=1,
 291             is_multipath=0,
 292             route=route
 293         )
 294         cmd3 = u"sw_interface_set_l2_bridge"
 295         args3 = dict(
 296             rx_sw_if_index=None,
 297             bd_id=None,
 298             shg=0,
 299             port_type=0,
 300             enable=1
 301         )
 302         args4 = dict(
 303             rx_sw_if_index=None,
 304             bd_id=None,
 305             shg=0,
 306             port_type=0,
 307             enable=1
 308         )
 309
 310         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 311             for i in range(0, vxlan_count):
 312                 args1[u"neighbor"][u"ip_address"] = \
 313                     str(dst_ip_start + i * ip_step)
 314                 args2[u"route"][u"prefix"][u"address"][u"un"] = \
 315                     IPAddress.union_addr(dst_ip_start + i * ip_step)
 316                 args2[u"route"][u"paths"][0][u"nh"][u"address"] = \
 317                     IPAddress.union_addr(dst_ip_start + i * ip_step)
 318                 args3[u"rx_sw_if_index"] = Topology.get_interface_sw_index(
 319                     node, f"vxlan_tunnel{i+1}"
 320                 )
 321                 args3[u"bd_id"] = int(bd_id_start+i)
 322                 args4[u"rx_sw_if_index"] = Topology.get_interface_sw_index(
 323                     node, f"vlan_subif{i+1}"
 324                 )
 325                 args4[u"bd_id"] = int(bd_id_start+i)
 326                 history = bool(not 1 < i < vxlan_count - 1)
 327                 papi_exec.add(cmd1, history=history, **args1)
 328                 papi_exec.add(cmd2, history=history, **args2)
 329                 papi_exec.add(cmd3, history=history, **args3)
 330                 # Yes, args4 go with cmd3, there is no cmd4.
 331                 papi_exec.add(cmd3, history=history, **args4)
 332             papi_exec.get_replies()