tests/topotests/srv6_locator_custom_bits_length/test_srv6_locator.py

   1 #!/usr/bin/env python
   2 # SPDX-License-Identifier: ISC
   3
   4 # Copyright (c) 2022, University of Rome Tor Vergata
   5 # Authored by Carmine Scarpitta <carmine.scarpitta@uniroma2.it>
   6 #
   7
   8 """
   9 test_srv6_manager.py:
  10 Test for SRv6 manager on zebra
  11 """
  12
  13 import os
  14 import sys
  15 import json
  16 import pytest
  17 import functools
  18
  19 CWD = os.path.dirname(os.path.realpath(__file__))
  20 sys.path.append(os.path.join(CWD, "../"))
  21
  22 # pylint: disable=C0413
  23 from lib import topotest
  24 from lib.topogen import Topogen, TopoRouter, get_topogen
  25 from lib.topolog import logger
  26
  27 pytestmark = [pytest.mark.bgpd, pytest.mark.sharpd]
  28
  29
  30 def open_json_file(filename):
  31     try:
  32         with open(filename, "r") as f:
  33             return json.load(f)
  34     except IOError:
  35         assert False, "Could not read file {}".format(filename)
  36
  37
  38 def setup_module(mod):
  39     tgen = Topogen({None: "r1"}, mod.__name__)
  40     tgen.start_topology()
  41     for rname, router in tgen.routers().items():
  42         router.run("/bin/bash {}/{}/setup.sh".format(CWD, rname))
  43         router.load_config(
  44             TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
  45         )
  46         router.load_config(
  47             TopoRouter.RD_BGP, os.path.join(CWD, "{}/bgpd.conf".format(rname))
  48         )
  49         router.load_config(
  50             TopoRouter.RD_SHARP, os.path.join(CWD, "{}/sharpd.conf".format(rname))
  51         )
  52     tgen.start_router()
  53
  54
  55 def teardown_module(mod):
  56     tgen = get_topogen()
  57     tgen.stop_topology()
  58
  59
  60 def test_srv6():
  61     tgen = get_topogen()
  62     if tgen.routers_have_failure():
  63         pytest.skip(tgen.errors)
  64     router = tgen.gears["r1"]
  65
  66     def _check_srv6_locator(router, expected_locator_file):
  67         logger.info("checking zebra locator status")
  68         output = json.loads(router.vtysh_cmd("show segment-routing srv6 locator json"))
  69         expected = open_json_file("{}/{}".format(CWD, expected_locator_file))
  70         return topotest.json_cmp(output, expected)
  71
  72     def _check_sharpd_chunk(router, expected_chunk_file):
  73         logger.info("checking sharpd locator chunk status")
  74         output = json.loads(router.vtysh_cmd("show sharp segment-routing srv6 json"))
  75         expected = open_json_file("{}/{}".format(CWD, expected_chunk_file))
  76         return topotest.json_cmp(output, expected)
  77
  78     def check_srv6_locator(router, expected_file):
  79         func = functools.partial(_check_srv6_locator, router, expected_file)
  80         success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
  81         assert result is None, "Failed"
  82
  83     def check_sharpd_chunk(router, expected_file):
  84         func = functools.partial(_check_sharpd_chunk, router, expected_file)
  85         success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
  86         assert result is None, "Failed"
  87
  88     # FOR DEVELOPER:
  89     # If you want to stop some specific line and start interactive shell,
  90     # please use tgen.mininet_cli() to start it.
  91
  92     logger.info("Test1 for Locator Configuration")
  93     check_srv6_locator(router, "expected_locators1.json")
  94     check_sharpd_chunk(router, "expected_chunks1.json")
  95
  96     logger.info("Test2 get chunk for locator loc1")
  97     router.vtysh_cmd("sharp srv6-manager get-locator-chunk loc1")
  98     check_srv6_locator(router, "expected_locators2.json")
  99     check_sharpd_chunk(router, "expected_chunks2.json")
 100
 101     logger.info("Test3 release chunk for locator loc1")
 102     router.vtysh_cmd("sharp srv6-manager release-locator-chunk loc1")
 103     check_srv6_locator(router, "expected_locators3.json")
 104     check_sharpd_chunk(router, "expected_chunks3.json")
 105
 106     logger.info("Test4 additional locator loc3")
 107     router.vtysh_cmd(
 108         """
 109         configure terminal
 110          segment-routing
 111           srv6
 112            locators
 113             locator loc3
 114              prefix 2001:db8:3::/48 block-len 32 node-len 16 func-bits 16
 115         """
 116     )
 117     check_srv6_locator(router, "expected_locators4.json")
 118     check_sharpd_chunk(router, "expected_chunks4.json")
 119
 120     logger.info("Test5 delete locator and chunk is released automatically")
 121     router.vtysh_cmd(
 122         """
 123         configure terminal
 124          segment-routing
 125           srv6
 126            locators
 127             no locator loc1
 128         """
 129     )
 130     check_srv6_locator(router, "expected_locators5.json")
 131     check_sharpd_chunk(router, "expected_chunks5.json")
 132
 133     logger.info("Test6 delete srv6 all configuration")
 134     router.vtysh_cmd(
 135         """
 136         configure terminal
 137          segment-routing
 138           no srv6
 139         """
 140     )
 141     check_srv6_locator(router, "expected_locators6.json")
 142     check_sharpd_chunk(router, "expected_chunks6.json")
 143
 144
 145 if __name__ == "__main__":
 146     args = ["-s"] + sys.argv[1:]
 147     sys.exit(pytest.main(args))