Kernel/iwd
3
0
Fork 0
mirror of https://git.kernel.org/pub/scm/network/wireless/iwd.git synced 2024-12-22 13:02:44 +01:00
Code Releases Activity

autotests: Test ACD failure during connect

Browse Source 
Previously we had an ACD failure scenario where a new client forces its
IP to create an IP conflict and an already-connected client detects the
conflict and reacts.  Now first test a scenario where a newly connecting
IWD client runs ACD before setting its statically configured IP, detects
a conflict and refuses to continue, then run the second scenario where
the newly connecting DHCP-configured client ignores the conflict and
starts ACD in defend-indefinitely mode and the older client in
defent-once mode gives up its IP.
This commit is contained in:
Andrew Zaborowski 2022-08-23 18:36:16 +02:00 committed by Denis Kenzior
parent 818ec70549
commit e28243a584
1 changed files with 61 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
autotests/testNetconfig/static_test.py
View File

@ -13,36 +13,28 @@ import testutil
from config import ctx
import os
import socket
import gc
class Test(unittest.TestCase):
def test_connection_success(self):
# Use a non-default storage_dir for one of the instances, the default for the other one
wd = IWD(True, iwd_storage_dir='/tmp/storage')
ns0 = ctx.get_namespace('ns0')
wd_ns0 = IWD(True, namespace=ns0)
psk_agent = PSKAgent("secret123")
psk_agent_ns0 = PSKAgent("secret123", namespace=ns0)
wd.register_psk_agent(psk_agent)
wd_ns0.register_psk_agent(psk_agent_ns0)
dev1 = wd.list_devices(1)[0]
dev2 = wd_ns0.list_devices(1)[0]
iwd_main = IWD(True, iwd_storage_dir='/tmp/storage-main')
psk_agent_main = PSKAgent("secret123")
iwd_main.register_psk_agent(psk_agent_main)
dev1 = iwd_main.list_devices(1)[0]
ordered_network = dev1.get_ordered_network('ap-main')
self.assertEqual(ordered_network.type, NetworkType.psk)
condition = 'not obj.connected'
wd.wait_for_object_condition(ordered_network.network_object, condition)
iwd_main.wait_for_object_condition(ordered_network.network_object, condition)
ordered_network.network_object.connect()
condition = 'obj.state == DeviceState.connected'
wd.wait_for_object_condition(dev1, condition)
iwd_main.wait_for_object_condition(dev1, condition)
testutil.test_iface_operstate()
testutil.test_ifaces_connected()
@ -80,10 +72,56 @@ class Test(unittest.TestCase):
# of the log since we care about the end result here.
self.assertEqual(expected_rclog, entries[-3:])
# Run our second client in a separate namespace to allow ACD (ARP) to
# work, and also be able to set identical IPs on both interfaces for
# the next part of this test.
ns0 = ctx.get_namespace('ns0')
iwd_ns0_1 = IWD(True, namespace=ns0, iwd_storage_dir='/tmp/storage-ns0-1')
psk_agent_ns0_1 = PSKAgent("secret123", namespace=ns0)
iwd_ns0_1.register_psk_agent(psk_agent_ns0_1)
dev2 = iwd_ns0_1.list_devices(1)[0]
ordered_network = dev2.get_ordered_network('ap-main')
condition = 'not obj.connected'
wd_ns0.wait_for_object_condition(ordered_network.network_object, condition)
iwd_ns0_1.wait_for_object_condition(ordered_network.network_object, condition)
# Attempt a connection to the same AP that iwd_main is connected to
# using the same static config. The new client's ACD client should
# detect an IP conflict and not allow the device to reach the
# "connected" state although the DBus .Connect call will succeed.
ordered_network.network_object.connect()
self.assertEqual(dev2.state, iwd.DeviceState.connecting)
try:
# We should either stay in "connecting" indefinitely or move to
# "disconnecting"
condition = 'obj.state != DeviceState.connecting'
iwd_ns0_1.wait_for_object_condition(dev2, condition, max_wait=21)
self.assertEqual(dev2.state, iwd.DeviceState.disconnecting)
except TimeoutError:
dev2.disconnect()
iwd_ns0_1.unregister_psk_agent(psk_agent_ns0_1)
del dev2
# Note: if any references to iwd_ns0_1 are left, the "del iwd_ns0_1"
# will not kill the IWD process the iwd_ns0_2 initialization will raise
# an exception. The iwd_ns0_1.wait_for_object_condition() above
# creates a circular reference (which is not wrong in itself) and
# gc.collect() gets rid of it. The actual solution is to eventually
# avoid executing anything important in .__del__ (which is wrong.)
gc.collect()
del iwd_ns0_1
iwd_ns0_2 = IWD(True, namespace=ns0, iwd_storage_dir='/tmp/storage-ns0-2')
psk_agent_ns0_2 = PSKAgent("secret123", namespace=ns0)
iwd_ns0_2.register_psk_agent(psk_agent_ns0_2)
dev2 = iwd_ns0_2.list_devices(1)[0]
ordered_network = dev2.get_ordered_network('ap-main')
condition = 'not obj.connected'
iwd_ns0_2.wait_for_object_condition(ordered_network.network_object, condition)
# Connect to the same network from a dynamically configured client. We
# block ICMP pings so that the DHCP server can't confirm that
@ -98,9 +136,9 @@ class Test(unittest.TestCase):
ordered_network.network_object.connect()
condition = 'obj.state == DeviceState.connected'
wd_ns0.wait_for_object_condition(dev2, condition)
iwd_ns0_2.wait_for_object_condition(dev2, condition)
wd.wait(1)
iwd_main.wait(1)
# Check dev1 is now disconnected or without its IPv4 address
if dev1.state == iwd.DeviceState.connected:
testutil.test_ip_address_match(dev1.name, None)
@ -109,9 +147,9 @@ class Test(unittest.TestCase):
dev2.disconnect()
condition = 'not obj.connected'
wd.wait_for_object_condition(ordered_network.network_object, condition)
iwd_main.wait_for_object_condition(ordered_network.network_object, condition)
wd.unregister_psk_agent(psk_agent)
iwd_main.unregister_psk_agent(psk_agent_main)
@classmethod
def setUpClass(cls):
@ -132,7 +170,8 @@ class Test(unittest.TestCase):
cls.dhcpd_pid = ctx.start_process(['dhcpd', '-f', '-cf', '/tmp/dhcpd.conf',
'-lf', '/tmp/dhcpd.leases',
hapd.ifname], cleanup=remove_lease)
IWD.copy_to_storage('static.psk', '/tmp/storage', 'ap-main.psk')
IWD.copy_to_storage('static.psk', '/tmp/storage-main', 'ap-main.psk')
IWD.copy_to_storage('static.psk', '/tmp/storage-ns0-1', 'ap-main.psk')
cls.orig_path = os.environ['PATH']
os.environ['PATH'] = '/tmp/test-bin:' + os.environ['PATH']
@ -141,7 +180,7 @@ class Test(unittest.TestCase):
@classmethod
def tearDownClass(cls):
cls.dhcpd_pid.kill()
os.system('rm -rf /tmp/resolvconf.log /tmp/test-bin /tmp/storage')
os.system('rm -rf /tmp/resolvconf.log /tmp/test-bin /tmp/storage-*')
os.environ['PATH'] = cls.orig_path
if __name__ == '__main__':