tests/topotests/zebra_seg6_route/test_zebra_seg6_route.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 #
   5 # test_zebra_seg6_route.py
   6 #
   7 # Copyright (c) 2020 by
   8 # LINE Corporation, Hiroki Shirokura <slank.dev@gmail.com>
   9 #
  10
  11 """
  12 test_zebra_seg6_route.py: Test seg6 route addition with zapi.
  13 """
  14
  15 import os
  16 import sys
  17 import pytest
  18 import json
  19 from functools import partial
  20
  21 CWD = os.path.dirname(os.path.realpath(__file__))
  22 sys.path.append(os.path.join(CWD, "../"))
  23
  24 # pylint: disable=C0413
  25 from lib import topotest
  26 from lib.topogen import Topogen, TopoRouter, get_topogen
  27 from lib.topolog import logger
  28
  29 pytestmark = [pytest.mark.sharpd]
  30
  31
  32 def open_json_file(filename):
  33     try:
  34         with open(filename, "r") as f:
  35             return json.load(f)
  36     except IOError:
  37         assert False, "Could not read file {}".format(filename)
  38
  39
  40 def setup_module(mod):
  41     tgen = Topogen({None: "r1"}, mod.__name__)
  42     tgen.start_topology()
  43     router_list = tgen.routers()
  44     for rname, router in tgen.routers().items():
  45         router.run(
  46             "/bin/bash {}".format(os.path.join(CWD, "{}/setup.sh".format(rname)))
  47         )
  48         router.load_config(
  49             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  50         )
  51         router.load_config(
  52             TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
  53         )
  54     tgen.start_router()
  55
  56
  57 def teardown_module(_mod):
  58     tgen = get_topogen()
  59     tgen.stop_topology()
  60
  61
  62 def test_zebra_seg6local_routes():
  63     tgen = get_topogen()
  64     if tgen.routers_have_failure():
  65         pytest.skip(tgen.errors)
  66     logger.info("Test for seg6local route install via ZAPI was start.")
  67     r1 = tgen.gears["r1"]
  68
  69     def check(router, dest, expected):
  70         output = json.loads(router.vtysh_cmd("show ipv6 route {} json".format(dest)))
  71         output = output.get("{}/128".format(dest))
  72         if output is None:
  73             return False
  74         return topotest.json_cmp(output, expected)
  75
  76     manifests = open_json_file(os.path.join(CWD, "{}/routes.json".format("r1")))
  77     for manifest in manifests:
  78         dest = manifest["in"]["dest"]
  79         nh = manifest["in"]["nh"]
  80         sid = manifest["in"]["sid"]
  81
  82         r1.vtysh_cmd(
  83             "sharp install seg6-routes {} nexthop-seg6 {} encap {} 1".format(
  84                 dest, nh, sid
  85             )
  86         )
  87         logger.info("CHECK {} {} {}".format(dest, nh, sid))
  88         test_func = partial(check, r1, dest, manifest["out"])
  89         success, result = topotest.run_and_expect(test_func, None, count=5, wait=1)
  90         assert result is None, "Failed"
  91
  92
  93 if __name__ == "__main__":
  94     args = ["-s"] + sys.argv[1:]
  95     sys.exit(pytest.main(args))