blob: d81a22325115cb8cbace5678e18a021d1e6b0f17 [file] [log] [blame]
# Copyright 2020 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Connect to an AP.
Description
-----------
Connect to an AP.
Test Procedure
--------------
Auto.
Dependency
----------
- connection_manager goofy plugin
Examples
--------
To run this test on DUT, add a test item in the test list:
.. test_list::
generic_wireless_examples:WirelessTests.ExampleWirelessConnect2G
To disconnect to all WiFi services.:
.. test_list::
generic_wireless_examples:WirelessTests.WirelessDisconnect
"""
import re
from typing import List, Optional
from cros.factory.device import device_utils
from cros.factory.goofy.plugins import plugin_controller
from cros.factory.test import session
from cros.factory.test import test_case
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import sync_utils
from cros.factory.utils import type_utils
SSID_RE = re.compile('SSID: (.*)$', re.MULTILINE)
class WirelessConnectTestArgs:
device_name: Optional[str]
service_name: List[dict]
retries: int
sleep_interval: int
class WirelessConnectTest(test_case.TestCase):
"""Basic wireless test class."""
related_components = (test_case.TestCategory.WIFI, )
ARGS = [
Arg('device_name', str, 'The wifi interface', default=None),
Arg('service_name', list,
'A list of wlan config. See net_utils.WLAN for more information',
default=[]),
Arg('retries', int, 'Times to retry.', default=10),
Arg('sleep_interval', int, 'Time to sleep.', default=3)
]
args: WirelessConnectTestArgs
def setUp(self):
self._dut = device_utils.CreateDUTInterface()
self._device_name = None
self._connection_manager = plugin_controller.GetPluginRPCProxy(
'connection_manager')
def _CheckConnected(self, ssid_list: List[str]) -> bool:
result = self._dut.CheckOutput(['iw', 'dev', self._device_name, 'link'],
log=True)
match = SSID_RE.search(result)
if match and match.group(1) in ssid_list:
return True
return False
def _CheckNotConnected(self):
result = self._dut.CheckOutput(['iw', 'dev', self._device_name, 'link'],
log=True)
return result.startswith('Not connected.')
def runTest(self):
self._device_name = self._dut.wifi.SelectInterface(self.args.device_name)
session.console.info('Selected device_name is %s.', self._device_name)
services = self.args.service_name
session.console.info('service = %r', services)
self._connection_manager.Reconnect(services)
ssid_list = [service.get('ssid') for service in services]
retry_wrapper = sync_utils.RetryDecorator(
max_attempt_count=self.args.retries,
interval_sec=self.args.sleep_interval, target_condition=bool)
try:
if ssid_list:
retry_wrapper(self._CheckConnected)(ssid_list)
else:
retry_wrapper(self._CheckNotConnected)()
except type_utils.MaxRetryError:
self.FailTask('Reach maximum retries.')
else:
self.PassTask()