--- /dev/null
+log timestamp precision 6
+log file frr.log
+
+debug northbound notifications
+debug northbound libyang
+debug northbound events
+debug northbound callbacks
+debug mgmt backend datastore frontend transaction
+debug mgmt client backend
+debug mgmt client frontend
+
+interface r4-eth0
+ ip address 101.0.0.4/24
+ ipv6 address 2101::4/64
+exit
+
+ip route 11.0.0.0/24 101.0.0.1
+ip route 12.0.0.0/24 101.0.0.2
+
+ipv6 route 2012::/48 2101::2
+ipv6 route 2013::/48 2101::3
"""
import datetime
-import ipaddress
import logging
-import math
import os
-import re
import pytest
-from lib.common_config import retry, step
+from lib.common_config import step
from lib.topogen import Topogen, TopoRouter
-from lib.topolog import logger
from munet.base import Timeout
+from util import check_kernel, check_vtysh_up, write_big_route_conf
CWD = os.path.dirname(os.path.realpath(__file__))
pytestmark = [pytest.mark.staticd]
-def get_ip_networks(super_prefix, count):
- count_log2 = math.log(count, 2)
- if count_log2 != int(count_log2):
- count_log2 = int(count_log2) + 1
- else:
- count_log2 = int(count_log2)
- network = ipaddress.ip_network(super_prefix)
- return tuple(network.subnets(count_log2))[0:count]
-
-
track = Timeout(0)
-ROUTE_COUNT = 5000
+ROUTE_COUNT = 2500
ROUTE_RANGE = [None, None]
-def write_big_route_conf(rtr, super_prefix, count):
- start = None
- end = None
- with open(f"{CWD}/{rtr.name}/big.conf", "w+", encoding="ascii") as f:
- for net in get_ip_networks(super_prefix, count):
- end = net
- if not start:
- start = net
- f.write(f"ip route {net} lo\n")
-
- return start, end
-
-
@pytest.fixture(scope="module")
def tgen(request):
"Setup/Teardown the environment and provide tgen argument to tests"
tgen = Topogen(topodef, request.module.__name__)
tgen.start_topology()
- start, end = write_big_route_conf(tgen.gears["r1"].net, "10.0.0.0/8", ROUTE_COUNT)
+ confpath = f"{tgen.gears['r1'].gearlogdir}/r1-late-big.conf"
+ start, end = write_big_route_conf("10.0.0.0/8", ROUTE_COUNT, confpath)
ROUTE_RANGE[0] = start
ROUTE_RANGE[1] = end
# configure mgmtd using current mgmtd config file
tgen.gears["r1"].load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
- tgen.gears["r1"].load_config(TopoRouter.RD_MGMTD, "big.conf")
+ tgen.gears["r1"].load_config(TopoRouter.RD_MGMTD, confpath)
track.started_on = datetime.datetime.now()
tgen.stop_topology()
-@retry(retry_timeout=3, initial_wait=0.1)
-def check_kernel(r1, prefix, expected=True):
- net = ipaddress.ip_network(prefix)
- if net.version == 6:
- kernel = r1.net.cmd_nostatus("ip -6 route show", warn=not expected)
- else:
- kernel = r1.net.cmd_nostatus("ip -4 route show", warn=not expected)
-
- logger.debug("checking kernel routing table:\n%s", kernel)
- route = f"{str(net)}(?: nhid [0-9]+)?.*proto (static|196)"
- m = re.search(route, kernel)
- if expected and not m:
- return f"Failed to find \n'{route}'\n in \n'{kernel}'"
- elif not expected and m:
- return f"Failed found \n'{route}'\n in \n'{kernel}'"
- return None
-
-
-def test_staticd_late_start(tgen):
+def test_staticd_latestart(tgen):
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
step(f"Verifying {ROUTE_COUNT} startup routes are present")
- timeo = Timeout(30)
- for remaining in timeo:
- rc, o, e = r1.net.cmd_status("vtysh -c 'show version'")
- if not rc:
- break
- print("nogo: ", rc, o, e)
- assert not timeo.is_expired()
+ check_vtysh_up(r1)
logging.info("r1: vtysh connected after %ss", track.elapsed())
result = check_kernel(r1, ROUTE_RANGE[0], retry_timeout=20)
"""
-import ipaddress
-import re
-
import pytest
-from lib.common_config import retry, step
+from lib.common_config import step
from lib.topogen import Topogen, TopoRouter
-from lib.topolog import logger
+from util import check_kernel
# pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
pytestmark = [pytest.mark.staticd]
"Setup/Teardown the environment and provide tgen argument to tests"
topodef = {
- "s1": ("r1", "r2", "r3"),
+ "s1": ("r1", "r2", "r3", "r4"),
}
tgen = Topogen(topodef, request.module.__name__)
# configure mgmtd using backup config file `zebra.conf`
tgen.gears["r3"].load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
+ # configure mgmtd using current mgmtd config file
+ tgen.gears["r4"].load_frr_config("frr.conf")
+
tgen.start_router()
yield tgen
tgen.stop_topology()
-@retry(retry_timeout=3, initial_wait=0.1)
-def check_kernel(r1, prefix, expected=True):
- net = ipaddress.ip_network(prefix)
- if net.version == 6:
- kernel = r1.net.cmd_nostatus("ip -6 route show", warn=not expected)
- else:
- kernel = r1.net.cmd_nostatus("ip -4 route show", warn=not expected)
-
- logger.debug("checking kernel routing table:\n%s", kernel)
- route = f"{str(net)}(?: nhid [0-9]+)?.*proto (static|196)"
- m = re.search(route, kernel)
- if expected and not m:
- return f"Failed to find \n'{route}'\n in \n'{kernel}'"
- elif not expected and m:
- return f"Failed found \n'{route}'\n in \n'{kernel}'"
- return None
-
-
-def test_staticd_late_start(tgen):
+def test_staticd_routes_present(tgen):
if tgen.routers_have_failure():
pytest.skip(tgen.errors)
- for x in ["r1", "r2", "r3"]:
+ for x in ["r1", "r2", "r3", "r4"]:
tgen.gears[x].net.cmd_nostatus(
"vtysh -c 'debug mgmt client frontend' "
"-c 'debug mgmt client backend' "
r1 = tgen.routers()["r1"]
r2 = tgen.routers()["r2"]
r3 = tgen.routers()["r3"]
+ r4 = tgen.routers()["r4"]
step("Verifying routes are present on r1")
result = check_kernel(r1, "12.0.0.0/24")
assert result is None
result = check_kernel(r3, "12.0.0.0/24")
assert result is None
+
+ step("Verifying routes are present on r4")
+ result = check_kernel(r4, "11.0.0.0/24")
+ assert result is None
+ result = check_kernel(r4, "12.0.0.0/24")
+ assert result is None
--- /dev/null
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: ISC
+#
+# May 2 2023, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2023, LabN Consulting, L.L.C.
+#
+
+"""
+Verify large set of routes present when staticd (backend client) is started after it's
+startup config is present during launch.
+"""
+
+import logging
+import os
+
+import pytest
+from lib.common_config import step
+from lib.topogen import Topogen, TopoRouter
+from munet.base import Timeout
+from util import check_kernel, check_vtysh_up, write_big_route_conf
+
+CWD = os.path.dirname(os.path.realpath(__file__))
+
+# pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
+pytestmark = [pytest.mark.staticd]
+
+track = Timeout(0)
+ROUTE_COUNT = 2500
+ROUTE_RANGE = [None, None]
+
+
+@pytest.fixture(scope="module")
+def tgen(request):
+ "Setup/Teardown the environment and provide tgen argument to tests"
+
+ global start_time
+ topodef = {
+ "s1": ("r1",),
+ }
+
+ tgen = Topogen(topodef, request.module.__name__)
+ tgen.start_topology()
+
+ confpath = f"{tgen.gears['r1'].gearlogdir}/r1-late-big.conf"
+ start, end = write_big_route_conf("10.0.0.0/8", ROUTE_COUNT, confpath)
+ ROUTE_RANGE[0] = start
+ ROUTE_RANGE[1] = end
+
+ # configure mgmtd using current mgmtd config file
+ tgen.gears["r1"].load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
+ tgen.gears["r1"].load_config(TopoRouter.RD_MGMTD, confpath)
+
+ # Explicit disable staticd now..
+ tgen.gears["r1"].net.daemons["staticd"] = 0
+
+ tgen.start_router()
+ yield tgen
+ tgen.stop_topology()
+
+
+def test_staticd_latestart(tgen):
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ r1 = tgen.routers()["r1"]
+
+ check_vtysh_up(r1)
+ logging.info("r1: vtysh connected after %ss", track.elapsed())
+
+ result = check_kernel(r1, ROUTE_RANGE[0], retry_timeout=20, expected=False)
+ assert result is not None, "first route present and should not be"
+ result = check_kernel(r1, ROUTE_RANGE[1], retry_timeout=20, expected=False)
+ assert result is not None, "last route present and should not be"
+
+ step("Starting staticd")
+ r1.startDaemons(["staticd"])
+
+ result = check_kernel(r1, ROUTE_RANGE[0], retry_timeout=60)
+ assert result is None, "first route not present and should be"
+ result = check_kernel(r1, ROUTE_RANGE[1], retry_timeout=20)
+ assert result is None, "last route not present and should be"
--- /dev/null
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: ISC
+#
+# May 2 2023, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2023, LabN Consulting, L.L.C.
+#
+
+"""
+Verify routes present when staticd (backend client) is started after it's startup
+config, contained inside a unified configuration file, is present during launch.
+"""
+import pytest
+from lib.topogen import Topogen
+from util import _test_staticd_late_start
+
+# pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
+pytestmark = [pytest.mark.staticd]
+
+
+@pytest.fixture(scope="module")
+def tgen(request):
+ "Setup/Teardown the environment and provide tgen argument to tests"
+
+ topodef = {
+ "s1": ("r4",),
+ }
+
+ tgen = Topogen(topodef, request.module.__name__)
+ tgen.start_topology()
+
+ # configure mgmtd using current mgmtd config file
+ tgen.gears["r4"].load_frr_config("frr.conf")
+
+ # Explicit disable staticd now..
+ tgen.gears["r4"].net.daemons["staticd"] = 0
+
+ tgen.start_router()
+ yield tgen
+ tgen.stop_topology()
+
+
+def test_staticd_late_start(tgen):
+ return _test_staticd_late_start(tgen, tgen.routers()["r4"])
--- /dev/null
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: ISC
+#
+# May 2 2023, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2023, LabN Consulting, L.L.C.
+#
+"""
+Verify routes present when staticd (backend client) is started after it's startup config
+is present during launch.
+"""
+
+import pytest
+from lib.topogen import Topogen, TopoRouter
+from util import _test_staticd_late_start
+
+# pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
+pytestmark = [pytest.mark.staticd]
+
+
+@pytest.fixture(scope="module")
+def tgen(request):
+ "Setup/Teardown the environment and provide tgen argument to tests"
+
+ topodef = {
+ "s1": ("r1",),
+ }
+
+ tgen = Topogen(topodef, request.module.__name__)
+ tgen.start_topology()
+
+ # configure mgmtd using current mgmtd config file
+ tgen.gears["r1"].load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
+ tgen.gears["r1"].load_config(TopoRouter.RD_MGMTD)
+
+ # Explicit disable staticd now..
+ tgen.gears["r1"].net.daemons["staticd"] = 0
+
+ tgen.start_router()
+ yield tgen
+ tgen.stop_topology()
+
+
+def test_staticd_late_start(tgen):
+ return _test_staticd_late_start(tgen, tgen.routers()["r1"])
+++ /dev/null
-# -*- coding: utf-8 eval: (blacken-mode 1) -*-
-# SPDX-License-Identifier: ISC
-#
-# May 2 2023, Christian Hopps <chopps@labn.net>
-#
-# Copyright (c) 2023, LabN Consulting, L.L.C.
-#
-"""
-Test static route functionality using old or new configuration files.
-
-User compat:
-
- - mgmtd split config will first look to `/etc/frr/zebra.conf`
- then `/etc/frr/staticd.conf` and finally `/etc/frr/mgmtd.conf`
-
- - When new components are converted to mgmtd their split config should be
- added here too.
-
-Topotest compat:
-
- - `mgmtd.conf` is copied to `/etc/frr/` for use by mgmtd when implicit load,
- or explicit load no config specified.
-
- - `staticd.conf` is copied to `/etc/frr/` for use by mgmtd when staticd
- is explicit load implict config, and explicit config.
-
-"""
-
-import ipaddress
-import re
-import time
-
-import pytest
-from lib.common_config import create_static_routes, retry, step, verify_rib
-from lib.topogen import Topogen, TopoRouter
-from lib.topolog import logger
-
-# pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
-pytestmark = [pytest.mark.staticd]
-
-
-@pytest.fixture(scope="module")
-def tgen(request):
- "Setup/Teardown the environment and provide tgen argument to tests"
-
- topodef = {
- "s1": ("r1",),
- }
-
- tgen = Topogen(topodef, request.module.__name__)
- tgen.start_topology()
-
- # configure mgmtd using current mgmtd config file
- tgen.gears["r1"].load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
- tgen.gears["r1"].load_config(TopoRouter.RD_MGMTD)
-
- # Explicit disable staticd now..
- tgen.gears["r1"].net.daemons["staticd"] = 0
-
- tgen.start_router()
- yield tgen
- tgen.stop_topology()
-
-
-@retry(retry_timeout=3, initial_wait=0.1)
-def check_kernel(r1, prefix, expected=True):
- net = ipaddress.ip_network(prefix)
- if net.version == 6:
- kernel = r1.net.cmd_nostatus("ip -6 route show", warn=not expected)
- else:
- kernel = r1.net.cmd_nostatus("ip -4 route show", warn=not expected)
-
- logger.debug("checking kernel routing table:\n%s", kernel)
- route = f"{str(net)}(?: nhid [0-9]+)?.*proto (static|196)"
- m = re.search(route, kernel)
- if expected and not m:
- return f"Failed to find \n'{route}'\n in \n'{kernel}'"
- elif not expected and m:
- return f"Failed found \n'{route}'\n in \n'{kernel}'"
- return None
-
-
-def test_staticd_late_start(tgen):
- if tgen.routers_have_failure():
- pytest.skip(tgen.errors)
-
- # for x in ["r1"]:
- # tgen.gears[x].net.cmd_nostatus(
- # "vtysh -c 'debug mgmt client frontend' "
- # "-c 'debug mgmt client backend' "
- # "-c 'debug mgmt backend frontend datastore transaction'"
- # )
-
- r1 = tgen.routers()["r1"]
-
- step("Verifying startup route is not present w/o staticd running")
- result = check_kernel(r1, "12.0.0.0/24", expected=False)
- assert result is not None
-
- step("Configure another static route verify is not present w/o staticd running")
- r1.net.cmd_nostatus("vtysh -c 'config t' -c 'ip route 12.1.0.0/24 101.0.0.2'")
- result = check_kernel(r1, "12.0.0.0/24", expected=False)
- assert result is not None
- result = check_kernel(r1, "12.1.0.0/24", expected=False)
- assert result is not None
-
- step("Starting staticd")
- r1.startDaemons(["staticd"])
-
- step("Verifying both routes are present")
- result = check_kernel(r1, "12.0.0.0/24")
- assert result is None
- result = check_kernel(r1, "12.1.0.0/24")
- assert result is None
--- /dev/null
+# -*- coding: utf-8 eval: (blacken-mode 1) -*-
+# SPDX-License-Identifier: ISC
+#
+# May 28 2023, Christian Hopps <chopps@labn.net>
+#
+# Copyright (c) 2023, LabN Consulting, L.L.C.
+#
+
+import ipaddress
+import math
+import re
+
+import pytest
+from lib.common_config import retry, step
+from lib.topolog import logger
+from munet.base import proc_error
+
+
+@retry(retry_timeout=30)
+def check_vtysh_up(router):
+ rc, o, e = router.net.cmd_status("vtysh -c 'show version'")
+ return None if not rc else proc_error(rc, o, e)
+
+
+@retry(retry_timeout=3, initial_wait=0.1)
+def check_kernel(r1, prefix, expected=True):
+ net = ipaddress.ip_network(prefix)
+ if net.version == 6:
+ kernel = r1.net.cmd_nostatus("ip -6 route show", warn=not expected)
+ else:
+ kernel = r1.net.cmd_nostatus("ip -4 route show", warn=not expected)
+
+ logger.debug("checking kernel routing table:\n%0.1920s", kernel)
+ route = f"{str(net)}(?: nhid [0-9]+)?.*proto (static|196)"
+ m = re.search(route, kernel)
+ if expected and not m:
+ return f"Failed to find \n'{route}'\n in \n'{kernel:.1920}'"
+ elif not expected and m:
+ return f"Failed found \n'{route}'\n in \n'{kernel:.1920}'"
+ return None
+
+
+def get_ip_networks(super_prefix, count):
+ count_log2 = math.log(count, 2)
+ if count_log2 != int(count_log2):
+ count_log2 = int(count_log2) + 1
+ else:
+ count_log2 = int(count_log2)
+ network = ipaddress.ip_network(super_prefix)
+ return tuple(network.subnets(count_log2))[0:count]
+
+
+def write_big_route_conf(super_prefix, count, confpath):
+ start = None
+ end = None
+
+ with open(confpath, "w+", encoding="ascii") as f:
+ for net in get_ip_networks(super_prefix, count):
+ end = net
+ if not start:
+ start = net
+ f.write(f"ip route {net} lo\n")
+
+ return start, end
+
+
+def _test_staticd_late_start(tgen, router):
+ if tgen.routers_have_failure():
+ pytest.skip(tgen.errors)
+
+ # for x in ["r1"]:
+ # tgen.gears[x].net.cmd_nostatus(
+ # "vtysh -c 'debug mgmt client frontend' "
+ # "-c 'debug mgmt client backend' "
+ # "-c 'debug mgmt backend frontend datastore transaction'"
+ # )
+
+ step("Verifying startup route is not present w/o staticd running")
+ result = check_kernel(router, "12.0.0.0/24", expected=False)
+ assert result is not None
+
+ step("Configure another static route verify is not present w/o staticd running")
+ router.net.cmd_nostatus("vtysh -c 'config t' -c 'ip route 12.1.0.0/24 101.0.0.2'")
+ result = check_kernel(router, "12.0.0.0/24", expected=False)
+ assert result is not None
+ result = check_kernel(router, "12.1.0.0/24", expected=False)
+ assert result is not None
+
+ step("Starting staticd")
+ router.startDaemons(["staticd"])
+
+ step("Verifying both routes are present")
+ result = check_kernel(router, "12.0.0.0/24")
+ assert result is None
+ result = check_kernel(router, "12.1.0.0/24")
+ assert result is None
projects / matthieu / frr.git / commitdiff