tests: add ospf api router ID topotest

Signed-off-by: Christian Hopps <chopps@labn.net>

2 files changed, 86 insertions, 3 deletions

diff --git a/tests/topotests/ospfapi/ctester.py b/tests/topotests/ospfapi/ctester.py
index 243fc0613..8fd202a02 100755
--- a/tests/topotests/ospfapi/ctester.py
+++ b/tests/topotests/ospfapi/ctester.py
@@ -42,6 +42,35 @@ sys.path[0:0] = [CLIENTDIR]
 import ospfclient as api  # pylint: disable=E0401 # noqa: E402
 
 
+async def do_monitor(c, args):
+    cv = asyncio.Condition()
+
+    async def cb(new_router_id, _):
+        assert new_router_id == c.router_id
+        logging.info("NEW ROUTER ID: %s", new_router_id)
+        sys.stdout.flush()
+        async with cv:
+            cv.notify_all()
+
+    logging.debug("API using monitor router ID callback")
+    await c.monitor_router_id(callback=cb)
+
+    for check in args.monitor:
+        logging.info("Waiting for %s", check)
+
+        while True:
+            async with cv:
+                got = c.router_id
+                if str(check) == str(got):
+                    break
+                logging.debug("expected '%s' != '%s'\nwaiting on notify", check, got)
+                await cv.wait()
+
+        logging.info("SUCCESS: %s", check)
+        print("SUCCESS: {}".format(check))
+        sys.stdout.flush()
+
+
 async def do_wait(c, args):
     cv = asyncio.Condition()
 
@@ -51,7 +80,7 @@ async def do_wait(c, args):
         async with cv:
             cv.notify_all()
 
-    logging.debug("API using callback")
+    logging.debug("API using monitor reachable callback")
     await c.monitor_reachable(callback=cb)
 
     for w in args.wait:
@@ -81,6 +110,8 @@ async def async_main(args):
             c._handle_msg_loop()  # pylint: disable=W0212
         )
 
+    if args.monitor:
+        await do_monitor(c, args)
     if args.wait:
         await do_wait(c, args)
     return 0
@@ -88,6 +119,9 @@ async def async_main(args):
 
 def main(*args):
     ap = argparse.ArgumentParser(args)
+    ap.add_argument(
+        "--monitor", action="append", help="monitor and wait for this router ID"
+    )
     ap.add_argument("--server", default="localhost", help="OSPF API server")
     ap.add_argument(
         "--wait", action="append", help="wait for comma-sep set of reachable routers"
diff --git a/tests/topotests/ospfapi/test_ospf_clientapi.py b/tests/topotests/ospfapi/test_ospf_clientapi.py
index dca91412d..631c8025b 100644
--- a/tests/topotests/ospfapi/test_ospf_clientapi.py
+++ b/tests/topotests/ospfapi/test_ospf_clientapi.py
@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 eval: (blacken-mode 1) -*-
 #
-# Copyright (c) 2021, LabN Consulting, L.L.C.
+# Copyright (c) 2021-2022, LabN Consulting, L.L.C.
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU General Public License
@@ -61,7 +61,10 @@ assert os.path.exists(
 def _tgen(request):
     "Setup/Teardown the environment and provide tgen argument to tests"
     nrouters = request.param
-    topodef = {f"sw{i}": (f"r{i}", f"r{i+1}") for i in range(1, nrouters)}
+    if nrouters == 1:
+        topodef = {"sw1:": ("r1",)}
+    else:
+        topodef = {f"sw{i}": (f"r{i}", f"r{i+1}") for i in range(1, nrouters)}
 
     tgen = Topogen(topodef, request.module.__name__)
     tgen.start_topology()
@@ -183,6 +186,52 @@ def test_ospf_reachability(tgen):
     _test_reachability(tgen, testbin)
 
 
+def _test_router_id(tgen, testbin):
+    r1 = tgen.gears["r1"]
+    waitlist = [
+        "192.168.0.1",
+        "1.1.1.1",
+        "192.168.0.1",
+    ]
+
+    mon_args = [f"--monitor={x}" for x in waitlist]
+
+    p = None
+    try:
+        step("router id: check for initial router id")
+        p = r1.popen(
+            ["/usr/bin/timeout", "120", testbin, "-v", *mon_args],
+            encoding=None,  # don't buffer
+            stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+        )
+        _wait_output(p, "SUCCESS: {}".format(waitlist[0]))
+
+        step("router id: check for modified router id")
+        r1.vtysh_multicmd("conf t\nrouter ospf\nospf router-id 1.1.1.1")
+        _wait_output(p, "SUCCESS: {}".format(waitlist[1]))
+
+        step("router id: check for restored router id")
+        r1.vtysh_multicmd("conf t\nrouter ospf\nospf router-id 192.168.0.1")
+        _wait_output(p, "SUCCESS: {}".format(waitlist[2]))
+    except Exception as error:
+        logging.error("ERROR: %s", error)
+        raise
+    finally:
+        if p:
+            p.terminate()
+            p.wait()
+
+
+@pytest.mark.parametrize("tgen", [2], indirect=True)
+def test_ospf_router_id(tgen):
+    testbin = os.path.join(TESTDIR, "ctester.py")
+    rc, o, e = tgen.gears["r1"].net.cmd_status([testbin, "--help"])
+    logging.info("%s --help: rc: %s stdout: '%s' stderr: '%s'", testbin, rc, o, e)
+    _test_router_id(tgen, testbin)
+
+
 def _test_add_data(tgen, apibin):
     "Test adding opaque data to domain"