tests/topotests/bgp_aspath_zero/test_bgp_aspath_zero.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # Copyright (c) 2021 by
   6 # Donatas Abraitis <donatas.abraitis@gmail.com>
   7 #
   8
   9 """
  10 Test if BGP UPDATE with AS-PATH attribute with value zero (0)
  11 is threated as withdrawal.
  12 """
  13
  14 import os
  15 import sys
  16 import json
  17 import pytest
  18 import functools
  19
  20 CWD = os.path.dirname(os.path.realpath(__file__))
  21 sys.path.append(os.path.join(CWD, "../"))
  22
  23 # pylint: disable=C0413
  24 from lib import topotest
  25 from lib.topogen import Topogen, TopoRouter, get_topogen
  26
  27 pytestmark = [pytest.mark.bgpd]
  28
  29
  30 def build_topo(tgen):
  31     r1 = tgen.add_router("r1")
  32     peer1 = tgen.add_exabgp_peer("peer1", ip="10.0.0.2", defaultRoute="via 10.0.0.1")
  33
  34     switch = tgen.add_switch("s1")
  35     switch.add_link(r1)
  36     switch.add_link(peer1)
  37
  38
  39 def setup_module(mod):
  40     tgen = Topogen(build_topo, mod.__name__)
  41     tgen.start_topology()
  42
  43     router = tgen.gears["r1"]
  44     router.load_config(TopoRouter.RD_ZEBRA, os.path.join(CWD, "r1/zebra.conf"))
  45     router.load_config(TopoRouter.RD_BGP, os.path.join(CWD, "r1/bgpd.conf"))
  46     router.start()
  47
  48     peer = tgen.gears["peer1"]
  49     peer.start(os.path.join(CWD, "peer1"), os.path.join(CWD, "exabgp.env"))
  50
  51
  52 def teardown_module(mod):
  53     tgen = get_topogen()
  54     tgen.stop_topology()
  55
  56
  57 def test_bgp_aggregator_zero():
  58     tgen = get_topogen()
  59
  60     if tgen.routers_have_failure():
  61         pytest.skip(tgen.errors)
  62
  63     def _bgp_converge():
  64         output = json.loads(
  65             tgen.gears["r1"].vtysh_cmd("show ip bgp neighbor 10.0.0.2 json")
  66         )
  67         expected = {
  68             "10.0.0.2": {
  69                 "bgpState": "Established",
  70                 "addressFamilyInfo": {"ipv4Unicast": {"acceptedPrefixCounter": 1}},
  71             }
  72         }
  73         return topotest.json_cmp(output, expected)
  74
  75     test_func = functools.partial(_bgp_converge)
  76     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
  77     assert result is None, "More than one prefix seen at r1, SHOULD be only one."
  78
  79     def _bgp_has_correct_routes_without_asn_0():
  80         output = json.loads(tgen.gears["r1"].vtysh_cmd("show ip bgp json"))
  81         expected = {"routes": {"192.168.100.101/32": [{"valid": True}]}}
  82         return topotest.json_cmp(output, expected)
  83
  84     test_func = functools.partial(_bgp_has_correct_routes_without_asn_0)
  85     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
  86     assert result is None, "Failed listing 192.168.100.101/32, SHOULD be accepted."
  87
  88
  89 def test_memory_leak():
  90     "Run the memory leak test and report results."
  91     tgen = get_topogen()
  92     if not tgen.is_memleak_enabled():
  93         pytest.skip("Memory leak test/report is disabled")
  94
  95     tgen.report_memory_leaks()
  96
  97
  98 if __name__ == "__main__":
  99     args = ["-s"] + sys.argv[1:]
 100     sys.exit(pytest.main(args))