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"""
17Script for exercising various ping scenarios
18
19"""
20from acts.base_test import BaseTestClass
21
22import os
23import threading
24import uuid
25
26from acts import signals
27from acts.controllers.ap_lib import hostapd_constants
28from acts.test_utils.abstract_devices.utils_lib.wlan_utils import setup_ap_and_associate
29from acts.test_utils.abstract_devices.wlan_device import create_wlan_device
30from acts.test_utils.tel.tel_test_utils import setup_droid_properties
31from acts.test_utils.fuchsia import utils
32from acts.utils import rand_ascii_str
33
34
35class PingStressTest(BaseTestClass):
36    # Timeout for ping thread in seconds
37    ping_thread_timeout_s = 60 * 5
38
39    # List to capture ping results
40    ping_threads_result = []
41
42    # IP addresses used in pings
43    google_dns_1 = '8.8.8.8'
44    google_dns_2 = '8.8.4.4'
45
46    def setup_class(self):
47        super().setup_class()
48
49        self.ssid = rand_ascii_str(10)
50        self.fd = self.fuchsia_devices[0]
51        self.wlan_device = create_wlan_device(self.fd)
52        self.ap = self.access_points[0]
53        setup_ap_and_associate(access_point=self.ap,
54                               client=self.wlan_device,
55                               profile_name='whirlwind',
56                               channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
57                               ssid=self.ssid)
58
59    def teardown_test(self):
60        self.wlan_device.disconnect()
61        self.wlan_device.reset_wifi()
62        self.ap.stop_all_aps()
63
64    def send_ping(self,
65                  dest_ip,
66                  count=3,
67                  interval=1000,
68                  timeout=1000,
69                  size=25):
70        ping_result = self.wlan_device.ping(dest_ip, count, interval, timeout,
71                                            size)
72        if ping_result:
73            self.log.info('Ping was successful.')
74        else:
75            if '8.8' in dest_ip:
76                raise signals.TestFailure('Ping was unsuccessful. Consider '
77                                          'possibility of server failure.')
78            else:
79                raise signals.TestFailure('Ping was unsuccessful.')
80        return True
81
82    def ping_thread(self, dest_ip):
83        ping_result = self.wlan_device.ping(dest_ip, count=10, size=50)
84        if ping_result:
85            self.log.info('Success pinging: %s' % dest_ip)
86        else:
87            self.log.info('Failure pinging: %s' % dest_ip)
88
89        self.ping_threads_result.append(ping_result)
90
91    def test_simple_ping(self):
92        return self.send_ping(self.google_dns_1)
93
94    def test_ping_local(self):
95        return self.send_ping('127.0.0.1')
96
97    def test_ping_AP(self):
98        return self.send_ping(self.ap.ssh_settings.hostname)
99
100    def test_ping_with_params(self):
101        return self.send_ping(self.google_dns_1,
102                              count=5,
103                              interval=800,
104                              size=50)
105
106    def test_long_ping(self):
107        return self.send_ping(self.google_dns_1, count=50)
108
109    def test_medium_packet_ping(self):
110        return self.send_ping(self.google_dns_1, size=64)
111
112    def test_medium_packet_long_ping(self):
113        return self.send_ping(self.google_dns_1,
114                              count=50,
115                              timeout=1500,
116                              size=64)
117
118    def test_large_packet_ping(self):
119        return self.send_ping(self.google_dns_1, size=500)
120
121    def test_large_packet_long_ping(self):
122        return self.send_ping(self.google_dns_1,
123                              count=50,
124                              timeout=5000,
125                              size=500)
126
127    def test_simultaneous_pings(self):
128        ping_urls = [
129            self.google_dns_1, self.google_dns_2, self.google_dns_1,
130            self.google_dns_2
131        ]
132        ping_threads = []
133
134        try:
135            # Start multiple ping at the same time
136            for index, url in enumerate(ping_urls):
137                self.log.info('Create and start thread %d.' % index)
138                t = threading.Thread(target=self.ping_thread, args=(url, ))
139                ping_threads.append(t)
140                t.start()
141
142            # Wait for all threads to complete or timeout
143            for t in ping_threads:
144                t.join(self.ping_thread_timeout_s)
145
146        finally:
147            is_alive = False
148
149            for index, t in enumerate(ping_threads):
150                if t.isAlive():
151                    t = None
152                    is_alive = True
153
154            if is_alive:
155                raise signals.TestFailure('Thread %d timedout' % index)
156
157        for index in range(0, len(self.ping_threads_result)):
158            if not self.ping_threads_result[index]:
159                self.log.info("Ping failed for %d" % index)
160                raise signals.TestFailure('Thread %d failed to ping. '
161                                          'Consider possibility of server '
162                                          'failure' % index)
163        return True
164