summaryrefslogtreecommitdiffstats
path: root/tests/topotests/mgmt_notif/test_notif.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/topotests/mgmt_notif/test_notif.py')
-rw-r--r--tests/topotests/mgmt_notif/test_notif.py228
1 files changed, 204 insertions, 24 deletions
diff --git a/tests/topotests/mgmt_notif/test_notif.py b/tests/topotests/mgmt_notif/test_notif.py
index e5286faae..526f051e6 100644
--- a/tests/topotests/mgmt_notif/test_notif.py
+++ b/tests/topotests/mgmt_notif/test_notif.py
@@ -10,9 +10,12 @@
Test YANG Notifications
"""
import json
+import logging
import os
+import re
import pytest
+from lib.micronet import Timeout, comm_error
from lib.topogen import Topogen
from lib.topotest import json_cmp
from oper import check_kernel_32
@@ -42,7 +45,57 @@ def tgen(request):
tgen.stop_topology()
-def test_frontend_notification(tgen):
+def myreadline(f):
+ buf = ""
+ while True:
+ # logging.debug("READING 1 CHAR")
+ c = f.read(1)
+ if not c:
+ return buf if buf else None
+ buf += c
+ # logging.debug("READ CHAR: '%s'", c)
+ if c == "\n":
+ return buf
+
+
+def _wait_output(f, regex, maxwait=120):
+ timeout = Timeout(maxwait)
+ while not timeout.is_expired():
+ # line = p.stdout.readline()
+ line = myreadline(f)
+ if not line:
+ assert None, "EOF waiting for '{}'".format(regex)
+ line = line.rstrip()
+ if line:
+ logging.debug("GOT LINE: '%s'", line)
+ m = re.search(regex, line)
+ if m:
+ return m
+ assert None, "Failed to get output matching '{}' withint {} actual {}s".format(
+ regex, maxwait, timeout.elapsed()
+ )
+
+
+def get_op_and_json(output):
+ op = ""
+ path = ""
+ data = ""
+ for line in output.split("\n"):
+ if not line:
+ continue
+ if not op:
+ m = re.match("#OP=([A-Z]*): (.*)", line)
+ if m:
+ op = m.group(1)
+ path = m.group(2)
+ continue
+ data += line + "\n"
+ if not op:
+ assert False, f"No notifcation op present in:\n{output}"
+ return op, path, data
+
+
+def test_frontend_datastore_notification(tgen):
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
@@ -50,30 +103,141 @@ def test_frontend_notification(tgen):
check_kernel_32(r1, "11.11.11.11", 1, "")
- fe_client_path = CWD + "/../lib/fe_client.py --verbose"
+ fe_client_path = CWD + "/../lib/fe_client.py"
rc, _, _ = r1.cmd_status(fe_client_path + " --help")
if rc:
pytest.skip("No protoc or present cannot run test")
- # The first notifications is a frr-ripd:authentication-type-failure
- # So we filter to avoid that, all the rest are frr-ripd:authentication-failure
- # making our test deterministic
- output = r1.cmd_raises(
- fe_client_path + " --listen /frr-ripd:authentication-failure"
+ # Start our FE client in the background
+ p = r1.popen(
+ [fe_client_path, "--datastore", "--listen=/frr-interface:lib/interface"]
)
- jsout = json.loads(output)
+ _wait_output(p.stderr, "Connected", maxwait=10)
+
+ r1.cmd_raises("ip link set r1-eth0 mtu 1200")
+
+ # {"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"if-index":2,"mtu":1200,"mtu6":1200,"speed":10000,"metric":0,"phy-address":"ba:fd:de:b5:8b:90"}}]}}
+
+ try:
+ # Wait for FE client to exit
+ output, error = p.communicate(timeout=10)
+ op, path, data = get_op_and_json(output)
+
+ assert op == "REPLACE"
+ assert path.startswith("/frr-interface:lib/interface[name='r1-eth0']/state")
+
+ jsout = json.loads(data)
+ expected = json.loads(
+ '{"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"mtu":1200}}]}}'
+ )
+ result = json_cmp(jsout, expected)
+ assert result is None
+ finally:
+ p.kill()
+ r1.cmd_raises("ip link set r1-eth0 mtu 1500")
+
+
+def test_frontend_notification(tgen):
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ r1 = tgen.gears["r1"].net
+
+ check_kernel_32(r1, "11.11.11.11", 1, "")
+
+ fe_client_path = CWD + "/../lib/fe_client.py"
+ rc, _, _ = r1.cmd_status(fe_client_path + " --help")
+
+ if rc:
+ pytest.skip("No protoc or present cannot run test")
+
+ # Update config to non-matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string bar
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
+
+ try:
+ output = r1.cmd_raises(
+ fe_client_path + " --listen /frr-ripd:authentication-failure"
+ )
- expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
- result = json_cmp(jsout, expected)
- assert result is None
+ jsout = json.loads(output)
+ expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
+ result = json_cmp(jsout, expected)
+ assert result is None
- output = r1.cmd_raises(fe_client_path + " --use-protobuf --listen")
- jsout = json.loads(output)
+ output = r1.cmd_raises(
+ fe_client_path + " --use-protobuf --listen /frr-ripd:authentication-failure"
+ )
+ jsout = json.loads(output)
+ expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
+ result = json_cmp(jsout, expected)
+ assert result is None
+ finally:
+ # Update config to matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string foo
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
- expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
- result = json_cmp(jsout, expected)
- assert result is None
+
+def test_frontend_all_notification(tgen):
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ r1 = tgen.gears["r1"].net
+
+ check_kernel_32(r1, "11.11.11.11", 1, "")
+
+ fe_client_path = CWD + "/../lib/fe_client.py"
+ rc, _, _ = r1.cmd_status(fe_client_path + " --help")
+
+ if rc:
+ pytest.skip("No protoc or present cannot run test")
+
+ # Update config to non-matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string bar
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
+
+ try:
+ # The first notifications is a frr-ripd:authentication-type-failure
+ # All the rest are frr-ripd:authentication-failure so we check for both.
+ output = r1.cmd_raises(fe_client_path + " --listen /")
+ jsout = json.loads(output)
+ expected = {
+ "frr-ripd:authentication-type-failure": {"interface-name": "r1-eth0"}
+ }
+ result = json_cmp(jsout, expected)
+ if result is not None:
+ expected = {
+ "frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}
+ }
+ result = json_cmp(jsout, expected)
+ assert result is None
+
+ output = r1.cmd_raises(fe_client_path + " --use-protobuf --listen /")
+ jsout = json.loads(output)
+ expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
+ result = json_cmp(jsout, expected)
+ assert result is None
+ finally:
+ # Update config to matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string foo
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
def test_backend_notification(tgen):
@@ -90,12 +254,28 @@ def test_backend_notification(tgen):
if rc:
pytest.skip("No mgmtd_testc")
- output = r1.cmd_raises(
- be_client_path + " --timeout 20 --log file:mgmt_testc.log --listen /frr-ripd"
- )
-
- jsout = json.loads(output)
+ # Update config to non-matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string bar
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
- expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
- result = json_cmp(jsout, expected)
- assert result is None
+ try:
+ output = r1.cmd_raises(
+ be_client_path
+ + " --timeout 20 --log file:mgmt_testc.log --listen /frr-ripd"
+ )
+ jsout = json.loads(output)
+ expected = {"frr-ripd:authentication-failure": {"interface-name": "r1-eth0"}}
+ result = json_cmp(jsout, expected)
+ assert result is None
+ finally:
+ # Update config to matching authentication.
+ conf = """
+ conf t
+ interface r1-eth0
+ ip rip authentication string foo
+ """
+ r1.cmd_raises("vtysh", stdin=conf)
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-18 11:22:52 +0100