tests/topotests/ospfapi/ctester.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
   3 # SPDX-License-Identifier: MIT
   4 #
   5 # January 17 2022, Christian Hopps <chopps@labn.net>
   6 #
   7 # Copyright 2022, LabN Consulting, L.L.C.
   8 import argparse
   9 import asyncio
  10 import logging
  11 import os
  12 import sys
  13
  14 CWD = os.path.dirname(os.path.realpath(__file__))
  15
  16 CLIENTDIR = os.path.abspath(os.path.join(CWD, "../../../ospfclient"))
  17 if not os.path.exists(CLIENTDIR):
  18     CLIENTDIR = os.path.join(CWD, "/usr/lib/frr")
  19 assert os.path.exists(
  20     os.path.join(CLIENTDIR, "ospfclient.py")
  21 ), "can't locate ospfclient.py"
  22
  23 sys.path[0:0] = [CLIENTDIR]
  24
  25 import ospfclient as api  # pylint: disable=E0401 # noqa: E402
  26
  27
  28 async def do_monitor(c, args):
  29     cv = asyncio.Condition()
  30
  31     async def cb(new_router_id, _):
  32         assert new_router_id == c.router_id
  33         logging.info("NEW ROUTER ID: %s", new_router_id)
  34         sys.stdout.flush()
  35         async with cv:
  36             cv.notify_all()
  37
  38     logging.debug("API using monitor router ID callback")
  39     await c.monitor_router_id(callback=cb)
  40
  41     for check in args.monitor:
  42         logging.info("Waiting for %s", check)
  43
  44         while True:
  45             async with cv:
  46                 got = c.router_id
  47                 if str(check) == str(got):
  48                     break
  49                 logging.debug("expected '%s' != '%s'\nwaiting on notify", check, got)
  50                 await cv.wait()
  51
  52         logging.info("SUCCESS: %s", check)
  53         print("SUCCESS: {}".format(check))
  54         sys.stdout.flush()
  55
  56
  57 async def do_wait(c, args):
  58     cv = asyncio.Condition()
  59
  60     async def cb(added, removed):
  61         logging.debug("callback: added: %s removed: %s", added, removed)
  62         sys.stdout.flush()
  63         async with cv:
  64             cv.notify_all()
  65
  66     logging.debug("API using monitor reachable callback")
  67     await c.monitor_reachable(callback=cb)
  68
  69     for w in args.wait:
  70         check = ",".join(sorted(list(w.split(","))))
  71         logging.info("Waiting for %s", check)
  72
  73         while True:
  74             async with cv:
  75                 got = ",".join(sorted([str(x) for x in c.reachable_routers]))
  76                 if check == got:
  77                     break
  78                 logging.debug("expected '%s' != '%s'\nwaiting on notify", check, got)
  79                 await cv.wait()
  80
  81         logging.info("SUCCESS: %s", check)
  82         print("SUCCESS: {}".format(check))
  83         sys.stdout.flush()
  84
  85
  86 async def async_main(args):
  87     c = api.OspfOpaqueClient(args.server)
  88     await c.connect()
  89     if sys.version_info[1] > 6:
  90         asyncio.create_task(c._handle_msg_loop())  # pylint: disable=W0212
  91     else:
  92         asyncio.get_event_loop().create_task(
  93             c._handle_msg_loop()  # pylint: disable=W0212
  94         )
  95
  96     if args.monitor:
  97         await do_monitor(c, args)
  98     if args.wait:
  99         await do_wait(c, args)
 100     return 0
 101
 102
 103 def main(*args):
 104     ap = argparse.ArgumentParser(args)
 105     ap.add_argument(
 106         "--monitor", action="append", help="monitor and wait for this router ID"
 107     )
 108     ap.add_argument("--server", default="localhost", help="OSPF API server")
 109     ap.add_argument(
 110         "--wait", action="append", help="wait for comma-sep set of reachable routers"
 111     )
 112     ap.add_argument("-v", "--verbose", action="store_true", help="be verbose")
 113     args = ap.parse_args()
 114
 115     level = logging.DEBUG if args.verbose else logging.INFO
 116     logging.basicConfig(
 117         level=level, format="%(asctime)s %(levelname)s: TESTER: %(name)s: %(message)s"
 118     )
 119
 120     # We need to flush this output to stdout right away
 121     h = logging.StreamHandler(sys.stdout)
 122     h.flush = sys.stdout.flush
 123     f = logging.Formatter("%(asctime)s %(name)s: %(levelname)s: %(message)s")
 124     h.setFormatter(f)
 125     logger = logging.getLogger("ospfclient")
 126     logger.addHandler(h)
 127     logger.propagate = False
 128
 129     logging.info("ctester: starting")
 130     sys.stdout.flush()
 131
 132     status = 3
 133     try:
 134         if sys.version_info[1] > 6:
 135             status = asyncio.run(async_main(args))
 136         else:
 137             loop = asyncio.get_event_loop()
 138             try:
 139                 status = loop.run_until_complete(async_main(args))
 140             finally:
 141                 loop.close()
 142     except KeyboardInterrupt:
 143         logging.info("Exiting, received KeyboardInterrupt in main")
 144     except Exception as error:
 145         logging.info("Exiting, unexpected exception %s", error, exc_info=True)
 146     else:
 147         logging.info("api: clean exit")
 148
 149     return status
 150
 151
 152 if __name__ == "__main__":
 153     exit_status = main()
 154     sys.exit(exit_status)