Skip to content

Commit

Permalink
Use the retry decorator in WiFi tests (Bugfix) (#1488)
Browse files Browse the repository at this point in the history 
* Add the function name when running a retry

This helps investigating issues when more than one functions are using
the retry decorator.

* Rename retry mock function

* Refactor open_connection() and secured_connection()

Both function are doing exactly the same thing, the only difference
being the command used.

Create a connection() function, and call it from open_connection() and
secured_connection().

Compared to the previous implementation, the connection() function is
more linear and does not rely on boolean return codes to decide what to
do. Instead, it's trying to do things and expect exceptions to be raised
in case of problem (this is why it's using subprocess.run(...,
check=True) for instance)

* Let exceptions bubble up instead of catching them

* Use the retry decorator

Parts of the wifi_nmcli_test.py script was implementing a basic retry
mechanism. Remove this to use the retry decorator instead.

In addition, raise exceptions rather than trying to catching them, or
using boolean return codes.

* Adjust unit tests

* Add mock_retry to the unit tests for the main function

This is to avoid waiting for the actual retry decorator implementation.

* Fix unit test to work with Python 3.5

* Fix long line that failed the flake8 tests

* Fix perform_ping_test to raise an exception

* Improve unit test coverage

* Update providers/base/bin/wifi_nmcli_test.py

Co-authored-by: Massimiliano <[email protected]>

* Update providers/base/bin/wifi_nmcli_test.py

Co-authored-by: Massimiliano <[email protected]>

* Cleanup tests and unused variables to make automated checks happy

* Replace run() with check_call()

* Add contributors in script header

* Fix more unit tests

---------

Co-authored-by: Massimiliano <[email protected]>
  • Loading branch information
@pieqq @Hook25
pieqq and Hook25 authored Sep 24, 2024
1 parent e721427 commit 0952eb2
Show file tree
Hide file tree
Showing 3 changed files with 210 additions and 239 deletions.
6 changes: 4 additions & 2 deletions checkbox-support/checkbox_support/helpers/retry.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@ def run_with_retry(f, max_attempts, delay, *args, **kwargs):
"delay should be at least 1 ({} was used)".format(delay)
)
for attempt in range(1, max_attempts + 1):
attempt_string = "Attempt {}/{}".format(attempt, max_attempts)
attempt_string = "Attempt {}/{} (function '{}')".format(
attempt, max_attempts, f.__name__
)
print()
print("=" * len(attempt_string))
print(attempt_string)
Expand Down Expand Up @@ -97,7 +99,7 @@ def fake_run_with_retry(f, max_attempts, delay, *args, **kwargs):
return f(*args, **kwargs)


mock_timeout = functools.partial(
mock_retry = functools.partial(
patch,
"checkbox_support.helpers.retry.run_with_retry",
new=fake_run_with_retry,
Expand Down
198 changes: 76 additions & 122 deletions providers/base/bin/wifi_nmcli_test.py
View file
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
#!/usr/bin/env python3
# Copyright 2017-2019 Canonical Ltd.
# Copyright 2017-2024 Canonical Ltd.
# All rights reserved.
#
# Written by:
# Jonathan Cave <[email protected]>
# Taihsiang Ho <[email protected]>
# Isaac Yang <[email protected]>
# Pierre Equoy <[email protected]>
#
# wireless connection tests using nmcli

Expand All @@ -15,11 +17,11 @@
import os
import subprocess as sp
import sys
import time
import shlex

from packaging import version as version_parser

from checkbox_support.helpers.retry import retry
from gateway_ping_test import ping


Expand Down Expand Up @@ -89,13 +91,10 @@ def turn_down_nm_connections():
for name, value in connections.items():
uuid = value["uuid"]
print("Turn down connection", name)
try:
cmd = "nmcli c down {}".format(uuid)
print_cmd(cmd)
sp.call(shlex.split(cmd))
print("{} {} is down now".format(name, uuid))
except sp.CalledProcessError as e:
print("Can't down {}: {}".format(uuid, str(e)))
cmd = "nmcli c down {}".format(uuid)
print_cmd(cmd)
sp.check_call(shlex.split(cmd))
print("{} {} is down now".format(name, uuid))
print()


Expand All @@ -105,26 +104,18 @@ def delete_test_ap_ssid_connection():
if "TEST_CON" not in connections:
print("No TEST_CON connection found, nothing to delete")
return
try:
cmd = "nmcli c delete TEST_CON"
print_cmd(cmd)
sp.call(shlex.split(cmd))
print("TEST_CON is deleted")
except Exception as e:
print("Can't delete TEST_CON : {}".format(str(e)))
cmd = "nmcli c delete TEST_CON"
print_cmd(cmd)
sp.check_call(shlex.split(cmd))
print("TEST_CON is deleted")


@retry(max_attempts=5, delay=60)
def device_rescan():
print_head("Calling a rescan")
cmd = "nmcli d wifi rescan"
print_cmd(cmd)
retcode = sp.call(shlex.split(cmd))
if retcode != 0:
# Most often the rescan request fails because NM has itself started
# a scan in recent past, we should let these operations complete before
# attempting a connection
print("Scan request failed, allow other operations to complete (15s)")
time.sleep(15)
sp.check_call(shlex.split(cmd))
print()


Expand Down Expand Up @@ -183,48 +174,59 @@ def perform_ping_test(interface):
if target:
count = 5
result = ping(target, interface, count, 10)
if result["received"] == count:
return True
if result["received"] != count:
raise ValueError(
"{} packets expected but only {} received".format(
count, result["received"]
)
)

return False

@retry(max_attempts=5, delay=1)
def wait_for_connected(interface, essid):
cmd = (
"nmcli -m tabular -t -f GENERAL.STATE,GENERAL.CONNECTION "
"d show {}".format(interface)
)
print_cmd(cmd)
output = sp.check_output(shlex.split(cmd), universal_newlines=True)
print(output)
state, ssid = output.strip().splitlines()

def wait_for_connected(interface, essid, max_wait=5):
connected = False
attempts = 0
while not connected and attempts < max_wait:
cmd = (
"nmcli -m tabular -t -f GENERAL.STATE,GENERAL.CONNECTION "
"d show {}".format(interface)
)
print_cmd(cmd)
output = sp.check_output(shlex.split(cmd))
state, ssid = output.decode(sys.stdout.encoding).strip().splitlines()
if state.startswith("100") and ssid == essid:
print("Reached connected state with ESSID: {}".format(essid))
elif ssid != essid:
error_msg = (
"ERROR: did not reach connected state with ESSID: {}\n"
"ESSID mismatch:\n Excepted:{}\n Actually:{}"
).format(essid, ssid, essid)
raise SystemExit(error_msg)
elif not state.startswith("100"):
error_msg = "State is not connected: {}".format(state)
raise SystemExit(error_msg)
print()

if state.startswith("100") and ssid == essid:
connected = True
break

time.sleep(1)
attempts += 1
def connection(cmd, device):
print_head("Connection attempt")
print_cmd(cmd)
sp.check_call(shlex.split(cmd))

if connected:
print("Reached connected state with ESSID: {}".format(essid))
else:
print(
"ERROR: did not reach connected state with ESSID: {}".format(essid)
)
if ssid != essid:
print(
"ESSID mismatch:\n Excepted:{}\n Actually:{}".format(
ssid, essid
)
)
if not state.startswith("100"):
print("State is not connected: {}".format(state))
# Make sure the connection is brought up
turn_up_connection("TEST_CON")

print()
return connected
print_head("Ensure interface is connected")
wait_for_connected(device, "TEST_CON")

print_head("Display address")
print_address_info(device)

print_head("Display route table")
print_route_info()

print_head("Perform a ping test")
perform_ping_test(device)
print("Connection test passed\n")


def open_connection(args):
Expand All @@ -233,7 +235,6 @@ def open_connection(args):
# ipv6.method ignore : I believe that NM can report the device as Connected
# if an IPv6 address is setup. This should ensure in
# this test we are using IPv4
print_head("Connection attempt")
cmd = (
"nmcli c add con-name TEST_CON "
"ifname {} "
Expand All @@ -244,31 +245,7 @@ def open_connection(args):
"ipv4.dhcp-timeout 30 "
"ipv6.method ignore".format(args.device, args.essid)
)
print_cmd(cmd)
sp.call(shlex.split(cmd))

# Make sure the connection is brought up
turn_up_connection("TEST_CON")

print_head("Ensure interface is connected")
reached_connected = wait_for_connected(args.device, "TEST_CON")

rc = 1
if reached_connected:
print_head("Display address")
print_address_info(args.device)

print_head("Display route table")
print_route_info()

print_head("Perform a ping test")
test_result = perform_ping_test(args.device)
if test_result:
rc = 0
print("Connection test passed\n")
else:
print("Connection test failed\n")
return rc
connection(cmd, args.device)


def secured_connection(args):
Expand All @@ -277,7 +254,6 @@ def secured_connection(args):
# ipv6.method ignore : I believe that NM can report the device as Connected
# if an IPv6 address is setup. This should ensure in
# this test we are using IPv4
print_head("Connection attempt")
cmd = (
"nmcli c add con-name TEST_CON "
"ifname {} "
Expand All @@ -292,31 +268,7 @@ def secured_connection(args):
args.device, args.essid, args.exchange, args.psk
)
)
print_cmd(cmd)
sp.call(shlex.split(cmd))

# Make sure the connection is brought up
turn_up_connection("TEST_CON")

print_head("Ensure interface is connected")
reached_connected = wait_for_connected(args.device, "TEST_CON")

rc = 1
if reached_connected:
print_head("Display address")
print_address_info(args.device)

print_head("Display route table")
print_route_info()

print_head("Perform a ping test")
test_result = perform_ping_test(args.device)
if test_result:
rc = 0
print("Connection test passed\n")
else:
print("Connection test failed\n")
return rc
connection(cmd, args.device)


def hotspot(args):
Expand Down Expand Up @@ -414,6 +366,7 @@ def parser_args():
return args


@retry(max_attempts=5, delay=60)
def main():
args = parser_args()
start_time = datetime.datetime.now()
Expand All @@ -424,32 +377,33 @@ def main():

if args.test_type == "scan":
if not aps_dict:
print("Failed to find any APs")
return 1
raise SystemExit("Failed to find any access point.")
else:
print("Found {} access points".format(len(aps_dict)))
return 0
return

if not aps_dict:
print("Targed access points: {} not found".format(args.essid))
return 1
raise SystemExit(
"Targed access point: {} not found".format(args.essid)
)

if args.func:
delete_test_ap_ssid_connection()
activated_uuid = get_nm_activate_connection()
turn_down_nm_connections()
try:
result = args.func(args)
args.func(args)
except Exception:
# The test is not required to run as root, but root access is
# required for journal access so only attempt to print when e.g.
# running under Remote
if os.geteuid() == 0:
print_journal_entries(start_time)
raise
finally:
turn_up_connection(activated_uuid)
delete_test_ap_ssid_connection()

# The test is not required to run as root, but root access is required for
# journal access so only attempt to print when e.g. running under Remote
if result != 0 and os.geteuid() == 0:
print_journal_entries(start_time)
return result


if __name__ == "__main__":
sys.exit(main())
Loading

0 comments on commit 0952eb2

Please sign in to comment.