tests/topotests/zebra_opaque/test_zebra_opaque.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 # Copyright (c) 2021 by
   5 # Donatas Abraitis <donatas.abraitis@gmail.com>
   6 #
   7
   8 """
   9 Test if Opaque Data is accessable from other daemons in Zebra
  10 """
  11
  12 import os
  13 import sys
  14 import json
  15 import pytest
  16 import functools
  17
  18 CWD = os.path.dirname(os.path.realpath(__file__))
  19 sys.path.append(os.path.join(CWD, "../"))
  20
  21 # pylint: disable=C0413
  22 from lib import topotest
  23 from lib.topogen import Topogen, TopoRouter, get_topogen
  24
  25 pytestmark = [pytest.mark.bgpd, pytest.mark.ospfd, pytest.mark.ospf6d]
  26
  27
  28 def setup_module(mod):
  29     topodef = {"s1": ("r1", "r2"), "s2": ("r3", "r4")}
  30     tgen = Topogen(topodef, mod.__name__)
  31     tgen.start_topology()
  32
  33     router_list = tgen.routers()
  34
  35     for i, (rname, router) in enumerate(router_list.items(), 1):
  36         router.load_config(
  37             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  38         )
  39         router.load_config(
  40             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  41         )
  42         router.load_config(
  43             TopoRouter.RD_OSPF, os.path.join(CWD, "{}/ospfd.conf".format(rname))
  44         )
  45         router.load_config(
  46             TopoRouter.RD_OSPF6, os.path.join(CWD, "{}/ospf6d.conf".format(rname))
  47         )
  48
  49     tgen.start_router()
  50
  51
  52 def teardown_module(mod):
  53     tgen = get_topogen()
  54     tgen.stop_topology()
  55
  56
  57 def test_zebra_opaque():
  58     tgen = get_topogen()
  59
  60     if tgen.routers_have_failure():
  61         pytest.skip(tgen.errors)
  62
  63     def _bgp_converge(router):
  64         output = json.loads(router.vtysh_cmd("show ip route 192.168.1.0/24 json"))
  65         expected = {
  66             "192.168.1.0/24": [
  67                 {
  68                     "communities": "65002:1 65002:2",
  69                     "largeCommunities": "65002:1:1 65002:2:1",
  70                 }
  71             ]
  72         }
  73         return topotest.json_cmp(output, expected)
  74
  75     def _ospf_converge(router):
  76         output = json.loads(router.vtysh_cmd("show ip route 192.168.1.0/24 json"))
  77         expected = {
  78             "192.168.1.0/24": [
  79                 {
  80                     "ospfPathType": "Intra-Area",
  81                     "ospfAreaId": "0.0.0.0",
  82                 }
  83             ]
  84         }
  85         return topotest.json_cmp(output, expected)
  86
  87     def _ospf6_converge(router):
  88         output = json.loads(router.vtysh_cmd("show ipv6 route 2001:db8:1::/64 json"))
  89         expected = {
  90             "2001:db8:1::/64": [
  91                 {
  92                     "ospfPathType": "Intra-Area",
  93                     "ospfAreaId": "0.0.0.0",
  94                 }
  95             ]
  96         }
  97         return topotest.json_cmp(output, expected)
  98
  99     router = tgen.gears["r1"]
 100     test_func = functools.partial(_bgp_converge, router)
 101     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 102     assert result is None, 'Cannot see BGP community aliases "{}"'.format(router)
 103
 104     router = tgen.gears["r3"]
 105     test_func = functools.partial(_ospf_converge, router)
 106     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 107     assert result is None, 'Cannot see OSPFv2 opaque attributes "{}"'.format(router)
 108
 109     router = tgen.gears["r3"]
 110     test_func = functools.partial(_ospf6_converge, router)
 111     success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
 112     assert result is None, 'Cannot see OSPFv3 opaque attributes "{}"'.format(router)
 113
 114
 115 if __name__ == "__main__":
 116     args = ["-s"] + sys.argv[1:]
 117     sys.exit(pytest.main(args))