#!/usr/bin/env python3.4
#
# Copyright 2018 - The Android Open Source Project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re
import random
import time
import acts.controllers.packet_capture as packet_capture
import acts.signals as signals
import acts_contrib.test_utils.wifi.rpm_controller_utils as rutils
import acts_contrib.test_utils.wifi.wifi_datastore_utils as dutils
import acts_contrib.test_utils.wifi.wifi_test_utils as wutils
from acts import asserts
from acts.base_test import BaseTestClass
from acts.controllers.ap_lib import hostapd_constants
from acts.test_decorators import test_tracker_info
from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest
WifiEnums = wutils.WifiEnums
WAIT_BEFORE_CONNECTION = 1
SINGLE_BAND = 1
DUAL_BAND = 2
TIMEOUT = 60
TEST = 'test_'
PING_ADDR = 'www.google.com'
NUM_LINK_PROBES = 3
PROBE_DELAY_SEC = 1
class WifiChaosTest(WifiBaseTest):
""" Tests for wifi IOT
Test Bed Requirement:
* One Android device
* Wi-Fi IOT networks visible to the device
"""
def __init__(self, configs):
BaseTestClass.__init__(self, configs)
self.generate_interop_tests()
def randomize_testcases(self):
"""Randomize the list of hosts and build a random order of tests,
based on SSIDs, keeping all the relevant bands of an AP together.
"""
temp_tests = list()
hosts = self.user_params['interop_host']
random.shuffle(hosts)
for host in hosts:
ssid_2g = None
ssid_5g = None
info = dutils.show_device(host)
# Based on the information in datastore, add to test list if
# AP has 2G band.
if 'ssid_2g' in info:
ssid_2g = info['ssid_2g']
temp_tests.append(TEST + ssid_2g)
# Based on the information in datastore, add to test list if
# AP has 5G band.
if 'ssid_5g' in info:
ssid_5g = info['ssid_5g']
temp_tests.append(TEST + ssid_5g)
self.tests = temp_tests
def generate_interop_testcase(self, base_test, testcase_name, ssid_dict):
"""Generates a single test case from the given data.
Args:
base_test: The base test case function to run.
testcase_name: The name of the test case.
ssid_dict: The information about the network under test.
"""
ssid = testcase_name
test_tracker_uuid = ssid_dict[testcase_name]['uuid']
hostname = ssid_dict[testcase_name]['host']
if not testcase_name.startswith('test_'):
testcase_name = 'test_%s' % testcase_name
test_case = test_tracker_info(
uuid=test_tracker_uuid)(lambda: base_test(ssid, hostname))
setattr(self, testcase_name, test_case)
self.tests.append(testcase_name)
def generate_interop_tests(self):
for ssid_dict in self.user_params['interop_ssid']:
testcase_name = list(ssid_dict)[0]
self.generate_interop_testcase(self.interop_base_test,
testcase_name, ssid_dict)
self.randomize_testcases()
def setup_class(self):
super().setup_class()
self.dut = self.android_devices[0]
self.admin = 'admin' + str(random.randint(1000001, 12345678))
wutils.wifi_test_device_init(self.dut)
# Set country code explicitly to "US".
wutils.set_wifi_country_code(self.dut, wutils.WifiEnums.CountryCode.US)
asserts.assert_true(
self.lock_pcap(),
"Could not lock a Packet Capture. Aborting Interop test.")
wutils.wifi_toggle_state(self.dut, True)
def lock_pcap(self):
"""Lock a Packet Capturere to use for the test."""
# Get list of devices from the datastore.
locked_pcap = False
devices = dutils.get_devices()
for device in devices:
device_name = device['hostname']
device_type = device['ap_label']
if device_type == 'PCAP' and not device['lock_status']:
if dutils.lock_device(device_name, self.admin):
self.pcap_host = device_name
host = device['ip_address']
self.log.info("Locked Packet Capture device: %s" %
device_name)
locked_pcap = True
break
else:
self.log.warning("Failed to lock %s PCAP." % device_name)
if not locked_pcap:
return False
pcap_config = {'ssh_config': {'user': 'root'}}
pcap_config['ssh_config']['host'] = host
self.pcap = packet_capture.PacketCapture(pcap_config)
return True
def setup_test(self):
super().setup_test()
self.dut.droid.wakeLockAcquireBright()
self.dut.droid.wakeUpNow()
def on_pass(self, test_name, begin_time):
wutils.stop_pcap(self.pcap, self.pcap_procs, True)
def on_fail(self, test_name, begin_time):
# Sleep to ensure all failed packets are captured.
time.sleep(5)
wutils.stop_pcap(self.pcap, self.pcap_procs, False)
super().on_fail(test_name, begin_time)
def teardown_test(self):
super().teardown_test()
self.dut.droid.wakeLockRelease()
self.dut.droid.goToSleepNow()
def teardown_class(self):
# Unlock the PCAP.
if not dutils.unlock_device(self.pcap_host):
self.log.warning("Failed to unlock %s PCAP. Check in datastore.")
"""Helper Functions"""
def scan_and_connect_by_id(self, network, net_id):
"""Scan for network and connect using network id.
Args:
net_id: Integer specifying the network id of the network.
"""
ssid = network[WifiEnums.SSID_KEY]
wutils.start_wifi_connection_scan_and_ensure_network_found(
self.dut, ssid)
wutils.wifi_connect_by_id(self.dut, net_id)
def run_ping(self, sec):
"""Run ping for given number of seconds.
Args:
sec: Time in seconds to run teh ping traffic.
"""
self.log.info("Finding Gateway...")
route_response = self.dut.adb.shell("ip route get 8.8.8.8")
gateway_ip = re.search('via (.*) dev', str(route_response)).group(1)
self.log.info("Gateway IP = %s" % gateway_ip)
self.log.info("Running ping for %d seconds" % sec)
result = self.dut.adb.shell("ping -w %d %s" % (sec, gateway_ip),
timeout=sec + 1)
self.log.debug("Ping Result = %s" % result)
if "100% packet loss" in result:
raise signals.TestFailure("100% packet loss during ping")
def send_link_probes(self, network):
"""
Send link probes, and verify that the device and AP did not crash.
Also verify that at least one link probe succeeded.
Steps:
1. Send a few link probes.
2. Ensure that the device and AP did not crash (by checking that the
device remains connected to the expected network).
"""
results = wutils.send_link_probes(self.dut, NUM_LINK_PROBES,
PROBE_DELAY_SEC)
self.log.info("Link Probe results: %s" % (results, ))
wifi_info = self.dut.droid.wifiGetConnectionInfo()
expected = network[WifiEnums.SSID_KEY]
actual = wifi_info[WifiEnums.SSID_KEY]
asserts.assert_equal(
expected, actual,
"Device did not remain connected after sending link probes!")
def unlock_and_turn_off_ap(self, hostname, rpm_port, rpm_ip):
"""UNlock the AP in datastore and turn off the AP.
Args:
hostname: Hostname of the AP.
rpm_port: Port number on the RPM for the AP.
rpm_ip: RPM's IP address.
"""
# Un-Lock AP in datastore.
self.log.debug("Un-lock AP in datastore")
if not dutils.unlock_device(hostname):
self.log.warning("Failed to unlock %s AP. Check AP in datastore.")
# Turn OFF AP from the RPM port.
rutils.turn_off_ap(rpm_port, rpm_ip)
def run_connect_disconnect(self, network, hostname, rpm_port, rpm_ip,
release_ap):
"""Run connect/disconnect to a given network in loop.
Args:
network: Dict, network information.
hostname: Hostanme of the AP to connect to.
rpm_port: Port number on the RPM for the AP.
rpm_ip: Port number on the RPM for the AP.
release_ap: Flag to determine if we should turn off the AP yet.
Raises: TestFailure if the network connection fails.
"""
for attempt in range(5):
try:
begin_time = time.time()
ssid = network[WifiEnums.SSID_KEY]
net_id = self.dut.droid.wifiAddNetwork(network)
asserts.assert_true(net_id != -1,
"Add network %s failed" % network)
self.log.info("Connecting to %s" % ssid)
self.scan_and_connect_by_id(network, net_id)
self.run_ping(10)
# TODO(b/133369482): uncomment once bug is resolved
# self.send_link_probes(network)
wutils.wifi_forget_network(self.dut, ssid)
time.sleep(WAIT_BEFORE_CONNECTION)
except Exception as e:
self.log.error("Connection to %s network failed on the %d "
"attempt with exception %s." %
(ssid, attempt, e))
# TODO:(bmahadev) Uncomment after scan issue is fixed.
# self.dut.take_bug_report(ssid, begin_time)
# self.dut.cat_adb_log(ssid, begin_time)
if release_ap:
self.unlock_and_turn_off_ap(hostname, rpm_port, rpm_ip)
raise signals.TestFailure("Failed to connect to %s" % ssid)
def get_band_and_chan(self, ssid):
"""Get the band and channel information from SSID.
Args:
ssid: SSID of the network.
"""
ssid_info = ssid.split('_')
self.band = ssid_info[-1]
for item in ssid_info:
# Skip over the router model part.
if 'ch' in item and item != ssid_info[0]:
self.chan = re.search(r'(\d+)', item).group(0)
return
raise signals.TestFailure("Channel information not found in SSID.")
def interop_base_test(self, ssid, hostname):
"""Base test for all the connect-disconnect interop tests.
Args:
ssid: string, SSID of the network to connect to.
hostname: string, hostname of the AP.
Steps:
1. Lock AP in datstore.
2. Turn on AP on the rpm switch.
3. Run connect-disconnect in loop.
4. Turn off AP on the rpm switch.
5. Unlock AP in datastore.
"""
network = {}
network['password'] = 'password'
network['SSID'] = ssid
release_ap = False
wutils.reset_wifi(self.dut)
# Lock AP in datastore.
self.log.info("Lock AP in datastore")
ap_info = dutils.show_device(hostname)
# If AP is locked by a different test admin, then we skip.
if ap_info['lock_status'] and ap_info['locked_by'] != self.admin:
raise signals.TestSkip("AP %s is locked, skipping test" % hostname)
if not dutils.lock_device(hostname, self.admin):
self.log.warning("Failed to lock %s AP. Unlock AP in datastore"
" and try again.")
raise signals.TestFailure("Failed to lock AP")
band = SINGLE_BAND
if ('ssid_2g' in ap_info) and ('ssid_5g' in ap_info):
band = DUAL_BAND
if (band == SINGLE_BAND) or (band == DUAL_BAND and '5G' in ssid):
release_ap = True
# Get AP RPM attributes and Turn ON AP.
rpm_ip = ap_info['rpm_ip']
rpm_port = ap_info['rpm_port']
rutils.turn_on_ap(self.pcap, ssid, rpm_port, rpm_ip=rpm_ip)
self.log.info("Finished turning ON AP.")
# Experimental. Some APs take upto a min to come online.
time.sleep(60)
self.get_band_and_chan(ssid)
self.pcap.configure_monitor_mode(self.band, self.chan)
self.pcap_procs = wutils.start_pcap(self.pcap, self.band.lower(),
self.test_name)
self.run_connect_disconnect(network, hostname, rpm_port, rpm_ip,
release_ap)
# Un-lock only if it's a single band AP or we are running the last band.
if release_ap:
self.unlock_and_turn_off_ap(hostname, rpm_port, rpm_ip)