tests/topotests/bgp_aspath_zero/test_bgp_aspath_zero.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # Copyright (c) 2021 by
   5 # Donatas Abraitis <donatas.abraitis@gmail.com>
   6 #
   7 # Permission to use, copy, modify, and/or distribute this software
   8 # for any purpose with or without fee is hereby granted, provided
   9 # that the above copyright notice and this permission notice appear
  10 # in all copies.
  11 #
  12 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  13 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  14 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  15 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  16 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  17 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  18 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  19 # OF THIS SOFTWARE.
  20 #
  21
  22 """
  23 Test if BGP UPDATE with AS-PATH attribute with value zero (0)
  24 is threated as withdrawal.
  25 """
  26
  27 import os
  28 import sys
  29 import json
  30 import time
  31 import pytest
  32 import functools
  33
  34 CWD = os.path.dirname(os.path.realpath(__file__))
  35 sys.path.append(os.path.join(CWD, "../"))
  36
  37 # pylint: disable=C0413
  38 from lib import topotest
  39 from lib.topogen import Topogen, TopoRouter, get_topogen
  40 from lib.topolog import logger
  41 from mininet.topo import Topo
  42
  43 pytestmark = [pytest.mark.bgpd]
  44
  45
  46 class BgpAggregatorAsnZero(Topo):
  47     def build(self, *_args, **_opts):
  48         tgen = get_topogen(self)
  49
  50         r1 = tgen.add_router("r1")
  51         peer1 = tgen.add_exabgp_peer(
  52             "peer1", ip="10.0.0.2", defaultRoute="via 10.0.0.1"
  53         )
  54
  55         switch = tgen.add_switch("s1")
  56         switch.add_link(r1)
  57         switch.add_link(peer1)
  58
  59
  60 def setup_module(mod):
  61     tgen = Topogen(BgpAggregatorAsnZero, mod.__name__)
  62     tgen.start_topology()
  63
  64     router = tgen.gears["r1"]
  65     router.load_config(TopoRouter.RD_ZEBRA, os.path.join(CWD, "r1/zebra.conf"))
  66     router.load_config(TopoRouter.RD_BGP, os.path.join(CWD, "r1/bgpd.conf"))
  67     router.start()
  68
  69     peer = tgen.gears["peer1"]
  70     peer.start(os.path.join(CWD, "peer1"), os.path.join(CWD, "exabgp.env"))
  71
  72
  73 def teardown_module(mod):
  74     tgen = get_topogen()
  75     tgen.stop_topology()
  76
  77
  78 def test_bgp_aggregator_zero():
  79     tgen = get_topogen()
  80
  81     if tgen.routers_have_failure():
  82         pytest.skip(tgen.errors)
  83
  84     def _bgp_converge():
  85         output = json.loads(
  86             tgen.gears["r1"].vtysh_cmd("show ip bgp neighbor 10.0.0.2 json")
  87         )
  88         expected = {
  89             "10.0.0.2": {
  90                 "bgpState": "Established",
  91                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 1}},
  92             }
  93         }
  94         return topotest.json_cmp(output, expected)
  95
  96     test_func = functools.partial(_bgp_converge)
  97     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
  98     assert result is None, "More than one prefix seen at r1, SHOULD be only one."
  99
 100     def _bgp_has_correct_routes_without_asn_0():
 101         output = json.loads(tgen.gears["r1"].vtysh_cmd("show ip bgp json"))
 102         expected = {"routes": {"192.168.100.101/32": [{"valid": True}]}}
 103         return topotest.json_cmp(output, expected)
 104
 105     test_func = functools.partial(_bgp_has_correct_routes_without_asn_0)
 106     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 107     assert result is None, "Failed listing 192.168.100.101/32, SHOULD be accepted."
 108
 109
 110 def test_memory_leak():
 111     "Run the memory leak test and report results."
 112     tgen = get_topogen()
 113     if not tgen.is_memleak_enabled():
 114         pytest.skip("Memory leak test/report is disabled")
 115
 116     tgen.report_memory_leaks()
 117
 118
 119 if __name__ == "__main__":
 120     args = ["-s"] + sys.argv[1:]
 121     sys.exit(pytest.main(args))