auto-t: remove direct scan() call when possible

All tests which could avoid calling scan() directly have been
changed to use the 'full_scan' argument to get_ordered_network.

This was done because of unreliable scanning behavior on slower
systems, like VMs. If we get unlucky with the scheduler some beacons
are not received in time and in turn scan results are missing.
Using full_scan=True works around this issue by repeatedly scanning
until the SSID is found.
This commit is contained in:
James Prestwood 2022-03-30 13:23:11 -07:00 committed by Denis Kenzior
parent f97b53608d
commit a773aa6a07
14 changed files with 22 additions and 140 deletions

View File

@ -47,16 +47,8 @@ class Test(unittest.TestCase):
dev1.start_ap('TestAP2', 'Password2') dev1.start_ap('TestAP2', 'Password2')
try: try:
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
dev2.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(dev2, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
networks = {} networks = {}
networks['TestAP1'] = dev2.get_ordered_network('TestAP1') networks['TestAP1'] = dev2.get_ordered_network('TestAP1', full_scan=True)
networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True) networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True)
self.assertEqual(networks['TestAP1'].type, NetworkType.psk) self.assertEqual(networks['TestAP1'].type, NetworkType.psk)

View File

@ -29,14 +29,6 @@ class Test(unittest.TestCase):
dev1.start_ap('APConfig') dev1.start_ap('APConfig')
try: try:
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
dev2.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(dev2, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
networks = {} networks = {}
networks['APConfig'] = dev2.get_ordered_network('APConfig', full_scan=True) networks['APConfig'] = dev2.get_ordered_network('APConfig', full_scan=True)

View File

@ -26,14 +26,6 @@ class Test(unittest.TestCase):
dev4.start_ap('TestAP4', 'Password4') dev4.start_ap('TestAP4', 'Password4')
try: try:
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
dev2.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(dev2, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
networks = {} networks = {}
networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True) networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True)

View File

@ -48,16 +48,8 @@ class Test(unittest.TestCase):
dev1.start_ap('TestAP2', 'Password2') dev1.start_ap('TestAP2', 'Password2')
try: try:
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
dev2.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(dev2, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(dev2, condition)
networks = {} networks = {}
networks['TestAP1'] = dev2.get_ordered_network('TestAP1') networks['TestAP1'] = dev2.get_ordered_network('TestAP1', full_scan=True)
networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True) networks['TestAP2'] = dev2.get_ordered_network('TestAP2', full_scan=True)
self.assertEqual(networks['TestAP1'].type, NetworkType.psk) self.assertEqual(networks['TestAP1'].type, NetworkType.psk)

View File

@ -244,15 +244,7 @@ class Test(unittest.TestCase):
device.disconnect() device.disconnect()
condition = 'not obj.scanning' ordered_network = device.get_ordered_network("TestBlacklist", full_scan=True)
wd.wait_for_object_condition(device, condition)
device.scan()
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
ordered_network = device.get_ordered_network("TestBlacklist")
self.assertEqual(ordered_network.type, NetworkType.psk) self.assertEqual(ordered_network.type, NetworkType.psk)

View File

@ -11,13 +11,9 @@ import subprocess
from config import ctx from config import ctx
class Test(unittest.TestCase): class Test(unittest.TestCase):
def check_connection_success(self, ssid): def check_connection_success(self, ssid):
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
condition = 'not obj.scanning'
self.wd.wait_for_object_condition(device, condition)
ordered_network = device.get_ordered_network(ssid) ordered_network = device.get_ordered_network(ssid)
condition = 'obj.state == DeviceState.connected' condition = 'obj.state == DeviceState.connected'
@ -28,9 +24,14 @@ class Test(unittest.TestCase):
condition = 'not obj.connected' condition = 'not obj.connected'
self.wd.wait_for_object_condition(ordered_network.network_object, condition) self.wd.wait_for_object_condition(ordered_network.network_object, condition)
def establish_network(self, ssid):
device = self.wd.list_devices(1)[0]
device.get_ordered_network(ssid)
def test_connection_with_passphrase(self): def test_connection_with_passphrase(self):
ssid = 'ssidPassphrase' ssid = 'ssidPassphrase'
self.establish_network(ssid)
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
# Use --dontaks cmd-line option # Use --dontaks cmd-line option
@ -45,6 +46,7 @@ class Test(unittest.TestCase):
def test_connection_with_username_and_password(self): def test_connection_with_username_and_password(self):
ssid = 'ssidUNameAndPWord' ssid = 'ssidUNameAndPWord'
self.establish_network(ssid)
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
ctx.start_process(['iwctl', '-u', 'user', '-p', 'password', 'station', \ ctx.start_process(['iwctl', '-u', 'user', '-p', 'password', 'station', \
@ -54,6 +56,7 @@ class Test(unittest.TestCase):
def test_connection_with_password(self): def test_connection_with_password(self):
ssid = 'ssidPWord' ssid = 'ssidPWord'
self.establish_network(ssid)
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
ctx.start_process(['iwctl', '-p', 'password', 'station', device.name, 'connect', ssid], ctx.start_process(['iwctl', '-p', 'password', 'station', device.name, 'connect', ssid],
@ -64,6 +67,7 @@ class Test(unittest.TestCase):
def test_connection_failure(self): def test_connection_failure(self):
ssid = 'ssidPassphrase' ssid = 'ssidPassphrase'
self.establish_network(ssid)
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
with self.assertRaises(subprocess.CalledProcessError): with self.assertRaises(subprocess.CalledProcessError):
@ -73,14 +77,13 @@ class Test(unittest.TestCase):
def test_invalid_command_line_option(self): def test_invalid_command_line_option(self):
ssid = 'ssidPassphrase' ssid = 'ssidPassphrase'
self.establish_network(ssid)
device = self.wd.list_devices(1)[0] device = self.wd.list_devices(1)[0]
with self.assertRaises(subprocess.CalledProcessError): with self.assertRaises(subprocess.CalledProcessError):
ctx.start_process(['iwctl', '-z', 'station', device.name, 'connect', ssid], check=True) ctx.start_process(['iwctl', '-z', 'station', device.name, 'connect', ssid], check=True)
def test_invalid_command(self): def test_invalid_command(self):
device = self.wd.list_devices(1)[0]
with self.assertRaises(subprocess.CalledProcessError): with self.assertRaises(subprocess.CalledProcessError):
ctx.start_process(['iwctl', 'inexistent', 'command'], check=True) ctx.start_process(['iwctl', 'inexistent', 'command'], check=True)
@ -91,19 +94,6 @@ class Test(unittest.TestCase):
cls.wd = IWD() cls.wd = IWD()
device = cls.wd.list_devices(1)[0]
condition = 'not obj.scanning'
cls.wd.wait_for_object_condition(device, condition)
device.scan()
condition = 'obj.scanning'
cls.wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
cls.wd.wait_for_object_condition(device, condition)
@classmethod @classmethod
def tearDownClass(cls): def tearDownClass(cls):
IWD.clear_storage() IWD.clear_storage()

View File

@ -21,12 +21,7 @@ class Test(unittest.TestCase):
hostapd = HostapdCLI(config='ssidOpen.conf') hostapd = HostapdCLI(config='ssidOpen.conf')
device.scan() ordered_network = device.get_ordered_network('ssidOpen', full_scan=True)
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
ordered_network = device.get_ordered_network('ssidOpen')
ordered_network.network_object.connect() ordered_network.network_object.connect()

View File

@ -13,21 +13,10 @@ class Test(unittest.TestCase):
def validate_connection(self, wd): def validate_connection(self, wd):
device = wd.list_devices(1)[0] device = wd.list_devices(1)[0]
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
# Scanning is unavoidable in this case since both FILS-SHA256 and # Scanning is unavoidable in this case since both FILS-SHA256 and
# FILS-SHA384 are tested. Without a new scan the cached scan results # FILS-SHA384 are tested. Without a new scan the cached scan results
# would cause IWD to choose an incorrect AKM for the second test. # would cause IWD to choose an incorrect AKM for the second test.
device.scan() ordered_network = device.get_ordered_network('TestFT', full_scan=True)
condition = 'obj.scanning'
wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
ordered_network = device.get_ordered_network('TestFT')
self.assertEqual(ordered_network.type, NetworkType.eap) self.assertEqual(ordered_network.type, NetworkType.eap)

View File

@ -14,22 +14,11 @@ from hostapd import HostapdCLI
class Test(unittest.TestCase): class Test(unittest.TestCase):
def do_connect(self, wd, device, hostapd): def do_connect(self, wd, device, hostapd):
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
device.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
# #
# Scanning must be explicitly done to get updated RSSI values. Therefore # Scanning must be explicitly done to get updated RSSI values. Therefore
# scan_if_needed is set false because of the previous scan. # full_scan is set.
# #
ordered_network = device.get_ordered_network('testSSID', scan_if_needed=False) ordered_network = device.get_ordered_network('testSSID', full_scan=True)
self.assertEqual(ordered_network.type, NetworkType.psk) self.assertEqual(ordered_network.type, NetworkType.psk)

View File

@ -47,12 +47,7 @@ class Test(unittest.TestCase):
testutil.test_ip_address_match(dev1.name, '192.168.1.10') testutil.test_ip_address_match(dev1.name, '192.168.1.10')
dev2.scan() ordered_network = dev2.get_ordered_network('ssidTKIP')
condition = 'not obj.scanning'
wd_ns0.wait_for_object_condition(dev2, condition)
ordered_network = dev2.get_ordered_network('ssidTKIP', scan_if_needed=True)
condition = 'not obj.connected' condition = 'not obj.connected'
wd_ns0.wait_for_object_condition(ordered_network.network_object, condition) wd_ns0.wait_for_object_condition(ordered_network.network_object, condition)

View File

@ -230,16 +230,6 @@ class Test(unittest.TestCase):
self.rule0.signal = -4000 self.rule0.signal = -4000
self.rule0.enabled = True self.rule0.enabled = True
devices = self.wd.list_devices(1)
device = devices[0]
device.scan()
condition = 'obj.scanning'
self.wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
self.wd.wait_for_object_condition(device, condition)
self.validate(self.wd, self.hapd_owe) self.validate(self.wd, self.hapd_owe)
def setUp(self): def setUp(self):

View File

@ -31,18 +31,7 @@ class Test(unittest.TestCase):
agent.calls = 0 agent.calls = 0
device.register_signal_agent(agent, [-20, -40, -60, -80]) device.register_signal_agent(agent, [-20, -40, -60, -80])
condition = 'not obj.scanning' ordered_network = device.get_ordered_network('TestOpen', full_scan=True)
wd.wait_for_object_condition(device, condition)
device.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
ordered_network = device.get_ordered_network('TestOpen')
self.assertEqual(ordered_network.type, NetworkType.open) self.assertEqual(ordered_network.type, NetworkType.open)
self.assertEqual(ordered_network.signal_strength, -4000) self.assertEqual(ordered_network.signal_strength, -4000)

View File

@ -21,16 +21,8 @@ class Test(unittest.TestCase):
devices = wd.list_devices(4) devices = wd.list_devices(4)
self.assertIsNotNone(devices) self.assertIsNotNone(devices)
for d in devices:
d.disconnect()
d.scan()
wd.wait_for_object_condition(d, 'obj.scanning')
for d in devices:
wd.wait_for_object_condition(d, 'not obj.scanning')
for i in range(len(devices)): for i in range(len(devices)):
network = devices[i].get_ordered_network('ssidSAE-Clogging') network = devices[i].get_ordered_network('ssidSAE-Clogging', full_scan=True)
self.assertEqual(network.type, NetworkType.psk) self.assertEqual(network.type, NetworkType.psk)
networks.append(network) networks.append(network)

View File

@ -16,16 +16,9 @@ class Test(unittest.TestCase):
def validate_connection(self, wd, ft=True): def validate_connection(self, wd, ft=True):
device = wd.list_devices(1)[0] device = wd.list_devices(1)[0]
condition = 'not obj.scanning' # This won't guarantee all BSS's are found, but at least ensures that
wd.wait_for_object_condition(device, condition) # at least one will be.
device.get_ordered_network('TestFT', full_scan=True)
device.scan()
condition = 'obj.scanning'
wd.wait_for_object_condition(device, condition)
condition = 'not obj.scanning'
wd.wait_for_object_condition(device, condition)
self.assertFalse(self.bss_hostapd[0].list_sta()) self.assertFalse(self.bss_hostapd[0].list_sta())
self.assertFalse(self.bss_hostapd[1].list_sta()) self.assertFalse(self.bss_hostapd[1].list_sta())