Commit 7009d493 authored by Štěpán Balážik's avatar Štěpán Balážik

flake8, pylint

parent ff5e0dbf
Pipeline #43448 failed with stage
in 2 minutes and 34 seconds
......@@ -136,6 +136,7 @@ def pytest_generate_tests(metafunc):
max_retries = 3
metafunc.parametrize("max_retries", [max_retries], ids=str)
def check_log_level_xdist(level):
if level < logging.ERROR:
pytest.exit("Advanced logging not available while running with xdist "
......
......@@ -332,28 +332,28 @@ def setup_daemons(tmpdir, prog_cfgs, template_ctx, ta_files):
def check_for_icmp():
""" Checks Deckards's PCAP for ICMP packets """
# Deckard's responses to resolvers might be delayed due to load which
# leads the resolver to close the port and to the test failing in the
# end. We partially detect these by checking the PCAP for ICMP packets.
path = os.environ["SOCKET_WRAPPER_PCAP_FILE"]
udp_seen = False
with open(path, "rb") as f:
pcap = dpkt.pcap.Reader(f)
for _, packet in pcap:
try:
ip = dpkt.ip.IP(packet)
except dpkt.dpkt.UnpackError:
ip = dpkt.ip6.IP6(packet)
if isinstance(ip.data, dpkt.udp.UDP):
udp_seen = True
if udp_seen:
if isinstance(ip.data, dpkt.icmp.ICMP) or isinstance(ip.data, dpkt.icmp6.ICMP6):
raise DeckardUnderLoadError("Deckard is under load. "
"Other errors might be false negatives. "
"Consider retrying the job later.")
return False
""" Checks Deckards's PCAP for ICMP packets """
# Deckard's responses to resolvers might be delayed due to load which
# leads the resolver to close the port and to the test failing in the
# end. We partially detect these by checking the PCAP for ICMP packets.
path = os.environ["SOCKET_WRAPPER_PCAP_FILE"]
udp_seen = False
with open(path, "rb") as f:
pcap = dpkt.pcap.Reader(f)
for _, packet in pcap:
try:
ip = dpkt.ip.IP(packet)
except dpkt.dpkt.UnpackError:
ip = dpkt.ip6.IP6(packet)
if isinstance(ip.data, dpkt.udp.UDP):
udp_seen = True
if udp_seen:
if isinstance(ip.data, dpkt.icmp.ICMP, dpkt.icmp6.ICMP6):
raise DeckardUnderLoadError("Deckard is under load. "
"Other errors might be false negatives. "
"Consider retrying the job later.")
return False
def run_testcase(daemons, case, root_addr, addr_family, prog_under_test_ip):
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment