tests/topotests/bgp_prefix_sid/test_bgp_prefix_sid.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_bgp_prefix_sid.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by LINE Corporation
   8 # Copyright (c) 2020 by Hiroki Shirokura <slank.dev@gmail.com>
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_bgp_prefix_sid.py: Test BGP topology with EBGP on prefix-sid
  27 """
  28
  29 import json
  30 import os
  31 import sys
  32 import functools
  33 import pytest
  34
  35 CWD = os.path.dirname(os.path.realpath(__file__))
  36 sys.path.append(os.path.join(CWD, "../"))
  37
  38 # pylint: disable=C0413
  39 from lib import topotest
  40 from lib.topogen import Topogen, TopoRouter, get_topogen
  41 from lib.topolog import logger
  42 from mininet.topo import Topo
  43
  44 pytestmark = [pytest.mark.bgpd]
  45
  46
  47 class TemplateTopo(Topo):
  48     def build(self, **_opts):
  49         tgen = get_topogen(self)
  50         router = tgen.add_router("r1")
  51         switch = tgen.add_switch("s1")
  52         switch.add_link(router)
  53
  54         switch = tgen.gears["s1"]
  55         peer1 = tgen.add_exabgp_peer(
  56             "peer1", ip="10.0.0.101", defaultRoute="via 10.0.0.1"
  57         )
  58         peer2 = tgen.add_exabgp_peer(
  59             "peer2", ip="10.0.0.102", defaultRoute="via 10.0.0.1"
  60         )
  61         switch.add_link(peer1)
  62         switch.add_link(peer2)
  63
  64
  65 def setup_module(module):
  66     tgen = Topogen(TemplateTopo, module.__name__)
  67     tgen.start_topology()
  68
  69     router = tgen.gears["r1"]
  70     router.load_config(
  71         TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format("r1"))
  72     )
  73     router.load_config(
  74         TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format("r1"))
  75     )
  76     router.start()
  77
  78     logger.info("starting exaBGP on peer1")
  79     peer_list = tgen.exabgp_peers()
  80     for pname, peer in peer_list.items():
  81         peer_dir = os.path.join(CWD, pname)
  82         env_file = os.path.join(CWD, "exabgp.env")
  83         logger.info("Running ExaBGP peer")
  84         peer.start(peer_dir, env_file)
  85         logger.info(pname)
  86
  87
  88 def teardown_module(module):
  89     tgen = get_topogen()
  90     tgen.stop_topology()
  91
  92
  93 def test_r1_receive_and_advertise_prefix_sid_type1():
  94     tgen = get_topogen()
  95     router = tgen.gears["r1"]
  96
  97     def _check_type1_r1(router, prefix, remoteLabel, labelIndex):
  98         output = router.vtysh_cmd(
  99             "show bgp ipv4 labeled-unicast {} json".format(prefix)
 100         )
 101         output = json.loads(output)
 102         expected = {
 103             "prefix": prefix,
 104             "advertisedTo": {"10.0.0.101": {}, "10.0.0.102": {}},
 105             "paths": [
 106                 {
 107                     "valid": True,
 108                     "remoteLabel": remoteLabel,
 109                     "labelIndex": labelIndex,
 110                 }
 111             ],
 112         }
 113         return topotest.json_cmp(output, expected)
 114
 115     test_func = functools.partial(_check_type1_r1, router, "3.0.0.1/32", 800001, 1)
 116     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 117     assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router)
 118
 119     test_func = functools.partial(_check_type1_r1, router, "3.0.0.2/32", 800002, 2)
 120     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 121     assert result is None, 'Failed _check_type1_r1 in "{}"'.format(router)
 122
 123
 124 def exabgp_get_update_prefix(filename, afi, nexthop, prefix):
 125     with open("/tmp/peer2-received.log") as f:
 126         for line in f.readlines():
 127             output = json.loads(line)
 128             ret = output.get("neighbor")
 129             if ret is None:
 130                 continue
 131             ret = ret.get("message")
 132             if ret is None:
 133                 continue
 134             ret = ret.get("update")
 135             if ret is None:
 136                 continue
 137             ret = ret.get("announce")
 138             if ret is None:
 139                 continue
 140             ret = ret.get(afi)
 141             if ret is None:
 142                 continue
 143             ret = ret.get(nexthop)
 144             if ret is None:
 145                 continue
 146             ret = ret.get(prefix)
 147             if ret is None:
 148                 continue
 149             return output
 150         return "Not found"
 151
 152
 153 def test_peer2_receive_prefix_sid_type1():
 154     tgen = get_topogen()
 155     peer2 = tgen.gears["peer2"]
 156
 157     def _check_type1_peer2(prefix, labelindex):
 158         output = exabgp_get_update_prefix(
 159             "/tmp/peer2-received.log", "ipv4 nlri-mpls", "10.0.0.101", prefix
 160         )
 161         expected = {
 162             "type": "update",
 163             "neighbor": {
 164                 "ip": "10.0.0.1",
 165                 "message": {
 166                     "update": {
 167                         "attribute": {
 168                             "attribute-0x28-0xE0": "0x010007000000{:08x}".format(
 169                                 labelindex
 170                             )
 171                         },
 172                         "announce": {"ipv4 nlri-mpls": {"10.0.0.101": {}}},
 173                     }
 174                 },
 175             },
 176         }
 177         return topotest.json_cmp(output, expected)
 178
 179     test_func = functools.partial(_check_type1_peer2, "3.0.0.1/32", labelindex=1)
 180     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 181     assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2")
 182
 183     test_func = functools.partial(_check_type1_peer2, "3.0.0.2/32", labelindex=2)
 184     success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
 185     assert result is None, 'Failed _check_type1_peer2 in "{}"'.format("peer2")
 186
 187
 188 if __name__ == "__main__":
 189     args = ["-s"] + sys.argv[1:]
 190     ret = pytest.main(args)
 191     sys.exit(ret)