# # 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 threading import time from acts import asserts from acts import base_test from acts import utils from acts.controllers import adb from acts.controllers.adb_lib.error import AdbError from acts.controllers.ap_lib import hostapd_constants from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.net import connectivity_const as cconst from acts_contrib.test_utils.net import connectivity_test_utils as cutils from acts_contrib.test_utils.net import net_test_utils as nutils from acts_contrib.test_utils.net.net_test_utils import start_tcpdump from acts_contrib.test_utils.net.net_test_utils import stop_tcpdump from acts_contrib.test_utils.tel import tel_test_utils as ttutils from acts_contrib.test_utils.tel.tel_test_utils import get_operator_name from acts_contrib.test_utils.tel.tel_test_utils import http_file_download_by_chrome from acts_contrib.test_utils.wifi import wifi_test_utils as wutils import queue from queue import Empty conn_test_class = "com.android.tests.connectivity.uid.ConnectivityTestActivity" android_os_class = "com.quicinc.cne.CNEService.CNEServiceApp" instr_cmd = ("am instrument -w -e command grant-all " "com.android.permissionutils/.PermissionInstrumentation") HOUR_IN_MILLIS = 1000 * 60 * 60 BYTE_TO_MB_ANDROID = 1000.0 * 1000.0 BYTE_TO_MB = 1024.0 * 1024.0 DOWNLOAD_PATH = "/sdcard/download/" DATA_USG_ERR = 2.2 DATA_ERR = 0.2 TIMEOUT = 2 * 60 INC_DATA = 10 AP_PASSPHRASE_LENGTH_2G = hostapd_constants.AP_PASSPHRASE_LENGTH_2G AP_SSID_LENGTH_2G = hostapd_constants.AP_SSID_LENGTH_2G RETRY_SLEEP = 3 WIFI_SLEEP = 180 class DataUsageTest(base_test.BaseTestClass): """Data usage tests. Requirements: Two Pixel devices - one device with TMO and other with VZW SIM WiFi network - Wifi network to test wifi data usage """ def __init__(self, controllers): base_test.BaseTestClass.__init__(self, controllers) def setup_class(self): """Setup devices for tests and unpack params.""" # unpack user params req_params = ("wifi_network", "file_size", "download_file_ipv4", "download_file_ipv6", "dbs_supported_models") self.unpack_userparams(req_params) self.file_path = DOWNLOAD_PATH + self.download_file_ipv4.split("/")[-1] self.file_size = float(self.file_size) for ad in self.android_devices: self.log.info("Operator on %s is %s" % \ (ad.serial, get_operator_name(self.log, ad))) ad.reboot() nutils.verify_lte_data_and_tethering_supported(ad) nutils.set_chrome_browser_permissions(ad) cutils.set_private_dns(ad, cconst.PRIVATE_DNS_MODE_OFF) try: ad.adb.shell(instr_cmd) except AdbError: self.log.warn("cmd %s failed on %s" % (instr_cmd, ad.serial)) self.tcpdumps = [] self.hs_enabled = [] def setup_test(self): for ad in self.android_devices: self.tcpdumps.append(start_tcpdump(ad, self.test_name)) ad.droid.wakeLockAcquireBright() ad.droid.wakeUpNow() def teardown_test(self): for ad in self.hs_enabled: try: wutils.stop_wifi_tethering(ad) except Exception as e: self.log.warn("Failed to stop wifi tethering: %s" % e) self.hs_enabled = [] for i in range(len(self.android_devices)): ad = self.android_devices[i] stop_tcpdump(ad, self.tcpdumps[i], self.test_name) wutils.reset_wifi(ad) sub_id = str(ad.droid.telephonyGetSubscriberId()) ad.droid.connectivityFactoryResetNetworkPolicies(sub_id) ad.droid.telephonyToggleDataConnection(True) ad.droid.wakeLockRelease() ad.droid.goToSleepNow() self.tcpdumps = [] def teardown_class(self): for ad in self.android_devices: cutils.set_private_dns(ad, cconst.PRIVATE_DNS_MODE_OPPORTUNISTIC) def on_fail(self, test_name, begin_time): for ad in self.android_devices: ad.take_bug_report(test_name, begin_time) ### Helper functions def _download_data_through_app(self, ad, file_link): """Download data through app on DUT. Args: ad: DUT to download the file on file_link: http link of the file to download Returns: True if file download is successful, False if not """ intent = ad.droid.createIntentForClassName(conn_test_class) json_obj = {"url": file_link} ad.droid.launchForResultWithIntent(intent, json_obj) download_status = False end_time = time.time() + TIMEOUT while time.time() < end_time: download_status = ttutils._check_file_existance( ad, self.file_path, self.file_size * BYTE_TO_MB) if download_status: self.log.info("Delete file: %s", self.file_path) ad.adb.shell("rm %s" % self.file_path, ignore_status=True) break time.sleep(RETRY_SLEEP*2) # wait to check if download is complete return download_status def _get_total_data_usage_for_device(self, ad, conn_type): """Get total data usage in MB for device. Args: ad: android device object conn_type: MOBILE/WIFI data usage Returns: Data usage in MB """ sub_id = str(ad.droid.telephonyGetSubscriberId()) end_time = int(time.time() * 1000) + 2 * HOUR_IN_MILLIS data_usage = ad.droid.connectivityQuerySummaryForDevice( conn_type, sub_id, 0, end_time) data_usage /= BYTE_TO_MB_ANDROID self.log.info("Total data usage is: %s" % data_usage) return data_usage def _get_data_usage_for_uid_rx(self, ad, conn_type, uid): """Get data usage for UID in Rx Bytes. Args: ad: DUT to get data usage from conn_type: MOBILE/WIFI data usage uid: UID of the app Returns: Data usage in MB """ subscriber_id = ad.droid.telephonyGetSubscriberId() end_time = int(time.time() * 1000) + 2 * HOUR_IN_MILLIS data_usage = ad.droid.connectivityQueryDetailsForUidRxBytes( conn_type, subscriber_id, 0, end_time, uid) return data_usage/BYTE_TO_MB_ANDROID def _get_data_usage_for_device_rx(self, ad, conn_type): """Get total data usage in rx bytes for device. Args: ad: DUT to get data usage from conn_type: MOBILE/WIFI data usage Returns: Data usage in MB """ subscriber_id = ad.droid.telephonyGetSubscriberId() end_time = int(time.time() * 1000) + 2 * HOUR_IN_MILLIS data_usage = ad.droid.connectivityQuerySummaryForDeviceRxBytes( conn_type, subscriber_id, 0, end_time) return data_usage/BYTE_TO_MB_ANDROID def _get_data_usage(self, ad, conn_type): """Get data usage. Args: ad: DUT to get data usage from conn_type: MOBILE/WIFI data usage Returns: Tuple of Android Os app, Conn UID app, Total data usages """ android_os_uid = ad.droid.getUidForPackage(android_os_class) conn_test_uid = ad.droid.getUidForPackage(conn_test_class) aos = self._get_data_usage_for_uid_rx(ad, conn_type, android_os_uid) app = self._get_data_usage_for_uid_rx(ad, conn_type, conn_test_uid) tot = self._get_data_usage_for_device_rx(ad, conn_type) self.log.info("Android Os data usage: %s" % aos) self.log.info("Conn UID Test data usage: %s" % app) self.log.info("Total data usage: %s" % tot) return (aos, app, tot) def _listen_for_network_callback(self, ad, event, q): """Verify network callback for data usage. Args: ad: DUT to get the network callback for event: Network callback event q: queue object Returns: True: if the expected network callback found, False if not """ key = ad.droid.connectivityRegisterDefaultNetworkCallback() ad.droid.connectivityNetworkCallbackStartListeningForEvent(key, event) curr_time = time.time() status = False while time.time() < curr_time + TIMEOUT: try: nc_event = ad.ed.pop_event("NetworkCallback") self.log.info("Received: %s" % nc_event["data"]["networkCallbackEvent"]) if nc_event["data"]["networkCallbackEvent"] == event: status = True break except Empty: pass ad.droid.connectivityNetworkCallbackStopListeningForEvent(key, event) ad.droid.connectivityUnregisterNetworkCallback(key) q.put(status) def _test_data_usage_downlink(self, dut_a, dut_b, file_link, data_usage, tethering=False, connect_wifi=False, app_inc=True): """Verify data usage. Steps: 1. Get the current data usage of ConnUIDTest and Android OS apps 2. DUT is on LTE data (on wifi if connect_wifi is True) 3. Download file of size xMB through ConnUIDTest app 4. Verify that data usage of Android OS app did not change 5. Verify that data usage of ConnUIDTest app increased by ~xMB 6. Verify that data usage of device also increased by ~xMB Args: dut_a: DUT on which data usage need to be verified dut_b: DUT on which data is downloaded file_link: Link used to download data data_usage: specifies if MOBILE or WIFI data is calculated tethering: if True, enable hotspot on dut_a and connect dut_b to it connect_wifi: if True, connect dut_a to a wifi network app_inc: if True, verify app data usage on dut_a is increased """ if connect_wifi: dut_a.droid.telephonyToggleDataConnection(False) wutils.start_wifi_connection_scan_and_ensure_network_found( dut_a, self.wifi_network["SSID"]) wutils.wifi_connect(dut_a, self.wifi_network) # sleep for some time after connecting to wifi. some apps use wifi # data when connected initially. waiting after connecting to wifi # ensures correct wifi data usage calculation time.sleep(WIFI_SLEEP) if tethering: ssid = "hs_%s" % utils.rand_ascii_str(AP_SSID_LENGTH_2G) pwd = utils.rand_ascii_str(AP_PASSPHRASE_LENGTH_2G) network = {wutils.WifiEnums.SSID_KEY: ssid, wutils.WifiEnums.PWD_KEY: pwd} dut_b.droid.telephonyToggleDataConnection(False) wutils.start_wifi_tethering(dut_a, network[wutils.WifiEnums.SSID_KEY], network[wutils.WifiEnums.PWD_KEY]) self.hs_enabled.append(dut_a) wutils.start_wifi_connection_scan_and_ensure_network_found( dut_b, network[wutils.WifiEnums.SSID_KEY]) wutils.wifi_connect(dut_b, network) # sleep for some time after connecting to wifi. though wifi hotspot # is metered, some apps would still use wifi data when connected # initially. this ensures correct wifi data usage calculation time.sleep(WIFI_SLEEP) # get pre data usage (aos_pre, app_pre, total_pre) = self._get_data_usage(dut_a, data_usage) # download file through app status = self._download_data_through_app(dut_b, file_link) asserts.assert_true(status, "Failed to download file: %s" % file_link) # get new data usage aos_exp = DATA_ERR app_exp = self.file_size + DATA_USG_ERR file_size = self.file_size if not app_inc: app_exp = DATA_ERR file_size = 0 total_exp = self.file_size + DATA_USG_ERR # somtimes data usage is not increased immediately. # re-tries until the data usage is closed to expected value. curr_time = time.time() while time.time() < curr_time + TIMEOUT: (aos_pst, app_pst, total_pst) = self._get_data_usage(dut_a, data_usage) if total_pst - total_pre >= self.file_size and \ app_pst - app_pre >= file_size: # wait for some time to verify that data doesn't increase time.sleep(RETRY_SLEEP*2) break time.sleep(RETRY_SLEEP) # wait before retry (aos_pst, app_pst, total_pst) = self._get_data_usage(dut_a, data_usage) # verify data usage aos_diff = aos_pst - aos_pre app_diff = app_pst - app_pre total_diff = total_pst - total_pre self.log.info("Usage of Android OS increased: %sMB, expected: %sMB" % \ (aos_diff, aos_exp)) self.log.info("Usage of ConnUID app increased: %sMB, expected: %sMB" % \ (app_diff, app_exp)) self.log.info("Usage on the device increased: %sMB, expected: %sMB" % \ (total_diff, total_exp)) asserts.assert_true( (aos_diff < aos_exp) and (file_size <= app_diff < app_exp) and \ (self.file_size <= total_diff < total_exp), "Incorrect data usage count") def _test_data_usage_limit_downlink(self, dut_a, dut_b, file_link, data_usage, tethering=False): """Verify data usage limit reached. Steps: 1. Set the data usage limit to current data usage + 10MB 2. If tested for tethered device, start wifi hotspot and connect DUT to it. 3. If tested for tethered device, download 20MB data from tethered device else download on the same DUT 4. Verify file download stops and data limit reached Args: dut_a: DUT on which data usage limit needs to be verified dut_b: DUT on which data needs to be downloaded file_link: specifies if the link is IPv4 or IPv6 data_usage: specifies if MOBILE or WIFI data usage tethering: if wifi hotspot should be enabled on dut_a """ sub_id = str(dut_a.droid.telephonyGetSubscriberId()) # enable hotspot and connect dut_b to it. if tethering: ssid = "hs_%s" % utils.rand_ascii_str(AP_SSID_LENGTH_2G) pwd = utils.rand_ascii_str(AP_PASSPHRASE_LENGTH_2G) network = {wutils.WifiEnums.SSID_KEY: ssid, wutils.WifiEnums.PWD_KEY: pwd} dut_b.droid.telephonyToggleDataConnection(False) wutils.start_wifi_tethering(dut_a, network[wutils.WifiEnums.SSID_KEY], network[wutils.WifiEnums.PWD_KEY]) self.hs_enabled.append(dut_a) wutils.start_wifi_connection_scan_and_ensure_network_found( dut_b, network[wutils.WifiEnums.SSID_KEY]) wutils.wifi_connect(dut_b, network) # get pre mobile data usage total_pre = self._get_total_data_usage_for_device(dut_a, data_usage) # set data usage limit to current usage limit + 10MB self.log.info("Set data usage limit to %sMB" % (total_pre + INC_DATA)) dut_a.droid.connectivitySetDataUsageLimit( sub_id, int((total_pre + INC_DATA) * BYTE_TO_MB_ANDROID)) # download file on dut_b and look for BlockedStatusChanged callback q = queue.Queue() t = threading.Thread(target=self._listen_for_network_callback, args=(dut_a, "BlockedStatusChanged", q)) t.daemon = True t.start() status = http_file_download_by_chrome( dut_b, file_link, self.file_size, True, TIMEOUT) t.join() cb_res = q.get() # verify file download fails and expected callback is recevied asserts.assert_true(cb_res, "Failed to verify blocked status network callback") asserts.assert_true(not status, "File download successful. Expected to fail") ### Test Cases @test_tracker_info(uuid="b2d9b36c-3a1c-47ca-a9c1-755450abb20c") def test_mobile_data_usage_downlink_ipv4_tmo(self): """Verify mobile data usage over IPv4 and TMO carrier.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv4, cconst.TYPE_MOBILE) @test_tracker_info(uuid="fbbb58ed-d573-4bbd-bd5d-0bc540507896") def test_mobile_data_usage_downlink_ipv6_tmo(self): """Verify mobile data usage over IPv6 and TMO carrier.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv6, cconst.TYPE_MOBILE) @test_tracker_info(uuid="6562d626-2271-4d93-96c0-f33138635330") def test_mobile_data_usage_downlink_ipv4_vzw(self): """Verify mobile data usage over IPv4 and VZW carrier.""" self._test_data_usage_downlink(self.android_devices[1], self.android_devices[1], self.download_file_ipv4, cconst.TYPE_MOBILE) @test_tracker_info(uuid="2edf94cf-d937-459a-a7e4-2c679803c4d3") def test_mobile_data_usage_downlink_ipv6_vzw(self): """Verify mobile data usage over IPv6 and VZW carrier.""" self._test_data_usage_downlink(self.android_devices[1], self.android_devices[1], self.download_file_ipv6, cconst.TYPE_MOBILE) @test_tracker_info(uuid="fe1390e5-635c-49a9-b050-032e66f52f40") def test_wifi_tethering_mobile_data_usage_downlink_ipv4_tmo(self): """Verify mobile data usage with tethered device over IPv4 and TMO.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[1], self.download_file_ipv4, cconst.TYPE_MOBILE, True, False, False) @test_tracker_info(uuid="d7fde6b2-6672-484a-a2fd-47858f5a163f") def test_wifi_tethering_mobile_data_usage_downlink_ipv6_tmo(self): """Verify mobile data usage with tethered device over IPv6 and TMO.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[1], self.download_file_ipv6, cconst.TYPE_MOBILE, True, False, False) @test_tracker_info(uuid="6285ef69-37a8-4069-8cb2-21dc266a57c3") def test_wifi_tethering_mobile_data_usage_downlink_ipv4_vzw(self): """Verify mobile data usage with tethered device over IPv4 and VZW.""" self._test_data_usage_downlink(self.android_devices[1], self.android_devices[0], self.download_file_ipv4, cconst.TYPE_MOBILE, True, False, False) @test_tracker_info(uuid="565fc7e3-d7cf-43cc-8982-4f17999cf579") def test_wifi_tethering_mobile_data_usage_downlink_ipv6_vzw(self): """Verify mobile data usage with tethered device over IPv6 and VZW.""" self._test_data_usage_downlink(self.android_devices[1], self.android_devices[0], self.download_file_ipv6, cconst.TYPE_MOBILE, True, False, False) @test_tracker_info(uuid="72ddb42a-5942-4a6a-8b20-2181c41b2765") def test_wifi_data_usage_downlink_ipv4(self): """Verify wifi data usage over IPv4.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv4, cconst.TYPE_WIFI, False, True, True) @test_tracker_info(uuid="76b316f5-2946-4757-b5d6-62a8b1fd627e") def test_wifi_data_usage_downlink_ipv6(self): """Verify wifi data usage over IPv6.""" self._test_data_usage_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv6, cconst.TYPE_WIFI, False, True, True) @test_tracker_info(uuid="4be5f2d4-9bc6-4190-813e-044f00d94aa6") def test_wifi_tethering_wifi_data_usage_downlink_ipv4(self): """Verify wifi data usage with tethered device over IPv4.""" asserts.skip_if( self.android_devices[0].model not in self.dbs_supported_models, "Device %s does not support dual interfaces." % self.android_devices[0].model) self._test_data_usage_downlink(self.android_devices[0], self.android_devices[1], self.download_file_ipv4, cconst.TYPE_WIFI, True, True, False) @test_tracker_info(uuid="ac4750fd-20d9-451d-a85b-79fdbaa7da97") def test_data_usage_limit_downlink_ipv4_tmo(self): """Verify data limit for TMO with IPv4 link.""" self._test_data_usage_limit_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv4, cconst.TYPE_MOBILE) @test_tracker_info(uuid="6598732e-f4d3-40c7-a73b-e89bb3d196c5") def test_data_usage_limit_downlink_ipv6_tmo(self): """Verify data limit for TMO with IPv6 link.""" self._test_data_usage_limit_downlink(self.android_devices[0], self.android_devices[0], self.download_file_ipv6, cconst.TYPE_MOBILE) @test_tracker_info(uuid="d1ef0405-cf58-4141-94c7-cfa0c9d14438") def test_data_usage_limit_downlink_ipv4_vzw(self): """Verify data limit for VZW with IPv4 link.""" self._test_data_usage_limit_downlink(self.android_devices[1], self.android_devices[1], self.download_file_ipv4, cconst.TYPE_MOBILE) @test_tracker_info(uuid="18d169d9-fc4f-4782-8f5f-fc96b8976d03") def test_data_usage_limit_downlink_ipv6_vzw(self): """Verify data limit for VZW with IPv6 link.""" self._test_data_usage_limit_downlink(self.android_devices[1], self.android_devices[1], self.download_file_ipv6, cconst.TYPE_MOBILE) @test_tracker_info(uuid="7c9ab330-9645-4030-bb1e-dcce126944a2") def test_wifi_tethering_data_usage_limit_downlink_ipv4_tmo(self): """Verify data limit for TMO from tethered device with IPv4 link.""" self._test_data_usage_limit_downlink(self.android_devices[0], self.android_devices[1], self.download_file_ipv4, cconst.TYPE_MOBILE, True) @test_tracker_info(uuid="1cafdaf7-a41e-4d19-b08a-ae094d796426") def test_wifi_tethering_data_usage_limit_downlink_ipv6_tmo(self): """Verify data limit for TMO from tethered device with IPv6 link.""" self._test_data_usage_limit_downlink(self.android_devices[0], self.android_devices[1], self.download_file_ipv6, cconst.TYPE_MOBILE, True) @test_tracker_info(uuid="bd5a8f93-4e2f-474d-a01d-4bb5117d5350") def test_wifi_tethering_data_usage_limit_downlink_ipv4_vzw(self): """Verify data limit for VZW from tethered device with IPv4 link.""" self._test_data_usage_limit_downlink(self.android_devices[1], self.android_devices[0], self.download_file_ipv4, cconst.TYPE_MOBILE, True) @test_tracker_info(uuid="ec082fe5-6af1-425e-ace6-4726930bf62f") def test_wifi_tethering_data_usage_limit_downlink_ipv6_vzw(self): """Verify data limit for VZW from tethered device with IPv6 link.""" self._test_data_usage_limit_downlink(self.android_devices[1], self.android_devices[0], self.download_file_ipv6, cconst.TYPE_MOBILE, True)