tests/topotests/zebra_seg6_route/test_zebra_seg6_route.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_zebra_seg6_route.py
   5 #
   6 # Copyright (c) 2020 by
   7 # LINE Corporation, Hiroki Shirokura <slank.dev@gmail.com>
   8 #
   9 # Permission to use, copy, modify, and/or distribute this software
  10 # for any purpose with or without fee is hereby granted, provided
  11 # that the above copyright notice and this permission notice appear
  12 # in all copies.
  13 #
  14 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  15 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  16 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  17 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  18 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  19 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  20 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  21 # OF THIS SOFTWARE.
  22 #
  23
  24 """
  25 test_zebra_seg6_route.py: Test seg6 route addition with zapi.
  26 """
  27
  28 import os
  29 import re
  30 import sys
  31 import pytest
  32 import json
  33 import platform
  34 from functools import partial
  35
  36 CWD = os.path.dirname(os.path.realpath(__file__))
  37 sys.path.append(os.path.join(CWD, "../"))
  38
  39 # pylint: disable=C0413
  40 from lib import topotest
  41 from lib.topogen import Topogen, TopoRouter, get_topogen
  42 from lib.topolog import logger
  43 from lib.common_config import shutdown_bringup_interface
  44 from mininet.topo import Topo
  45
  46 pytestmark = [pytest.mark.sharpd]
  47
  48
  49 def open_json_file(filename):
  50     try:
  51         with open(filename, "r") as f:
  52             return json.load(f)
  53     except IOError:
  54         assert False, "Could not read file {}".format(filename)
  55
  56
  57 class TemplateTopo(Topo):
  58     def build(self, **_opts):
  59         tgen = get_topogen(self)
  60         tgen.add_router("r1")
  61
  62
  63 def setup_module(mod):
  64     tgen = Topogen(TemplateTopo, mod.__name__)
  65     tgen.start_topology()
  66     router_list = tgen.routers()
  67     for rname, router in tgen.routers().items():
  68         router.run("/bin/bash {}".format(os.path.join(CWD, "{}/setup.sh".format(rname))))
  69         router.load_config(TopoRouter.RD_ZEBRA, os.path.join(CWD, '{}/zebra.conf'.format(rname)))
  70         router.load_config(TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname)))
  71     tgen.start_router()
  72
  73
  74 def teardown_module(_mod):
  75     tgen = get_topogen()
  76     tgen.stop_topology()
  77
  78
  79 def test_zebra_seg6local_routes():
  80     tgen = get_topogen()
  81     if tgen.routers_have_failure():
  82         pytest.skip(tgen.errors)
  83     logger.info("Test for seg6local route install via ZAPI was start.")
  84     r1 = tgen.gears["r1"]
  85
  86     def check(router, dest, nh, sid, expected):
  87         router.vtysh_cmd("sharp install seg6-routes {} "\
  88                          "nexthop-seg6 {} encap {} 1".format(dest, nh, sid))
  89         output = json.loads(router.vtysh_cmd("show ipv6 route {} json".format(dest)))
  90         output = output.get('{}/128'.format(dest))
  91         if output is None:
  92             return False
  93         return topotest.json_cmp(output, expected)
  94
  95     manifests = open_json_file(os.path.join(CWD, "{}/routes.json".format("r1")))
  96     for manifest in manifests:
  97         logger.info("CHECK {} {} {}".format(manifest['in']['dest'],
  98                                             manifest['in']['nh'],
  99                                             manifest['in']['sid']))
 100         test_func = partial(check, r1,
 101                             manifest['in']['dest'],
 102                             manifest['in']['nh'],
 103                             manifest['in']['sid'],
 104                             manifest['out'])
 105         success, result = topotest.run_and_expect(test_func, None, count=5, wait=1)
 106         assert result is None, 'Failed'
 107
 108
 109 if __name__ == "__main__":
 110     args = ["-s"] + sys.argv[1:]
 111     sys.exit(pytest.main(args))