mirror of
https://git.proxmox.com/git/mirror_frr
synced 2025-06-05 17:50:04 +00:00
tests: split notify test to regular and datastore notify tests
Signed-off-by: Christian Hopps <chopps@labn.net>
This commit is contained in:
Christian Hopps
parent
d54ab3b112
commit
61949e4f79
238
tests/topotests/mgmt_notif/test_ds_notify.py
Normal file
238
tests/topotests/mgmt_notif/test_ds_notify.py
Normal file
@ -0,0 +1,238 @@
|
||||
# -*- coding: utf-8 eval: (blacken-mode 1) -*-
|
||||
#
|
||||
# January 14 2025, Christian Hopps <chopps@labn.net>
|
||||
#
|
||||
# Copyright (c) 2025, LabN Consulting, L.L.C.
|
||||
#
|
||||
"""
|
||||
Test YANG Datastore Notifications
|
||||
"""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from lib.topogen import Topogen
|
||||
from lib.topotest import json_cmp
|
||||
from munet.testing.util import waitline
|
||||
from oper import check_kernel_32
|
||||
|
||||
pytestmark = [pytest.mark.ripd, pytest.mark.staticd, pytest.mark.mgmtd]
|
||||
|
||||
CWD = os.path.dirname(os.path.realpath(__file__))
|
||||
FE_CLIENT = CWD + "/../lib/fe_client.py"
|
||||
|
||||
|
||||
@pytest.fixture(scope="module")
|
||||
def tgen(request):
|
||||
"Setup/Teardown the environment and provide tgen argument to tests"
|
||||
|
||||
topodef = {
|
||||
"s1": ("r1", "r2"),
|
||||
}
|
||||
|
||||
tgen = Topogen(topodef, request.module.__name__)
|
||||
tgen.start_topology()
|
||||
|
||||
router_list = tgen.routers()
|
||||
for _, router in router_list.items():
|
||||
router.load_frr_config("frr.conf")
|
||||
|
||||
tgen.start_router()
|
||||
yield tgen
|
||||
tgen.stop_topology()
|
||||
|
||||
|
||||
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)
|
||||
|
||||
r1 = tgen.gears["r1"].net
|
||||
|
||||
check_kernel_32(r1, "11.11.11.11", 1, "")
|
||||
|
||||
rc, _, _ = r1.cmd_status(FE_CLIENT + " --help")
|
||||
|
||||
if rc:
|
||||
pytest.skip("No protoc or present cannot run test")
|
||||
|
||||
# Start our FE client in the background
|
||||
p = r1.popen(
|
||||
[FE_CLIENT, "--datastore", "--listen=/frr-interface:lib/interface/state"]
|
||||
)
|
||||
assert waitline(p.stderr, "Connected", timeout=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_backend_datastore_update(tgen):
|
||||
if tgen.routers_have_failure():
|
||||
pytest.skip(tgen.errors)
|
||||
|
||||
r1 = tgen.gears["r1"].net
|
||||
|
||||
check_kernel_32(r1, "11.11.11.11", 1, "")
|
||||
|
||||
be_client_path = "/usr/lib/frr/mgmtd_testc"
|
||||
rc, _, _ = r1.cmd_status(be_client_path + " --help")
|
||||
|
||||
if rc:
|
||||
pytest.skip("No mgmtd_testc")
|
||||
|
||||
# Start our BE client in the background
|
||||
p = r1.popen(
|
||||
[
|
||||
be_client_path,
|
||||
"--timeout=20",
|
||||
"--log=file:/dev/stderr",
|
||||
"--datastore",
|
||||
"--listen",
|
||||
"/frr-interface:lib/interface",
|
||||
]
|
||||
)
|
||||
assert waitline(p.stderr, "Got SUBSCR_REPLY success 1", timeout=10)
|
||||
|
||||
r1.cmd_raises("ip link set r1-eth0 mtu 1200")
|
||||
try:
|
||||
expected = json.loads(
|
||||
'{"frr-interface:lib":{"interface":[{"name":"r1-eth0","state":{"mtu":1200}}]}}'
|
||||
)
|
||||
|
||||
output, error = p.communicate(timeout=10)
|
||||
op, path, data = get_op_and_json(output)
|
||||
jsout = json.loads(data)
|
||||
result = json_cmp(jsout, expected)
|
||||
assert result is None
|
||||
finally:
|
||||
p.kill()
|
||||
r1.cmd_raises("ip link set r1-eth0 mtu 1500")
|
||||
|
||||
|
||||
def test_backend_datastore_add_delete(tgen):
|
||||
if tgen.routers_have_failure():
|
||||
pytest.skip(tgen.errors)
|
||||
|
||||
r1 = tgen.gears["r1"].net
|
||||
|
||||
check_kernel_32(r1, "11.11.11.11", 1, "")
|
||||
|
||||
be_client_path = "/usr/lib/frr/mgmtd_testc"
|
||||
rc, _, _ = r1.cmd_status(be_client_path + " --help")
|
||||
|
||||
if rc:
|
||||
pytest.skip("No mgmtd_testc")
|
||||
|
||||
# Start our BE client in the background
|
||||
p = r1.popen(
|
||||
[
|
||||
be_client_path,
|
||||
"--timeout=20",
|
||||
"--log=file:/dev/stderr",
|
||||
"--notify-count=2",
|
||||
"--datastore",
|
||||
"--listen",
|
||||
"/frr-interface:lib/interface",
|
||||
]
|
||||
)
|
||||
assert waitline(p.stderr, "Got SUBSCR_REPLY success 1", timeout=10)
|
||||
|
||||
r1.cmd_raises('vtysh -c "conf t" -c "int foobar"')
|
||||
try:
|
||||
assert waitline(
|
||||
p.stdout,
|
||||
re.escape('#OP=REPLACE: /frr-interface:lib/interface[name="foobar"]/state'),
|
||||
timeout=2,
|
||||
)
|
||||
|
||||
r1.cmd_raises('vtysh -c "conf t" -c "no int foobar"')
|
||||
assert waitline(
|
||||
p.stdout,
|
||||
re.escape('#OP=DELETE: /frr-interface:lib/interface[name="foobar"]/state'),
|
||||
timeout=2,
|
||||
)
|
||||
finally:
|
||||
p.kill()
|
||||
r1.cmd_raises('vtysh -c "conf t" -c "no int foobar"')
|
||||
|
||||
|
||||
def test_datastore_backend_filters(tgen):
|
||||
if tgen.routers_have_failure():
|
||||
pytest.skip(tgen.errors)
|
||||
|
||||
r1 = tgen.gears["r1"].net
|
||||
|
||||
check_kernel_32(r1, "11.11.11.11", 1, "")
|
||||
|
||||
rc, _, _ = r1.cmd_status(FE_CLIENT + " --help")
|
||||
if rc:
|
||||
pytest.skip("No protoc or present cannot run test")
|
||||
|
||||
# Start our FE client in the background
|
||||
p = r1.popen(
|
||||
[FE_CLIENT, "--datastore", "--listen=/frr-interface:lib/interface/state"]
|
||||
)
|
||||
assert waitline(p.stderr, "Connected", timeout=10)
|
||||
time.sleep(1)
|
||||
|
||||
try:
|
||||
output = r1.cmd_raises(
|
||||
'vtysh -c "show mgmt get-data /frr-backend:clients/client/state/notify-selectors"'
|
||||
)
|
||||
jsout = json.loads(output)
|
||||
|
||||
#
|
||||
# Verify only zebra has the notify selector as it's the only provider currently
|
||||
#
|
||||
state = {"notify-selectors": ["/frr-interface:lib/interface/state"]}
|
||||
expected = {
|
||||
"frr-backend:clients": {"client": [{"name": "zebra", "state": state}]}
|
||||
}
|
||||
|
||||
result = json_cmp(jsout, expected, exact=True)
|
||||
assert result is None
|
||||
except Exception as error:
|
||||
logging.error("got exception: %s", error)
|
||||
raise
|
||||
finally:
|
||||
p.kill()
|
||||
@ -5,17 +5,13 @@
|
||||
#
|
||||
# Copyright (c) 2024, LabN Consulting, L.L.C.
|
||||
#
|
||||
|
||||
"""
|
||||
Test YANG Notifications
|
||||
Test Traditional 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
|
||||
@ -45,99 +41,6 @@ def tgen(request):
|
||||
tgen.stop_topology()
|
||||
|
||||
|
||||
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)
|
||||
|
||||
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")
|
||||
|
||||
# Start our FE client in the background
|
||||
p = r1.popen(
|
||||
[fe_client_path, "--datastore", "--listen=/frr-interface:lib/interface"]
|
||||
)
|
||||
_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)
|
||||
@ -240,7 +143,7 @@ def test_frontend_all_notification(tgen):
|
||||
r1.cmd_raises("vtysh", stdin=conf)
|
||||
|
||||
|
||||
def test_backend_notification(tgen):
|
||||
def test_backend_yang_notification(tgen):
|
||||
if tgen.routers_have_failure():
|
||||
pytest.skip(tgen.errors)
|
||||
|
||||
|
||||
Loading…
Reference in New Issue
Block a user