1#!/usr/bin/env python3
2#
3# Copyright (C) 2018 The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License"); you may not
6# use this file except in compliance with the License. You may obtain a copy of
7# 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, WITHOUT
13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14# License for the specific language governing permissions and limitations under
15# the License.
16"""
17PingStressTest exercises sending ICMP and ICMPv6 pings to a wireless access
18router and another device behind the AP. Note, this does not reach out to the
19internet. The DUT is only responsible for sending a routable packet; any
20communication past the first-hop is not the responsibility of the DUT.
21"""
22
23import threading
24
25from collections import namedtuple
26
27from acts import signals
28from acts import utils
29
30from acts.controllers.access_point import setup_ap
31from acts.controllers.ap_lib import hostapd_constants
32from acts_contrib.test_utils.wifi.WifiBaseTest import WifiBaseTest
33from acts_contrib.test_utils.abstract_devices.wlan_device import create_wlan_device
34from acts.utils import rand_ascii_str
35
36LOOPBACK_IPV4 = '127.0.0.1'
37LOOPBACK_IPV6 = '::1'
38PING_RESULT_TIMEOUT_SEC = 60 * 5
39
40Test = namedtuple(
41    typename='Args',
42    field_names=['name', 'dest_ip', 'count', 'interval', 'timeout', 'size'],
43    defaults=[3, 1000, 1000, 25])
44
45Addrs = namedtuple(
46    typename='Addrs',
47    field_names=['gateway_ipv4', 'gateway_ipv6', 'remote_ipv4', 'remote_ipv6'])
48
49
50class PingStressTest(WifiBaseTest):
51
52    def setup_generated_tests(self):
53        self.generate_tests(
54            self.send_ping, lambda test_name, *_: f'test_{test_name}', [
55                Test("loopback_ipv4", LOOPBACK_IPV4),
56                Test("loopback_ipv6", LOOPBACK_IPV6),
57                Test("gateway_ipv4", lambda addrs: addrs.gateway_ipv4),
58                Test("gateway_ipv6", lambda addrs: addrs.gateway_ipv6),
59                Test("remote_ipv4_small_packet",
60                     lambda addrs: addrs.remote_ipv4),
61                Test("remote_ipv6_small_packet",
62                     lambda addrs: addrs.remote_ipv6),
63                Test("remote_ipv4_small_packet_long",
64                     lambda addrs: addrs.remote_ipv4,
65                     count=50),
66                Test("remote_ipv6_small_packet_long",
67                     lambda addrs: addrs.remote_ipv6,
68                     count=50),
69                Test("remote_ipv4_medium_packet",
70                     lambda addrs: addrs.remote_ipv4,
71                     size=64),
72                Test("remote_ipv6_medium_packet",
73                     lambda addrs: addrs.remote_ipv6,
74                     size=64),
75                Test("remote_ipv4_medium_packet_long",
76                     lambda addrs: addrs.remote_ipv4,
77                     count=50,
78                     timeout=1500,
79                     size=64),
80                Test("remote_ipv6_medium_packet_long",
81                     lambda addrs: addrs.remote_ipv6,
82                     count=50,
83                     timeout=1500,
84                     size=64),
85                Test("remote_ipv4_large_packet",
86                     lambda addrs: addrs.remote_ipv4,
87                     size=500),
88                Test("remote_ipv6_large_packet",
89                     lambda addrs: addrs.remote_ipv6,
90                     size=500),
91                Test("remote_ipv4_large_packet_long",
92                     lambda addrs: addrs.remote_ipv4,
93                     count=50,
94                     timeout=5000,
95                     size=500),
96                Test("remote_ipv6_large_packet_long",
97                     lambda addrs: addrs.remote_ipv6,
98                     count=50,
99                     timeout=5000,
100                     size=500),
101            ])
102
103    def setup_class(self):
104        super().setup_class()
105        self.ssid = rand_ascii_str(10)
106        self.dut = create_wlan_device(self.fuchsia_devices[0])
107        self.access_point = self.access_points[0]
108        self.iperf_server = self.iperf_servers[0]
109        setup_ap(access_point=self.access_point,
110                 profile_name='whirlwind',
111                 channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
112                 ssid=self.ssid,
113                 setup_bridge=True,
114                 is_ipv6_enabled=True,
115                 is_nat_enabled=False)
116
117        ap_bridges = self.access_point.interfaces.get_bridge_interface()
118        if len(ap_bridges) != 1:
119            raise signals.TestAbortClass(
120                f'Expected one bridge interface on the AP, got {ap_bridges}')
121        self.ap_ipv4 = utils.get_addr(self.access_point.ssh, ap_bridges[0])
122        self.ap_ipv6 = utils.get_addr(self.access_point.ssh,
123                                      ap_bridges[0],
124                                      addr_type='ipv6_link_local')
125        self.log.info(
126            f"Gateway finished setup ({self.ap_ipv4} | {self.ap_ipv6})")
127
128        self.iperf_server.renew_test_interface_ip_address()
129        self.iperf_server_ipv4 = self.iperf_server.get_addr()
130        self.iperf_server_ipv6 = self.iperf_server.get_addr(
131            addr_type='ipv6_private_local')
132        self.log.info(
133            f"Remote finished setup ({self.iperf_server_ipv4} | {self.iperf_server_ipv6})"
134        )
135
136        self.dut.associate(self.ssid)
137
138        # Wait till the DUT has valid IP addresses after connecting.
139        self.dut.device.wait_for_ipv4_addr(
140            self.dut.device.wlan_client_test_interface_name)
141        self.dut.device.wait_for_ipv6_addr(
142            self.dut.device.wlan_client_test_interface_name)
143        self.log.info("DUT has valid IP addresses on test network")
144
145    def teardown_class(self):
146        self.dut.disconnect()
147        self.dut.reset_wifi()
148        self.download_ap_logs()
149        self.access_point.stop_all_aps()
150
151    def send_ping(self,
152                  _,
153                  get_addr_fn,
154                  count=3,
155                  interval=1000,
156                  timeout=1000,
157                  size=25):
158        dest_ip = get_addr_fn(
159            Addrs(
160                gateway_ipv4=self.ap_ipv4,
161                # IPv6 link-local addresses require specification of the
162                # outgoing interface as the scope ID when sending packets.
163                gateway_ipv6=
164                f'{self.ap_ipv6}%{self.dut.get_default_wlan_test_interface()}',
165                remote_ipv4=self.iperf_server_ipv4,
166                # IPv6 global addresses do not require scope IDs.
167                remote_ipv6=self.iperf_server_ipv6)) if callable(
168                    get_addr_fn) else get_addr_fn
169
170        self.log.info(f'Attempting to ping {dest_ip}...')
171        ping_result = self.dut.can_ping(dest_ip, count, interval, timeout,
172                                        size)
173        if ping_result:
174            self.log.info('Ping was successful.')
175        else:
176            raise signals.TestFailure('Ping was unsuccessful.')
177
178    def test_simultaneous_pings(self):
179        ping_urls = [
180            self.iperf_server_ipv4,
181            self.ap_ipv4,
182            self.iperf_server_ipv6,
183            f'{self.ap_ipv6}%{self.dut.get_default_wlan_test_interface()}',
184        ]
185        ping_threads = []
186        ping_results = []
187
188        def ping_thread(self, dest_ip, ping_results):
189            self.log.info('Attempting to ping %s...' % dest_ip)
190            ping_result = self.dut.can_ping(dest_ip, count=10, size=50)
191            if ping_result:
192                self.log.info('Success pinging: %s' % dest_ip)
193            else:
194                self.log.info('Failure pinging: %s' % dest_ip)
195            ping_results.append(ping_result)
196
197        try:
198            # Start multiple ping at the same time
199            for index, url in enumerate(ping_urls):
200                t = threading.Thread(target=ping_thread,
201                                     args=(self, url, ping_results))
202                ping_threads.append(t)
203                t.start()
204
205            # Wait for all threads to complete or timeout
206            for t in ping_threads:
207                t.join(PING_RESULT_TIMEOUT_SEC)
208
209        finally:
210            is_alive = False
211
212            for index, t in enumerate(ping_threads):
213                if t.is_alive():
214                    t = None
215                    is_alive = True
216
217            if is_alive:
218                raise signals.TestFailure(
219                    f'Timed out while pinging {ping_urls[index]}')
220
221        for index in range(0, len(ping_results)):
222            if not ping_results[index]:
223                raise signals.TestFailure(f'Failed to ping {ping_urls[index]}')
224        return True
225