if tgen.routers_have_failure():
pytest.skip(tgen.errors)
- failures = 0
router_list = tgen.routers().values()
for router in router_list:
output = router.vtysh_cmd("show ip route json", isjson=True)
import os
import sys
-import json
import time
import pytest
import platform
verify_attributes_for_evpn_routes,
verify_evpn_routes,
)
-from lib.topojson import build_topo_from_json, build_config_from_json
+from lib.topojson import build_config_from_json
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd]
"""
import os
-import re
import sys
import pytest
import json
# Import topogen and topotest helpers
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
pytestmark = [pytest.mark.fpm, pytest.mark.sharpd]
from lib.common_config import step
from lib.micronet import commander
from lib.topogen import Topogen, TopoRouter
-from lib.topolog import logger
from lib.topotest import json_cmp
CWD = os.path.dirname(os.path.realpath(__file__))
tgen.start_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf", f"-M grpc:{GRPCP_ZEBRA}")
router.load_config(TopoRouter.RD_STATIC, "", f"-M grpc:{GRPCP_STATICD}")
# router.load_config(TopoRouter.RD_BFDD, "", f"-M grpc:{GRPCP_BFDD}")
import sys
import pytest
import json
-from time import sleep
-from functools import partial
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
# pylint: disable=C0413
# Import topogen and topotest helpers
-from lib import topotest
from lib.common_config import (
retry,
stop_router,
import sys
import pytest
import json
-import time
import tempfile
from functools import partial
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
rname = "rt1"
router = tgen.gears[rname]
test_func = partial(_rt2_neigh_down, router)
- success, result = topotest.run_and_expect(test_func, None, count=200, wait=0.05)
+ _, result = topotest.run_and_expect(test_func, None, count=200, wait=0.05)
assert result is None, 'rt2 neighbor is still present on "{}"'.format(router)
router_compare_json_output(
rname = "rt1"
router = tgen.gears[rname]
test_func = partial(_bfd_down, router)
- success, result = topotest.run_and_expect(test_func, None, count=30, wait=0.3)
+ _, result = topotest.run_and_expect(test_func, None, count=30, wait=0.3)
assert result is None, 'BFD session is still up on "{}"'.format(router)
router_compare_json_output(
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
import sys
import pytest
import json
-import tempfile
from copy import deepcopy
from functools import partial
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
import sys
import pytest
import json
-import time
from functools import partial
# Save the Current Working Directory to find configuration files.
def build_topo(tgen):
"Build function"
- routers = []
for i in range(0, 10):
rt = tgen.add_router("rt{}".format(i))
rt.run("sysctl -w net.ipv4.fib_multipath_hash_policy=1")
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
logger.info('[+] check {} "{}" {}'.format(name, cmd, expected_file))
tgen = get_topogen()
func = partial(_check, name, cmd, expected_file)
- success, result = topotest.run_and_expect(func, None, count=120, wait=0.5)
+ _, result = topotest.run_and_expect(func, None, count=120, wait=0.5)
assert result is None, "Failed"
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
)
# (re-)build the SR Policy two times to ensure that reinstalling still works
- for i in [1, 2]:
+ for _ in [1, 2]:
cmp_json_output(
"rt1", "show ip route bgp json", "step5/show_ip_route_bgp_inactive_srte.ref"
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
"""
import os
-import re
import sys
import json
import functools
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
# Teardown the topology
logger.info("[+] check {} {} {}".format(name, dest_addr, match))
tgen = get_topogen()
func = functools.partial(_check, name, dest_addr, match)
- success, result = topotest.run_and_expect(func, None, count=10, wait=1)
+ _, result = topotest.run_and_expect(func, None, count=10, wait=1)
assert result is None, "Failed"
import sys
import pytest
import json
-import tempfile
from functools import partial
from time import sleep
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
return topotest.json_cmp(actual, expected)
test_func = functools.partial(compare_isis_installed_routes, router, expected)
- (result, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
+ (result, _) = topotest.run_and_expect(test_func, None, wait=1, count=10)
assertmsg = "Router '{}' routes mismatch".format(rname)
assert result, assertmsg
test_func = functools.partial(
compare_isis_v6_installed_routes, router, expected
)
- (result, diff) = topotest.run_and_expect(test_func, None, wait=1, count=10)
+ (result, _) = topotest.run_and_expect(test_func, None, wait=1, count=10)
assertmsg = "Router '{}' routes mismatch".format(rname)
assert result, assertmsg
pytest.skip(tgen.errors)
logger.info("Checking 'show isis summary json'")
- for rname, router in tgen.routers().items():
+ for rname, _ in tgen.routers().items():
logger.info("Checking router %s", rname)
json_output = tgen.gears[rname].vtysh_cmd("show isis summary json", isjson=True)
assertmsg = "Test isis summary json failed in '{}' data '{}'".format(
pytest.skip(tgen.errors)
logger.info("Checking 'show isis interface json'")
- for rname, router in tgen.routers().items():
+ for rname, _ in tgen.routers().items():
logger.info("Checking router %s", rname)
json_output = tgen.gears[rname].vtysh_cmd(
"show isis interface json", isjson=True
# tgen.mininet_cli()
logger.info("Checking 'show isis neighbor json'")
- for rname, router in tgen.routers().items():
+ for rname, _ in tgen.routers().items():
logger.info("Checking router %s", rname)
json_output = tgen.gears[rname].vtysh_cmd(
"show isis neighbor json", isjson=True
# tgen.mininet_cli()
logger.info("Checking 'show isis database json'")
- for rname, router in tgen.routers().items():
+ for rname, _ in tgen.routers().items():
logger.info("Checking router %s", rname)
json_output = tgen.gears[rname].vtysh_cmd(
"show isis database json", isjson=True
Source:
https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
"""
- for k, v in merge_dct.items():
+ for k, _ in merge_dct.items():
if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
dict_merge(dct[k], merge_dct[k])
else:
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
# move back rx-eth0 to default VRF
Source:
https://gist.github.com/angstwad/bf22d1822c38a92ec0a9
"""
- for k, v in merge_dct.items():
+ for k, _ in merge_dct.items():
if k in dct and isinstance(dct[k], dict) and topotest.is_mapping(merge_dct[k]):
dict_merge(dct[k], merge_dct[k])
else:
tgen.start_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_frr_config("frr.conf")
tgen.start_router()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
else:
del_action = False
- for rs_timer, value in timer.items():
+ for rs_timer, _ in timer.items():
rs_timer_value = timer.setdefault(rs_timer, None)
if rs_timer_value and rs_timer != "delete":
# Copy bgp config file to /etc/frr
for dut in input_dict.keys():
router_list = tgen.routers()
- for router, rnode in router_list.items():
+ for router, _ in router_list.items():
if router != dut:
continue
for bgp_neighbor, peer_data in bgp_neighbors.items():
remote_as = input_dict[bgp_neighbor]["bgp"]["local_as"]
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
neighbor_ip = None
data = topo["routers"][bgp_neighbor]["links"]
return errormsg
for vrf, addr_family_data in show_bgp_json.items():
- for address_family, neighborship_data in addr_family_data.items():
+ for _, neighborship_data in addr_family_data.items():
total_peer = 0
no_of_peer = 0
bgp_neighbors = bgp_addr_type[addr_type]["unicast"]["neighbor"]
for bgp_neighbor, peer_data in bgp_neighbors.items():
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
if bgp_neighbor != peer:
continue
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
if bgp_neighbor != peer:
continue
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
if bgp_neighbor != peer:
continue
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
if bgp_neighbor != peer:
continue
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
if bgp_neighbor != peer:
continue
- for dest_link, peer_dict in peer_data["dest_link"].items():
+ for dest_link, _ in peer_data["dest_link"].items():
data = topo["routers"][bgp_neighbor]["links"]
if dest_link in data:
logger.info("[+] check {} {} {}".format(name, dest_addr, match))
tgen = get_topogen()
func = functools.partial(_check, name, dest_addr, source_addr, match)
- success, result = topotest.run_and_expect(func, None, count=count, wait=wait)
+ _, result = topotest.run_and_expect(func, None, count=count, wait=wait)
assert result is None, "Failed"
import functools
import ipaddress
-import json
import os
import platform
import socket
try:
router_list = tgen.routers()
- for router, rnode in router_list.items():
+ for _, rnode in router_list.items():
result = rnode.check_router_running()
if result != "":
daemons = []
"""
routers = tgen.routers()
- for rname, router in routers.items():
+ for rname, _ in routers.items():
destname = "{}/{}/{}".format(tgen.logdir, rname, FRRCFG_FILE)
wmode = "w"
for cfbase in config_name_list:
"""
router_list = tgen.routers()
- for rname, rnode in router_list.items():
+ for rname, _ in router_list.items():
if rname != router:
continue
cmd = "show interface vrf {}".format(vrf)
else:
cmd = "show interface"
- for chk_ll in range(0, 60):
+ for _ in range(0, 60):
sleep(1 / 4)
ifaces = router_list[router].run('vtysh -c "{}"'.format(cmd))
# Fix newlines (make them all the same)
"{} dev {} master {}".format(ip_cmd, brctl_name, vrf)
)
- for intf_name, data in topo["routers"][dut]["links"].items():
+ for _, data in topo["routers"][dut]["links"].items():
if "vrf" not in data:
continue
sender_interface = intf
rnode = tgen.routers()[senderRouter]
- for destLink, data in topo["routers"][senderRouter]["links"].items():
+ for _, data in topo["routers"][senderRouter]["links"].items():
if "type" in data and data["type"] == "loopback":
continue
# noqa: E501
#
import argparse
-import json
import logging
import os
import socket
self.sess_id = reply.session_reply.session_id
else:
self.sess_id = 0
- mdata, req_id = self.get_native_msg_header(MSG_CODE_SESSION_REQ)
+ mdata, _ = self.get_native_msg_header(MSG_CODE_SESSION_REQ)
mdata += struct.pack(MSG_SESSION_REQ_FMT)
mdata += "test-client".encode("utf-8") + b"\x00"
def get_data(self, query, data=True, config=False):
# Create the message
- mdata, req_id = self.get_native_msg_header(MSG_CODE_GET_DATA)
+ mdata, _ = self.get_native_msg_header(MSG_CODE_GET_DATA)
flags = GET_DATA_FLAG_STATE if data else 0
flags |= GET_DATA_FLAG_CONFIG if config else 0
mdata += struct.pack(MSG_GET_DATA_FMT, MSG_FORMAT_JSON, flags)
self.send_native_msg(mdata)
logging.debug("Sent GET-TREE")
- mhdr, mfixed, mdata = self.recv_native_msg()
+ _, mfixed, mdata = self.recv_native_msg()
assert mdata[-1] == 0
result = mdata[:-1].decode("utf-8")
def add_notify_select(self, replace, notif_xpaths):
# Create the message
- mdata, req_id = self.get_native_msg_header(MSG_CODE_NOTIFY_SELECT)
+ mdata, _ = self.get_native_msg_header(MSG_CODE_NOTIFY_SELECT)
mdata += struct.pack(MSG_NOTIFY_SELECT_FMT, replace)
for xpath in notif_xpaths:
if xpaths:
self.add_notify_select(True, xpaths)
- for remaining in Timeout(60):
+ for _ in Timeout(60):
logging.debug("Waiting for Notify Message")
mhdr, mfixed, mdata = self.recv_native_msg()
if mhdr[HDR_FIELD_CODE] == MSG_CODE_NOTIFY:
)
return errormsg
if ospf_external_lsa:
- for ospf_ext_lsa, ext_lsa_data in ospf_external_lsa.items():
+ for ospf_ext_lsa, _ in ospf_external_lsa.items():
if ospf_ext_lsa in show_ospf_json["AS External Link States"]:
logger.info(
"[DUT: %s] OSPF LSDB:External LSA %s", router, ospf_ext_lsa
raise ValueError(errormsg)
return errormsg
- for ospf_gr, gr_data in input_dict.items():
+ for ospf_gr, _ in input_dict.items():
try:
if input_dict[ospf_gr] == show_ospf_json[ospf_gr]:
logger.info(
# At least one interface must be enabled for PIM on the router
pim_if_enabled = False
pim6_if_enabled = False
- for destLink, data in topo[dut]["links"].items():
+ for _, data in topo[dut]["links"].items():
if "pim" in data:
pim_if_enabled = True
if "pim6" in data:
# ip address of RP
rp_addr = rp_dict["rp_addr"]
- for link, data in topo["routers"][router]["links"].items():
+ for _, data in topo["routers"][router]["links"].items():
if data["ipv4"].split("/")[0] == rp_addr:
rp_details[router] = rp_addr
)
return True
else:
- for destLink, data in topo["routers"][dut]["links"].items():
+ for _, data in topo["routers"][dut]["links"].items():
if "type" in data and data["type"] == "loopback":
continue
# Waiting for maximum 60 sec
fail_intf = []
- for retry in range(1, 13):
+ for _ in range(1, 13):
sleep(5)
logger.info("[DUT: %s]: Waiting for 5 sec for PIM neighbors" " to come up", dut)
run_json_after = run_frr_cmd(rnode, "show ip pim neighbor json", isjson=True)
total_groups_before_clear = igmp_json["totalGroups"]
- for key, value in igmp_json.items():
+ for _, value in igmp_json.items():
if type(value) is not dict:
continue
result = run_frr_cmd(rnode, "clear ip igmp interfaces")
# Waiting for maximum 60 sec
- for retry in range(1, 13):
+ for _ in range(1, 13):
logger.info(
"[DUT: %s]: Waiting for 5 sec for igmp interfaces" " to come up", dut
)
# RFC 3376: 8.2. Query Interval - Default: 125 seconds
# So waiting for maximum 130 sec to get the igmp report
- for retry in range(1, 26):
+ for _ in range(1, 26):
logger.info("[DUT: %s]: Waiting for 2 sec for mroutes" " to come up", dut)
sleep(5)
keys_json1 = mroute_json_1.keys()
try:
config_data = []
- for group, rp_list in rp_mapping.items():
+ for _, rp_list in rp_mapping.items():
for _rp in rp_list:
config_data.append("interface {}".format(interface))
config_data.append("ip address {}".format(_rp))
script_path = os.path.join(CWD, "send_bsr_packet.py")
node = tgen.net[senderRouter]
- for destLink, data in topo["routers"][senderRouter]["links"].items():
+ for _, data in topo["routers"][senderRouter]["links"].items():
if "type" in data and data["type"] == "loopback":
continue
# RP with lowest priority
if len(priority_dict) != 1:
- rp_p, lowest_priority = sorted(rp_priority.items(), key=lambda x: x[1])[0]
+ rp_p, _ = sorted(rp_priority.items(), key=lambda x: x[1])[0]
rp_details[group] = rp_p
# RP with highest hash value
if len(priority_dict) == 1:
- rp_h, highest_hash = sorted(rp_hash.items(), key=lambda x: x[1])[-1]
+ rp_h, _ = sorted(rp_hash.items(), key=lambda x: x[1])[-1]
rp_details[group] = rp_h
# RP with highest IP address
interface_json = show_pim_join_json[interface]
grp_addr = grp_addr.split("/")[0]
- for source, data in interface_json[grp_addr].items():
+ for _, data in interface_json[grp_addr].items():
# Verify pim join
if pim_join:
if data["group"] == grp_addr and data["channelJoinName"] == "JOIN":
tgen = Topogen(topodef, request.module.__name__)
tgen.start_topology()
- for rname, router in tgen.routers().items():
+ for _, router in tgen.routers().items():
router.load_frr_config("frr.conf")
tgen.start_router()
Test YANG Notifications
"""
import json
-import logging
import os
import pytest
tgen.start_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_frr_config("frr.conf")
tgen.start_router()
# logger.debug("checking kernel routing table%s:\n%s", vrfstr, kernel)
- for i, net in enumerate(get_ip_networks(super_prefix, count)):
+ for _, net in enumerate(get_ip_networks(super_prefix, count)):
if not add:
assert str(net) not in kernel
continue
if vrf:
f.write("vrf {}\n".format(vrf))
- for i, net in enumerate(get_ip_networks(super_prefix, count)):
+ for _, net in enumerate(get_ip_networks(super_prefix, count)):
if add:
f.write("ip route {} {}\n".format(net, via))
else:
import ipaddress
import math
-import time
import pytest
from lib.topogen import Topogen
from oper import check_kernel_32, do_oper_test
-try:
- from deepdiff import DeepDiff as dd_json_cmp
-except ImportError:
- dd_json_cmp = None
-
pytestmark = [pytest.mark.staticd, pytest.mark.mgmtd]
tgen.start_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_frr_config("frr.conf")
tgen.start_router()
start_router_daemons,
)
from lib.topolog import logger
-from lib.bgp import verify_bgp_convergence, create_router_bgp, verify_bgp_rib
from lib.topojson import build_config_from_json
pytestmark = [pytest.mark.bgpd, pytest.mark.staticd, pytest.mark.mgmtd]
app_helper.init(tgen)
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
app_helper.cleanup()
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
import os
import sys
-import json
import time
-import datetime
-from time import sleep
import pytest
# Save the Current Working Directory to find configuration files.
write_test_footer,
step,
reset_config_on_routers,
- shutdown_bringup_interface,
apply_raw_config,
add_interfaces_to_vlan,
- kill_router_daemons,
- start_router_daemons,
- create_static_routes,
check_router_status,
- topo_daemons,
required_linux_kernel_version,
)
from lib.pim import (
verify_mroutes,
clear_mroute,
clear_pim_interface_traffic,
- verify_pim_config,
- verify_upstream_iif,
- verify_multicast_traffic,
verify_multicast_flag_state,
verify_igmp_groups,
McastTesterHelper,
reset_config_on_routers,
shutdown_bringup_interface,
required_linux_kernel_version,
- topo_daemons,
)
from lib.pim import (
create_pim_config,
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
start_router_daemons,
stop_router,
required_linux_kernel_version,
- topo_daemons,
)
from lib.pim import (
create_pim_config,
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.common_config import (
start_topology,
write_test_header,
apply_raw_config,
check_router_status,
required_linux_kernel_version,
- topo_daemons,
)
from lib.pim import (
create_pim_config,
"""
for router, stats_data in stats_before.items():
- for stats, value in stats_data.items():
+ for stats, _ in stats_data.items():
if stats_before[router][stats] >= stats_after[router][stats]:
errormsg = (
"[DUT: %s]: state %s value has not"
apply_raw_config,
create_static_routes,
required_linux_kernel_version,
- topo_daemons,
)
from lib.pim import (
create_pim_config,
"""
for router, state_data in stats.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
stats[router][state] = 0
logger.info(
"[DUT: %s]: stats %s value has reset" " reset, Current value: %s",
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] >= state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
import os
import sys
import time
-from time import sleep
import datetime
import pytest
reset_config_on_routers,
step,
shutdown_bringup_interface,
- kill_router_daemons,
- start_router_daemons,
create_static_routes,
- topo_daemons,
)
from lib.pim import (
create_pim_config,
verify_pim_rp_info,
verify_pim_state,
clear_pim_interface_traffic,
- clear_igmp_interfaces,
- clear_pim_interfaces,
clear_mroute,
- clear_mroute_verify,
McastTesterHelper,
)
reset_config_on_routers,
step,
shutdown_bringup_interface,
- kill_router_daemons,
- start_router_daemons,
- create_static_routes,
)
from lib.pim import (
create_pim_config,
verify_join_state_and_timer,
verify_mroutes,
verify_pim_neighbors,
- get_pim_interface_traffic,
verify_pim_rp_info,
- verify_pim_state,
clear_pim_interface_traffic,
clear_igmp_interfaces,
clear_pim_interfaces,
verify_join_state_and_timer,
verify_mroutes,
verify_pim_neighbors,
- get_pim_interface_traffic,
verify_pim_rp_info,
- verify_pim_state,
clear_pim_interface_traffic,
- clear_igmp_interfaces,
- clear_pim_interfaces,
clear_mroute,
- clear_mroute_verify,
McastTesterHelper,
)
import os
import sys
-import json
import time
import pytest
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] > state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
step("Shut and No shut source interface multiple time")
- for i in range(0, 2):
+ for _ in range(0, 2):
step("Shut and no shut the source interface from DUT")
intf_r1_i2 = topo["routers"]["r1"]["links"]["i2"]["interface"]
shutdown_bringup_interface(tgen, "r1", intf_r1_i2, False)
"""
for router, state_data in state_before.items():
- for state, value in state_data.items():
+ for state, _ in state_data.items():
if state_before[router][state] > state_after[router][state]:
errormsg = (
"[DUT: %s]: state %s value has not"
tgen.start_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_frr_config("frr.conf")
tgen.start_router()
import os
import sys
import json
-from time import sleep
from functools import partial
import pytest
router_list = tgen.routers()
# Check NHRP cache on servers and clients
- for rname, router in router_list.items():
+ for _, router in router_list.items():
json_file = "{}/{}/nhrp_cache.json".format(CWD, router.name)
if not os.path.isfile(json_file):
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
# pylint: disable=C0413
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
def setup_module(mod):
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
# tgen.mininet_cli()
# Verify OSPFv3 Routing Table
- for router, rnode in tgen.routers().items():
+ for router, _ in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
# Load expected results from the command
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
# tgen.mininet_cli()
# Verify OSPFv3 Routing Table
- for router, rnode in tgen.routers().items():
+ for router, _ in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
# Load expected results from the command
r1.vtysh_cmd("clear ipv6 ospf interface r1-sw5")
# Verify OSPFv3 Routing Table
- for router, rnode in tgen.routers().items():
+ for router, _ in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
# Load expected results from the command
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
# For debugging, uncomment the next line
# tgen.mininet_cli()
# Verify OSPFv3 Routing Table
- for router, rnode in tgen.routers().items():
+ for router, _ in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
# Load expected results from the command
r1.vtysh_cmd("clear ipv6 ospf interface r1-sw5")
# Verify OSPFv3 Routing Table
- for router, rnode in tgen.routers().items():
+ for router, _ in tgen.routers().items():
logger.info('Waiting for router "%s" convergence', router)
# Load expected results from the command
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
step("Configure and re configure all the commands 10 times in a loop.")
- for itrate in range(0, 10):
+ for _ in range(0, 10):
ospf_summ_r1 = {
"r0": {
"ospf": {
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
step("Verify that route in R2 in stalled with 8 next hops.")
nh = []
- for item in range(1, 7):
+ for _ in range(1, 7):
nh.append(topo["routers"]["r0"]["links"]["r1-link1"]["ipv4"].split("/")[0])
nh2 = topo["routers"]["r0"]["links"]["r1"]["ipv4"].split("/")[0]
create_static_routes,
step,
topo_daemons,
- shutdown_bringup_interface,
check_router_status,
start_topology,
write_test_header,
write_test_header,
write_test_footer,
reset_config_on_routers,
- stop_router,
- start_router,
step,
create_static_routes,
kill_router_daemons,
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
dut = "r1"
- for num in range(0, nretry):
+ for _ in range(0, nretry):
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh, expected=False)
if result is not True:
break
assert result is True, "Testcase {} : Failed \n Error: {}".format(tc_name, result)
dut = "r1"
- for num in range(0, nretry):
+ for _ in range(0, nretry):
result = verify_ospf_rib(tgen, dut, input_dict, next_hop=nh, expected=False)
if result is not True:
break
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
write_test_footer,
reset_config_on_routers,
step,
- create_interfaces_cfg,
scapy_send_raw_packet,
)
from lib.ospf import (
verify_ospf_neighbor,
- clear_ospf,
verify_ospf_gr_helper,
create_router_ospf,
)
write_test_footer,
reset_config_on_routers,
step,
- create_interfaces_cfg,
scapy_send_raw_packet,
)
from lib.ospf import (
verify_ospf_neighbor,
- clear_ospf,
verify_ospf_gr_helper,
create_router_ospf,
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
"""
import os
-import re
import sys
import pytest
import json
import os
import sys
import json
-from time import sleep
from functools import partial
import pytest
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
tgen.set_error("unsupported version")
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
tgen.set_error("unsupported version")
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
tgen.set_error("unsupported version")
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
# Move interfaces out of vrf namespace and delete the namespace
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for rname, _ in router_list.items():
tgen.net[rname].reset_intf_netns(rname + "-eth0")
tgen.net[rname].reset_intf_netns(rname + "-eth1")
tgen.net[rname].delete_netns(rname + "-ospf-cust1")
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
import os
import sys
-import json
-from time import sleep
from functools import partial
import pytest
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
from lib.common_config import (
- run_frr_cmd,
- shutdown_bringup_interface,
- start_router_daemons,
step,
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
import os
import sys
-import json
from time import sleep
from functools import partial
import pytest
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
from lib.common_config import (
- run_frr_cmd,
- shutdown_bringup_interface,
- start_router_daemons,
step,
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
import os
import sys
-import json
-from time import sleep
from functools import partial
import pytest
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
from lib.common_config import (
- run_frr_cmd,
- shutdown_bringup_interface,
- start_router_daemons,
step,
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
import pytest
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.topolog import logger
from lib.common_config import verify_rib
from lib.ospf import verify_ospf_rib
-from _ast import Try
pytestmark = pytest.mark.ospfd
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Tear-down the test environment"
tgen = get_topogen()
tgen.stop_topology()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
import sys
import json
from functools import partial
-import re
import pytest
# Save the Current Working Directory to find configuration files.
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_frr_config("frr.conf")
tgen.start_router()
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
import subprocess
import sys
import time
-from datetime import datetime, timedelta
from functools import partial
import pytest
# pylint: disable=C0413
# Import topogen and topotest helpers
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
+from lib.topogen import Topogen, TopoRouter
pytestmark = [pytest.mark.ospfd]
from lib.common_config import (
start_topology,
write_test_header,
- kill_router_daemons,
write_test_footer,
reset_config_on_routers,
stop_router,
verify_rib,
create_static_routes,
step,
- start_router_daemons,
create_route_maps,
shutdown_bringup_interface,
create_prefix_lists,
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
step("Configure and re configure all the commands 10 times in a loop.")
- for itrate in range(0, 10):
+ for _ in range(0, 10):
ospf_summ_r1 = {
"r0": {
"ospf6": {
import time
import pytest
from time import sleep
-from copy import deepcopy
import json
-from lib.topotest import frr_unicode
pytestmark = pytest.mark.ospf6d
shutdown_bringup_interface,
)
from lib.topolog import logger
-from lib.topojson import build_topo_from_json, build_config_from_json
-from lib.ospf import verify_ospf6_neighbor, config_ospf6_interface, clear_ospf
-from ipaddress import IPv4Address
-
-# Global variables
+from lib.topojson import build_config_from_json
+from lib.ospf import verify_ospf6_neighbor, config_ospf6_interface
topo = None
# Reading the data from JSON File for topology creation
jsonFile = "{}/ospfv3_authentication.json".format(CWD)
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
* `mod`: module name
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
step("Verify that route in R2 in stalled with 8 next hops.")
nh = []
- for item in range(1, 7):
+ for _ in range(1, 7):
nh.append(llip)
llip = get_llip("r0", "r1")
import sys
import time
import pytest
-import json
-from copy import deepcopy
-from ipaddress import IPv4Address
-from lib.topotest import frr_unicode
-import ipaddress
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
verify_rib,
create_static_routes,
step,
- create_route_maps,
- shutdown_bringup_interface,
- create_interfaces_cfg,
get_frr_ipv6_linklocal,
)
from lib.topolog import logger
from lib.ospf import (
verify_ospf6_neighbor,
- config_ospf_interface,
clear_ospf,
verify_ospf6_rib,
create_router_ospf,
- verify_ospf6_interface,
- verify_ospf6_database,
- config_ospf6_interface,
)
-from ipaddress import IPv6Address
pytestmark = [pytest.mark.ospfd, pytest.mark.staticd]
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
import sys
import time
import pytest
-from copy import deepcopy
import ipaddress
from lib.ospf import (
verify_ospf6_neighbor,
- config_ospf6_interface,
- clear_ospf,
verify_ospf6_rib,
- verify_ospf6_interface,
- verify_ospf6_database,
create_router_ospf,
)
# Import topogen and topotest helpers
from lib.topogen import Topogen, get_topogen
-from lib.bgp import (
- verify_bgp_convergence,
- create_router_bgp,
- clear_bgp_and_verify,
- verify_bgp_rib,
-)
from lib.topolog import logger
from lib.common_config import (
start_topology,
verify_rib,
create_static_routes,
step,
- create_route_maps,
- shutdown_bringup_interface,
create_interfaces_cfg,
check_router_status,
)
-from ipaddress import IPv4Address
from lib.topolog import logger
from lib.topojson import build_config_from_json
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
verify_ospf6_neighbor,
clear_ospf,
verify_ospf6_rib,
- verify_ospf_database,
create_router_ospf,
config_ospf6_interface,
verify_ospf6_interface,
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
step,
create_interfaces_cfg,
create_debug_log_config,
- apply_raw_config,
)
from lib.topolog import logger
from lib.topojson import build_config_from_json
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
tgen.start_router()
-def teardown_module(module):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# tgen.mininet_cli()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
from lib.topotest import iproute2_is_vrf_capable
-from lib.common_config import required_linux_kernel_version, retry
+from lib.common_config import required_linux_kernel_version
from lib.pim import McastTesterHelper
)
-def teardown_module(module):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# pylint: disable=C0413
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.common_config import step
pytestmark = [pytest.mark.ripd]
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
from functools import partial
from lib import topotest
from lib.topogen import Topogen, TopoRouter
-from lib.topolog import logger
pytestmark = [
pytest.mark.bfdd,
# pylint: disable=C0413
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.common_config import step
pytestmark = [pytest.mark.ripd]
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# pylint: disable=C0413
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
+from lib.topogen import Topogen, get_topogen
from lib.common_config import step
pytestmark = [pytest.mark.ripngd]
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# pylint: disable=C0413
from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.common_config import step
+from lib.topogen import Topogen, get_topogen
pytestmark = [pytest.mark.ripngd]
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
# Build up a list of dicts with params for each step of the test;
# use defaults where the step doesn't supply a value
- scale_setups = []
s = scale_steps[iter]
d = dict(zip(scale_keys, s))
"""
import os
-import re
import sys
import pytest
-import json
-from functools import partial
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
# pylint: disable=C0413
# Import topogen and topotest helpers
-from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
from scale_test_common import (
scale_build_common,
"""
import os
-import re
import sys
import pytest
-import json
-from functools import partial
# Save the Current Working Directory to find configuration files.
CWD = os.path.dirname(os.path.realpath(__file__))
# pylint: disable=C0413
# Import topogen and topotest helpers
-from lib import topotest
-from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
from scale_test_common import (
scale_build_common,
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
import sys
import json
import pytest
-import functools
CWD = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CWD, "../"))
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
def check_srv6_locator(router, expected_file):
func = functools.partial(_check_srv6_locator, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
+ _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed"
def check_sharpd_chunk(router, expected_file):
func = functools.partial(_check_sharpd_chunk, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
+ _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed"
# FOR DEVELOPER:
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
def check_srv6_locator(router, expected_file):
func = functools.partial(_check_srv6_locator, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
+ _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed"
def check_sharpd_chunk(router, expected_file):
func = functools.partial(_check_sharpd_chunk, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
+ _, result = topotest.run_and_expect(func, None, count=10, wait=0.5)
assert result is None, "Failed"
# FOR DEVELOPER:
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
def check_srv6_locator(router, expected_file):
func = functools.partial(_check_srv6_locator, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=5, wait=3)
+ _, result = topotest.run_and_expect(func, None, count=5, wait=3)
assert result is None, "Failed"
def check_sharpd_chunk(router, expected_file):
func = functools.partial(_check_sharpd_chunk, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=5, wait=3)
+ _, result = topotest.run_and_expect(func, None, count=5, wait=3)
assert result is None, "Failed"
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
def check_srv6_static_route(router, expected_file):
func = functools.partial(_check_srv6_static_route, router, expected_file)
- success, result = topotest.run_and_expect(func, None, count=15, wait=1)
+ _, result = topotest.run_and_expect(func, None, count=15, wait=1)
assert result is None, "Failed"
# FOR DEVELOPER:
"""
import os
-import re
import sys
import pytest
-import json
-from functools import partial
import functools
# Save the Current Working Directory to find configuration files.
test_func = functools.partial(
_check_mpls_state_interface, router, interface, up=configured
)
- success, result = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
+ success, _ = topotest.run_and_expect(test_func, None, count=10, wait=0.5)
return success
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment
logger.info("Running setup_module() done")
-def teardown_module(mod):
+def teardown_module():
"""
Teardown the pytest environment.
import ipaddress
import math
import os
-import sys
import re
import pytest
-from lib.topogen import TopoRouter, Topogen, get_topogen
+from lib.topogen import TopoRouter, Topogen
from lib.topolog import logger
from lib.common_config import retry, step
kernel = r1.run(f"ip -4 route show{vrfstr}")
logger.debug("checking kernel routing table%s:\n%s", vrfstr, kernel)
- for i, net in enumerate(get_ip_networks(super_prefix, count)):
+ for _, net in enumerate(get_ip_networks(super_prefix, count)):
if not add:
assert str(net) not in kernel
continue
else:
super_prefix = "2001::/48" if do_ipv6 else "10.0.0.0/8"
- matchtype = ""
matchvia = ""
if via == "blackhole":
pass
if vrf:
f.write("vrf {}\n".format(vrf))
- for i, net in enumerate(get_ip_networks(super_prefix, count)):
+ for _, net in enumerate(get_ip_networks(super_prefix, count)):
if add:
f.write("ip route {} {}\n".format(net, via))
else:
"""
import os
-import re
import sys
import pytest
import json
# Import topogen and topotest helpers
from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
-from lib.topolog import logger
# Required to instantiate the topology builder class.
test_func = partial(
topotest.router_output_cmp, router, "show ip route 192.168.44.0/24", expected
)
- result, diff = topotest.run_and_expect(test_func, "", count=20, wait=1)
+ result, _ = topotest.run_and_expect(test_func, "", count=20, wait=1)
assert result, "Connected Route should not have been added"
"""
# pylint: disable=C0413
import ipaddress
-import json
import sys
-from functools import partial
import pytest
from lib import topotest
# Initialize all routers.
router_list = tgen.routers()
- for rname, router in router_list.items():
+ for _, router in router_list.items():
router.load_config(TopoRouter.RD_ZEBRA, "zebra.conf")
router.load_config(TopoRouter.RD_SHARP)
import pytest
from lib.common_config import (
- start_topology,
verify_rib,
verify_ip_nht,
step,
)
# pylint: disable=C0413
-from lib import topotest
from lib.topogen import Topogen, TopoRouter, get_topogen
from lib.topolog import logger
router_list = tgen.routers()
- for i, (rname, router) in enumerate(router_list.items(), 1):
+ for _, (rname, router) in enumerate(router_list.items(), 1):
router.load_config(
TopoRouter.RD_ZEBRA, os.path.join(CWD, "{}/zebra.conf".format(rname))
)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
tgen = get_topogen()
tgen.stop_topology()
router = tgen.gears["r1"]
test_func = functools.partial(_bgp_converge, router)
- success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+ _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Cannot see BGP community aliases "{}"'.format(router)
router = tgen.gears["r3"]
test_func = functools.partial(_ospf_converge, router)
- success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+ _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Cannot see OSPFv2 opaque attributes "{}"'.format(router)
router = tgen.gears["r3"]
test_func = functools.partial(_ospf6_converge, router)
- success, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
+ _, result = topotest.run_and_expect(test_func, None, count=60, wait=0.5)
assert result is None, 'Cannot see OSPFv3 opaque attributes "{}"'.format(router)
tgen.start_router()
-def teardown_module(mod):
+def teardown_module():
"Teardown the pytest environment"
tgen = get_topogen()
tgen.stop_topology()
return topotest.json_cmp(output, expected)
test_func = partial(check_initial_routes_installed, r1)
- success, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
+ _, result = topotest.run_and_expect(test_func, None, count=40, wait=1)
static_rmapfile = "%s/r1/static_rmap.ref" % (thisDir)
expected = open(static_rmapfile).read().rstrip()
expected = open_json_file(os.path.join(CWD, "{}/routes_setup.json".format("r1")))
test_func = partial(check_connected, r1, "2001::/64", expected)
- success, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
+ _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
assert result is None, "Failed to fully setup connected routes needed"
manifests = open_json_file(os.path.join(CWD, "{}/routes.json".format("r1")))
)
logger.info("CHECK {} {} {}".format(dest, nh, sid))
test_func = partial(check, r1, dest, manifest["out"])
- success, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
+ _, result = topotest.run_and_expect(test_func, None, count=20, wait=1)
assert result is None, "Failed"
dest,
manifest["out"],
)
- success, result = topotest.run_and_expect(test_func, None, count=25, wait=1)
+ _, result = topotest.run_and_expect(test_func, None, count=25, wait=1)
assert result is None, "Failed"