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 from resources.libraries.python.VatExecutor import VatExecutor
  27
  28
  29 class TestConfig:
  30     """Contains special test configurations implemented in python for faster
  31     execution."""
  32
  33     @staticmethod
  34     def vpp_create_multiple_vxlan_ipv4_tunnels(
  35             node, node_vxlan_if, node_vlan_if, op_node, op_node_if,
  36             n_tunnels, vni_start, src_ip_start, dst_ip_start, ip_step,
  37             bd_id_start):
  38         """Create multiple VXLAN tunnel interfaces and VLAN sub-interfaces on
  39         VPP node.
  40
  41         Put each pair of VXLAN tunnel interface and VLAN sub-interface to
  42         separate bridge-domain.
  43
  44         :param node: VPP node to create VXLAN tunnel interfaces.
  45         :param node_vxlan_if: VPP node interface key to create VXLAN tunnel
  46             interfaces.
  47         :param node_vlan_if: VPP node interface key to create VLAN
  48             sub-interface.
  49         :param op_node: Opposite VPP node for VXLAN tunnel interfaces.
  50         :param op_node_if: Opposite VPP node interface key for VXLAN tunnel
  51             interfaces.
  52         :param n_tunnels: Number of tunnel interfaces to create.
  53         :param vni_start: VNI start ID.
  54         :param src_ip_start: VXLAN tunnel source IP address start.
  55         :param dst_ip_start: VXLAN tunnel destination IP address start.
  56         :param ip_step: IP address incremental step.
  57         :param bd_id_start: Bridge-domain ID start.
  58         :type node: dict
  59         :type node_vxlan_if: str
  60         :type node_vlan_if: str
  61         :type op_node: dict
  62         :type op_node_if: str
  63         :type n_tunnels: int
  64         :type vni_start: int
  65         :type src_ip_start: str
  66         :type dst_ip_start: str
  67         :type ip_step: int
  68         :type bd_id_start: int
  69         """
  70         # configure IPs, create VXLAN interfaces and VLAN sub-interfaces
  71         vxlan_count = TestConfig.vpp_create_vxlan_and_vlan_interfaces(
  72             node, node_vxlan_if, node_vlan_if, n_tunnels, vni_start,
  73             src_ip_start, dst_ip_start, ip_step
  74         )
  75
  76         # update topology with VXLAN interfaces and VLAN sub-interfaces data
  77         # and put interfaces up
  78         TestConfig.vpp_put_vxlan_and_vlan_interfaces_up(
  79             node, vxlan_count, node_vlan_if
  80         )
  81
  82         # configure bridge domains, ARPs and routes
  83         TestConfig.vpp_put_vxlan_and_vlan_interfaces_to_bridge_domain(
  84             node, node_vxlan_if, vxlan_count, op_node, op_node_if, dst_ip_start,
  85             ip_step, bd_id_start
  86         )
  87
  88     @staticmethod
  89     def vpp_create_vxlan_and_vlan_interfaces(
  90             node, node_vxlan_if, node_vlan_if, vxlan_count, vni_start,
  91             src_ip_start, dst_ip_start, ip_step):
  92         """
  93         Configure IPs, create VXLAN interfaces and VLAN sub-interfaces on VPP
  94         node.
  95
  96         :param node: VPP node.
  97         :param node_vxlan_if: VPP node interface key to create VXLAN tunnel
  98             interfaces.
  99         :param node_vlan_if: VPP node interface key to create VLAN
 100             sub-interface.
 101         :param vxlan_count: Number of tunnel interfaces to create.
 102         :param vni_start: VNI start ID.
 103         :param src_ip_start: VXLAN tunnel source IP address start.
 104         :param dst_ip_start: VXLAN tunnel destination IP address start.
 105         :param ip_step: IP address incremental step.
 106         :type node: dict
 107         :type node_vxlan_if: str
 108         :type node_vlan_if: str
 109         :type vxlan_count: int
 110         :type vni_start: int
 111         :type src_ip_start: str
 112         :type dst_ip_start: str
 113         :type ip_step: int
 114         :returns: Number of created VXLAN interfaces.
 115         :rtype: int
 116         """
 117         src_ip_start = ip_address(src_ip_start)
 118         dst_ip_start = ip_address(dst_ip_start)
 119
 120         if vxlan_count > 10:
 121             commands = list()
 122             for i in range(0, vxlan_count):
 123                 try:
 124                     src_ip = src_ip_start + i * ip_step
 125                     dst_ip = dst_ip_start + i * ip_step
 126                 except AddressValueError:
 127                     logger.warn(
 128                         u"Can't do more iterations - IP address limit "
 129                         u"has been reached."
 130                     )
 131                     vxlan_count = i
 132                     break
 133                 commands.append(
 134                     f"sw_interface_add_del_address sw_if_index "
 135                     f"{Topology.get_interface_sw_index(node, node_vxlan_if)} "
 136                     f"{src_ip}/{128 if src_ip.version == 6 else 32}\n"
 137                 )
 138                 commands.append(
 139                     f"vxlan_add_del_tunnel src {src_ip} dst {dst_ip} "
 140                     f"vni {vni_start + i}\n"
 141                 )
 142                 commands.append(
 143                     f"create_vlan_subif sw_if_index "
 144                     f"{Topology.get_interface_sw_index(node, node_vlan_if)} "
 145                     f"vlan {i + 1}\n"
 146                 )
 147             VatExecutor().write_and_execute_script(
 148                 node, u"/tmp/create_vxlan_interfaces.config", commands
 149             )
 150             return vxlan_count
 151
 152         cmd1 = u"sw_interface_add_del_address"
 153         args1 = dict(
 154             sw_if_index=InterfaceUtil.get_interface_index(node, node_vxlan_if),
 155             is_add=True,
 156             del_all=False,
 157             prefix=None
 158         )
 159         cmd2 = u"vxlan_add_del_tunnel_v3"
 160         args2 = dict(
 161             is_add=True,
 162             instance=Constants.BITWISE_NON_ZERO,
 163             src_address=None,
 164             dst_address=None,
 165             mcast_sw_if_index=Constants.BITWISE_NON_ZERO,
 166             encap_vrf_id=0,
 167             decap_next_index=Constants.BITWISE_NON_ZERO,
 168             vni=None
 169         )
 170         cmd3 = u"create_vlan_subif"
 171         args3 = dict(
 172             sw_if_index=InterfaceUtil.get_interface_index(
 173                 node, node_vlan_if),
 174             vlan_id=None
 175         )
 176
 177         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 178             for i in range(0, vxlan_count):
 179                 try:
 180                     src_ip = src_ip_start + i * ip_step
 181                     dst_ip = dst_ip_start + i * ip_step
 182                 except AddressValueError:
 183                     logger.warn(
 184                         u"Can't do more iterations - IP address limit "
 185                         u"has been reached."
 186                     )
 187                     vxlan_count = i
 188                     break
 189                 args1[u"prefix"] = IPUtil.create_prefix_object(
 190                     src_ip, 128 if src_ip_start.version == 6 else 32
 191                 )
 192                 args2[u"src_address"] = IPAddress.create_ip_address_object(
 193                     src_ip
 194                 )
 195                 args2[u"dst_address"] = IPAddress.create_ip_address_object(
 196                     dst_ip
 197                 )
 198                 args2[u"vni"] = int(vni_start) + i
 199                 args3[u"vlan_id"] = i + 1
 200                 history = bool(not 1 < i < vxlan_count - 1)
 201                 papi_exec.add(cmd1, history=history, **args1)
 202                 papi_exec.add(cmd2, history=history, **args2)
 203                 papi_exec.add(cmd3, history=history, **args3)
 204             papi_exec.get_replies()
 205
 206         return vxlan_count
 207
 208     @staticmethod
 209     def vpp_put_vxlan_and_vlan_interfaces_up(node, vxlan_count, node_vlan_if):
 210         """
 211         Update topology with VXLAN interfaces and VLAN sub-interfaces data
 212         and put interfaces up.
 213
 214         :param node: VPP node.
 215         :param vxlan_count: Number of tunnel interfaces.
 216         :param node_vlan_if: VPP node interface key where VLAN sub-interfaces
 217             have been created.
 218         :type node: dict
 219         :type vxlan_count: int
 220         :type node_vlan_if: str
 221         """
 222         if_data = InterfaceUtil.vpp_get_interface_data(node)
 223         if vxlan_count > 10:
 224             commands = list()
 225             for i in range(0, vxlan_count):
 226                 vxlan_subif_key = Topology.add_new_port(node, u"vxlan_tunnel")
 227                 vxlan_subif_name = f"vxlan_tunnel{i}"
 228                 founds = dict(vxlan=False, vlan=False)
 229                 vxlan_subif_idx = None
 230                 vlan_subif_key = Topology.add_new_port(node, u"vlan_subif")
 231                 vlan_subif_name = \
 232                     f"{Topology.get_interface_name(node, node_vlan_if)}.{i + 1}"
 233                 vlan_idx = None
 234                 for data in if_data:
 235                     if_name = data[u"interface_name"]
 236                     if not founds[u"vxlan"] and if_name == vxlan_subif_name:
 237                         vxlan_subif_idx = data[u"sw_if_index"]
 238                         founds[u"vxlan"] = True
 239                     elif not founds[u"vlan"] and if_name == vlan_subif_name:
 240                         vlan_idx = data[u"sw_if_index"]
 241                         founds[u"vlan"] = True
 242                     if founds[u"vxlan"] and founds[u"vlan"]:
 243                         break
 244                 Topology.update_interface_sw_if_index(
 245                     node, vxlan_subif_key, vxlan_subif_idx)
 246                 Topology.update_interface_name(
 247                     node, vxlan_subif_key, vxlan_subif_name)
 248                 commands.append(
 249                     f"sw_interface_set_flags sw_if_index {vxlan_subif_idx} "
 250                     f"admin-up link-up\n"
 251                 )
 252                 Topology.update_interface_sw_if_index(
 253                     node, vlan_subif_key, vlan_idx
 254                 )
 255                 Topology.update_interface_name(
 256                     node, vlan_subif_key, vlan_subif_name
 257                 )
 258                 commands.append(
 259                     f"sw_interface_set_flags sw_if_index {vlan_idx} admin-up "
 260                     f"link-up\n"
 261                 )
 262             VatExecutor().write_and_execute_script(
 263                 node, u"/tmp/put_subinterfaces_up.config", commands
 264             )
 265             return
 266
 267         cmd = u"sw_interface_set_flags"
 268         args1 = dict(
 269             sw_if_index=None,
 270             flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
 271         )
 272         args2 = dict(
 273             sw_if_index=None,
 274             flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
 275         )
 276
 277         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 278             for i in range(0, vxlan_count):
 279                 vxlan_subif_key = Topology.add_new_port(node, u"vxlan_tunnel")
 280                 vxlan_subif_name = f"vxlan_tunnel{i}"
 281                 founds = dict(vxlan=False, vlan=False)
 282                 vxlan_subif_idx = None
 283                 vlan_subif_key = Topology.add_new_port(node, u"vlan_subif")
 284                 vlan_subif_name = \
 285                     f"{Topology.get_interface_name(node, node_vlan_if)}.{i+1}"
 286                 vlan_idx = None
 287                 for data in if_data:
 288                     if not founds[u"vxlan"] \
 289                             and data[u"interface_name"] == vxlan_subif_name:
 290                         vxlan_subif_idx = data[u"sw_if_index"]
 291                         founds[u"vxlan"] = True
 292                     elif not founds[u"vlan"] \
 293                             and data[u"interface_name"] == vlan_subif_name:
 294                         vlan_idx = data[u"sw_if_index"]
 295                         founds[u"vlan"] = True
 296                     if founds[u"vxlan"] and founds[u"vlan"]:
 297                         break
 298                 Topology.update_interface_sw_if_index(
 299                     node, vxlan_subif_key, vxlan_subif_idx
 300                 )
 301                 Topology.update_interface_name(
 302                     node, vxlan_subif_key, vxlan_subif_name
 303                 )
 304                 args1[u"sw_if_index"] = vxlan_subif_idx
 305                 Topology.update_interface_sw_if_index(
 306                     node, vlan_subif_key, vlan_idx
 307                 )
 308                 Topology.update_interface_name(
 309                     node, vlan_subif_key, vlan_subif_name
 310                 )
 311                 args2[u"sw_if_index"] = vlan_idx
 312                 history = bool(not 1 < i < vxlan_count - 1)
 313                 papi_exec.add(cmd, history=history, **args1)
 314                 papi_exec.add(cmd, history=history, **args2)
 315             papi_exec.get_replies()
 316
 317     @staticmethod
 318     def vpp_put_vxlan_and_vlan_interfaces_to_bridge_domain(
 319             node, node_vxlan_if, vxlan_count, op_node, op_node_if, dst_ip_start,
 320             ip_step, bd_id_start):
 321         """
 322         Configure ARPs and routes for VXLAN interfaces and put each pair of
 323         VXLAN tunnel interface and VLAN sub-interface to separate bridge-domain.
 324
 325         :param node: VPP node.
 326         :param node_vxlan_if: VPP node interface key where VXLAN tunnel
 327             interfaces have been created.
 328         :param vxlan_count: Number of tunnel interfaces.
 329         :param op_node: Opposite VPP node for VXLAN tunnel interfaces.
 330         :param op_node_if: Opposite VPP node interface key for VXLAN tunnel
 331             interfaces.
 332         :param dst_ip_start: VXLAN tunnel destination IP address start.
 333         :param ip_step: IP address incremental step.
 334         :param bd_id_start: Bridge-domain ID start.
 335         :type node: dict
 336         :type node_vxlan_if: str
 337         :type vxlan_count: int
 338         :type op_node: dict
 339         :type op_node_if:
 340         :type dst_ip_start: str
 341         :type ip_step: int
 342         :type bd_id_start: int
 343         """
 344         dst_ip_start = ip_address(dst_ip_start)
 345
 346         if vxlan_count > 1:
 347             idx_vxlan_if = Topology.get_interface_sw_index(node, node_vxlan_if)
 348             commands = list()
 349             for i in range(0, vxlan_count):
 350                 dst_ip = dst_ip_start + i * ip_step
 351                 commands.append(
 352                     f"exec ip neighbor "
 353                     f"{Topology.get_interface_name(node, node_vxlan_if)} "
 354                     f"{dst_ip} "
 355                     f"{Topology.get_interface_mac(op_node, op_node_if)} static "
 356                     f"\n"
 357                 )
 358                 commands.append(
 359                     f"ip_route_add_del "
 360                     f"{dst_ip}/{128 if dst_ip.version == 6 else 32} count 1 "
 361                     f"via {dst_ip} sw_if_index {idx_vxlan_if}\n"
 362                 )
 363                 sw_idx_vxlan = Topology.get_interface_sw_index(
 364                     node, f"vxlan_tunnel{i + 1}"
 365                 )
 366                 commands.append(
 367                     f"sw_interface_set_l2_bridge sw_if_index {sw_idx_vxlan} "
 368                     f"bd_id {bd_id_start + i} shg 0 enable\n"
 369                 )
 370                 sw_idx_vlan = Topology.get_interface_sw_index(
 371                     node, f"vlan_subif{i + 1}"
 372                 )
 373                 commands.append(
 374                     f"sw_interface_set_l2_bridge sw_if_index {sw_idx_vlan} "
 375                     f"bd_id {bd_id_start + i} shg 0 enable\n"
 376                 )
 377             VatExecutor().write_and_execute_script(
 378                 node, u"/tmp/configure_routes_and_bridge_domains.config",
 379                 commands
 380             )
 381             return
 382
 383         cmd1 = u"ip_neighbor_add_del"
 384         neighbor = dict(
 385             sw_if_index=Topology.get_interface_sw_index(node, node_vxlan_if),
 386             flags=0,
 387             mac_address=Topology.get_interface_mac(op_node, op_node_if),
 388             ip_address=u""
 389         )
 390         args1 = dict(
 391             is_add=1,
 392             neighbor=neighbor
 393         )
 394         cmd2 = u"ip_route_add_del"
 395         kwargs = dict(
 396             interface=node_vxlan_if,
 397             gateway=str(dst_ip_start)
 398         )
 399         route = IPUtil.compose_vpp_route_structure(
 400             node, str(dst_ip_start),
 401             128 if dst_ip_start.version == 6 else 32, **kwargs
 402         )
 403         args2 = dict(
 404             is_add=1,
 405             is_multipath=0,
 406             route=route
 407         )
 408         cmd3 = u"sw_interface_set_l2_bridge"
 409         args3 = dict(
 410             rx_sw_if_index=None,
 411             bd_id=None,
 412             shg=0,
 413             port_type=0,
 414             enable=1
 415         )
 416         args4 = dict(
 417             rx_sw_if_index=None,
 418             bd_id=None,
 419             shg=0,
 420             port_type=0,
 421             enable=1
 422         )
 423
 424         with PapiSocketExecutor(node, is_async=True) as papi_exec:
 425             for i in range(0, vxlan_count):
 426                 args1[u"neighbor"][u"ip_address"] = \
 427                     str(dst_ip_start + i * ip_step)
 428                 args2[u"route"][u"prefix"][u"address"][u"un"] = \
 429                     IPAddress.union_addr(dst_ip_start + i * ip_step)
 430                 args2[u"route"][u"paths"][0][u"nh"][u"address"] = \
 431                     IPAddress.union_addr(dst_ip_start + i * ip_step)
 432                 args3[u"rx_sw_if_index"] = Topology.get_interface_sw_index(
 433                     node, f"vxlan_tunnel{i+1}"
 434                 )
 435                 args3[u"bd_id"] = int(bd_id_start+i)
 436                 args4[u"rx_sw_if_index"] = Topology.get_interface_sw_index(
 437                     node, f"vlan_subif{i+1}"
 438                 )
 439                 args4[u"bd_id"] = int(bd_id_start+i)
 440                 history = bool(not 1 < i < vxlan_count - 1)
 441                 papi_exec.add(cmd1, history=history, **args1)
 442                 papi_exec.add(cmd2, history=history, **args2)
 443                 papi_exec.add(cmd3, history=history, **args3)
 444                 papi_exec.add(cmd3, history=history, **args4)
 445             papi_exec.get_replies()