tests/topotests/zebra_opaque/test_zebra_opaque.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2021 by
   4 # Donatas Abraitis <donatas.abraitis@gmail.com>
   5 #
   6 # Permission to use, copy, modify, and/or distribute this software
   7 # for any purpose with or without fee is hereby granted, provided
   8 # that the above copyright notice and this permission notice appear
   9 # in all copies.
  10 #
  11 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  12 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  13 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  14 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  15 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  16 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  17 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  18 # OF THIS SOFTWARE.
  19 #
  20
  21 """
  22 Test if Opaque Data is accessable from other daemons in Zebra
  23 """
  24
  25 import os
  26 import sys
  27 import json
  28 import time
  29 import pytest
  30 import functools
  31
  32 CWD = os.path.dirname(os.path.realpath(__file__))
  33 sys.path.append(os.path.join(CWD, "../"))
  34
  35 # pylint: disable=C0413
  36 from lib import topotest
  37 from lib.topogen import Topogen, TopoRouter, get_topogen
  38 from lib.topolog import logger
  39 from mininet.topo import Topo
  40
  41 pytestmark = [pytest.mark.bgpd]
  42
  43
  44 class TemplateTopo(Topo):
  45     def build(self, *_args, **_opts):
  46         tgen = get_topogen(self)
  47
  48         for routern in range(1, 3):
  49             tgen.add_router("r{}".format(routern))
  50
  51         switch = tgen.add_switch("s1")
  52         switch.add_link(tgen.gears["r1"])
  53         switch.add_link(tgen.gears["r2"])
  54
  55
  56 def setup_module(mod):
  57     tgen = Topogen(TemplateTopo, mod.__name__)
  58     tgen.start_topology()
  59
  60     router_list = tgen.routers()
  61
  62     for i, (rname, router) in enumerate(router_list.items(), 1):
  63         router.load_config(
  64             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  65         )
  66         router.load_config(
  67             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  68         )
  69
  70     tgen.start_router()
  71
  72
  73 def teardown_module(mod):
  74     tgen = get_topogen()
  75     tgen.stop_topology()
  76
  77
  78 def test_zebra_opaque():
  79     tgen = get_topogen()
  80
  81     if tgen.routers_have_failure():
  82         pytest.skip(tgen.errors)
  83
  84     router = tgen.gears["r1"]
  85
  86     def _bgp_converge(router):
  87         output = json.loads(router.vtysh_cmd("show ip route 192.168.1.0/24 json"))
  88         expected = {
  89             "192.168.1.0/24": [
  90                 {
  91                     "communities": "65002:1 65002:2",
  92                     "largeCommunities": "65002:1:1 65002:2:1",
  93                 }
  94             ]
  95         }
  96         return topotest.json_cmp(output, expected)
  97
  98     test_func = functools.partial(_bgp_converge, router)
  99     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 100
 101     assert result is None, 'Cannot see BGP community aliases "{}"'.format(router)
 102
 103
 104 if __name__ == "__main__":
 105     args = ["-s"] + sys.argv[1:]
 106     sys.exit(pytest.main(args))