tests/topotests/srv6_locator/test_srv6_locator.py

   1 #!/usr/bin/env python
   2
   3 #
   4 # test_srv6_manager.py
   5 # Part of NetDEF Topology Tests
   6 #
   7 # Copyright (c) 2020 by
   8 # LINE Corporation, Hiroki Shirokura <slank.dev@gmail.com>
   9 #
  10 # Permission to use, copy, modify, and/or distribute this software
  11 # for any purpose with or without fee is hereby granted, provided
  12 # that the above copyright notice and this permission notice appear
  13 # in all copies.
  14 #
  15 # THE SOFTWARE IS PROVIDED "AS IS" AND NETDEF DISCLAIMS ALL WARRANTIES
  16 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  17 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NETDEF BE LIABLE FOR
  18 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY
  19 # DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS,
  20 # WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  21 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE
  22 # OF THIS SOFTWARE.
  23 #
  24
  25 """
  26 test_srv6_manager.py:
  27 Test for SRv6 manager on zebra
  28 """
  29
  30 import os
  31 import sys
  32 import json
  33 import pytest
  34 import functools
  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
  44 pytestmark = [pytest.mark.bgpd, pytest.mark.sharpd]
  45
  46
  47 def open_json_file(filename):
  48     try:
  49         with open(filename, "r") as f:
  50             return json.load(f)
  51     except IOError:
  52         assert False, "Could not read file {}".format(filename)
  53
  54
  55 def setup_module(mod):
  56     tgen = Topogen({None: "r1"}, mod.__name__)
  57     tgen.start_topology()
  58     for rname, router in tgen.routers().items():
  59         router.run("/bin/bash {}/{}/setup.sh".format(CWD, rname))
  60         router.load_config(
  61             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  62         )
  63         router.load_config(
  64             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  65         )
  66         router.load_config(
  67             TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
  68         )
  69     tgen.start_router()
  70
  71
  72 def teardown_module(mod):
  73     tgen = get_topogen()
  74     tgen.stop_topology()
  75
  76
  77 def test_srv6():
  78     tgen = get_topogen()
  79     if tgen.routers_have_failure():
  80         pytest.skip(tgen.errors)
  81     router = tgen.gears["r1"]
  82
  83     def _check_srv6_locator(router, expected_locator_file):
  84         logger.info("checking zebra locator status")
  85         output = json.loads(router.vtysh_cmd("show segment-routing srv6 locator json"))
  86         expected = open_json_file("{}/{}".format(CWD, expected_locator_file))
  87         return topotest.json_cmp(output, expected)
  88
  89     def _check_sharpd_chunk(router, expected_chunk_file):
  90         logger.info("checking sharpd locator chunk status")
  91         output = json.loads(router.vtysh_cmd("show sharp segment-routing srv6 json"))
  92         expected = open_json_file("{}/{}".format(CWD, expected_chunk_file))
  93         return topotest.json_cmp(output, expected)
  94
  95     def check_srv6_locator(router, expected_file):
  96         func = functools.partial(_check_srv6_locator, router, expected_file)
  97         success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
  98         assert result is None, "Failed"
  99
 100     def check_sharpd_chunk(router, expected_file):
 101         func = functools.partial(_check_sharpd_chunk, router, expected_file)
 102         success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
 103         assert result is None, "Failed"
 104
 105     # FOR DEVELOPER:
 106     # If you want to stop some specific line and start interactive shell,
 107     # please use tgen.mininet_cli() to start it.
 108
 109     logger.info("Test1 for Locator Configuration")
 110     check_srv6_locator(router, "expected_locators1.json")
 111     check_sharpd_chunk(router, "expected_chunks1.json")
 112
 113     logger.info("Test2 get chunk for locator loc1")
 114     router.vtysh_cmd("sharp srv6-manager get-locator-chunk loc1")
 115     check_srv6_locator(router, "expected_locators2.json")
 116     check_sharpd_chunk(router, "expected_chunks2.json")
 117
 118     logger.info("Test3 release chunk for locator loc1")
 119     router.vtysh_cmd("sharp srv6-manager release-locator-chunk loc1")
 120     check_srv6_locator(router, "expected_locators3.json")
 121     check_sharpd_chunk(router, "expected_chunks3.json")
 122
 123     logger.info("Test4 additional locator loc3")
 124     router.vtysh_cmd(
 125         """
 126         configure terminal
 127          segment-routing
 128           srv6
 129            locators
 130             locator loc3
 131              prefix 2001:db8:3:3::/64
 132         """
 133     )
 134     check_srv6_locator(router, "expected_locators4.json")
 135     check_sharpd_chunk(router, "expected_chunks4.json")
 136
 137     logger.info("Test5 delete locator and chunk is released automatically")
 138     router.vtysh_cmd(
 139         """
 140         configure terminal
 141          segment-routing
 142           srv6
 143            locators
 144             no locator loc1
 145         """
 146     )
 147     check_srv6_locator(router, "expected_locators5.json")
 148     check_sharpd_chunk(router, "expected_chunks5.json")
 149
 150     logger.info("Test6 delete srv6 all configuration")
 151     router.vtysh_cmd(
 152         """
 153         configure terminal
 154          segment-routing
 155           no srv6
 156         """
 157     )
 158     check_srv6_locator(router, "expected_locators6.json")
 159     check_sharpd_chunk(router, "expected_chunks6.json")
 160
 161
 162 if __name__ == "__main__":
 163     args = ["-s"] + sys.argv[1:]
 164     sys.exit(pytest.main(args))