1#!/usr/bin/python3.4
2#
3#   Copyright 2017 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17import random
18import time
19
20from acts import asserts
21from acts.test_decorators import test_tracker_info
22from acts.test_utils.wifi.rtt import rtt_const as rconsts
23from acts.test_utils.wifi.rtt import rtt_test_utils as rutils
24from acts.test_utils.wifi.rtt.RttBaseTest import RttBaseTest
25
26
27class RttRequestManagementTest(RttBaseTest):
28  """Test class for RTT request management flows."""
29
30  SPAMMING_LIMIT = 20
31
32  def __init__(self, controllers):
33    RttBaseTest.__init__(self, controllers)
34
35  #############################################################################
36
37  @test_tracker_info(uuid="29ff4a02-2952-47df-bf56-64f30c963093")
38  def test_cancel_ranging(self):
39    """Request a 'large' number of range operations with various UIDs (using the
40    work-source API), then cancel some of them.
41
42    We can't guarantee a reaction time - it is possible that a cancelled test
43    was already finished and it's results dispatched back. The test therefore
44    stacks the request queue. The sequence is:
45
46    - Request:
47      - 50 tests @ UIDs = {uid1, uid2, uid3}
48      - 2 tests @ UIDs = {uid2, uid3}
49      - 1 test2 @ UIDs = {uid1, uid2, uid3}
50    - Cancel UIDs = {uid2, uid3}
51
52    Expect to receive only 51 results.
53    """
54    dut = self.android_devices[0]
55    max_peers = dut.droid.wifiRttMaxPeersInRequest()
56
57    all_uids = [1000, 20, 30] # 1000 = System Server (makes requests foreground)
58    some_uids = [20, 30]
59
60    aps = rutils.select_best_scan_results(
61      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
62      select_count=1)
63    dut.log.info("RTT Supporting APs=%s", aps)
64
65    asserts.assert_true(
66        len(aps) > 0,
67        "Need at least one AP which supports 802.11mc!")
68    if len(aps) > max_peers:
69      aps = aps[0:max_peers]
70
71    group1_ids = []
72    group2_ids = []
73    group3_ids = []
74
75    # step 1: request <spam_limit> ranging operations on [uid1, uid2, uid3]
76    for i in range(self.SPAMMING_LIMIT):
77      group1_ids.append(
78        dut.droid.wifiRttStartRangingToAccessPoints(aps, all_uids))
79
80    # step 2: request 2 ranging operations on [uid2, uid3]
81    for i in range(2):
82      group2_ids.append(
83        dut.droid.wifiRttStartRangingToAccessPoints(aps, some_uids))
84
85    # step 3: request 1 ranging operation on [uid1, uid2, uid3]
86    for i in range(1):
87      group3_ids.append(
88          dut.droid.wifiRttStartRangingToAccessPoints(aps, all_uids))
89
90    # step 4: cancel ranging requests on [uid2, uid3]
91    dut.droid.wifiRttCancelRanging(some_uids)
92
93    # collect results
94    for i in range(len(group1_ids)):
95      rutils.wait_for_event(dut, rutils.decorate_event(
96        rconsts.EVENT_CB_RANGING_ON_RESULT, group1_ids[i]))
97    time.sleep(rutils.EVENT_TIMEOUT) # optimize time-outs below to single one
98    for i in range(len(group2_ids)):
99      rutils.fail_on_event(dut, rutils.decorate_event(
100          rconsts.EVENT_CB_RANGING_ON_RESULT, group2_ids[i]), 0)
101    for i in range(len(group3_ids)):
102      rutils.wait_for_event(dut, rutils.decorate_event(
103          rconsts.EVENT_CB_RANGING_ON_RESULT, group3_ids[i]))
104
105  @test_tracker_info(uuid="48297480-c026-4780-8c13-476e7bea440c")
106  def test_throttling(self):
107    """Request sequential range operations using a bogus UID (which will
108    translate as a throttled process) and similarly using the ACTS/sl4a as
109    the source (a foreground/unthrottled process)."""
110    dut = self.android_devices[0]
111    max_peers = dut.droid.wifiRttMaxPeersInRequest()
112
113    # Need to use a random number since the system keeps states and so the
114    # background uid will be throttled on the next run of this script
115    fake_uid = [random.randint(10, 9999)]
116
117    aps = rutils.select_best_scan_results(
118      rutils.scan_with_rtt_support_constraint(dut, True, repeat=10),
119      select_count=1)
120    dut.log.info("RTT Supporting APs=%s", aps)
121
122    asserts.assert_true(
123        len(aps) > 0,
124        "Need at least one AP which supports 802.11mc!")
125    if len(aps) > max_peers:
126      aps = aps[0:max_peers]
127
128    id1 = dut.droid.wifiRttStartRangingToAccessPoints(aps) # as ACTS/sl4a
129    id2 = dut.droid.wifiRttStartRangingToAccessPoints(aps, fake_uid)
130    id3 = dut.droid.wifiRttStartRangingToAccessPoints(aps, fake_uid)
131    id4 = dut.droid.wifiRttStartRangingToAccessPoints(aps) # as ACTS/sl4a
132
133    rutils.wait_for_event(dut, rutils.decorate_event(
134      rconsts.EVENT_CB_RANGING_ON_RESULT, id1))
135    rutils.wait_for_event(dut, rutils.decorate_event(
136        rconsts.EVENT_CB_RANGING_ON_RESULT, id2))
137    rutils.wait_for_event(dut, rutils.decorate_event(
138        rconsts.EVENT_CB_RANGING_ON_FAIL, id3))
139    rutils.wait_for_event(dut, rutils.decorate_event(
140        rconsts.EVENT_CB_RANGING_ON_RESULT, id4))
141