• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#
2#   Copyright 2018 - The Android Open Source Project
3#
4#   Licensed under the Apache License, Version 2.0 (the "License");
5#   you may not use this file except in compliance with the License.
6#   You may obtain a copy of the License at
7#
8#       http://www.apache.org/licenses/LICENSE-2.0
9#
10#   Unless required by applicable law or agreed to in writing, software
11#   distributed under the License is distributed on an "AS IS" BASIS,
12#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13#   See the License for the specific language governing permissions and
14#   limitations under the License.
15
16from acts import asserts
17from acts import base_test
18from acts import utils
19from acts.controllers import adb
20from acts.test_decorators import test_tracker_info
21from acts.test_utils.net.net_test_utils import start_tcpdump
22from acts.test_utils.net.net_test_utils import stop_tcpdump
23from acts.test_utils.net import connectivity_test_utils as cutils
24from acts.test_utils.wifi import wifi_test_utils as wutils
25
26import random
27import time
28
29WLAN = "wlan0"
30PKTS = 5
31SERVER_UDP_KEEPALIVE = "python /root/udp_nat_keepalive.py"
32KEEPALIVE_DATA = "ff"
33
34
35class NattKeepAliveTest(base_test.BaseTestClass):
36    """ Tests for NATT keepalive """
37
38    def setup_class(self):
39        """ Setup devices for tests and unpack params """
40
41        self.dut = self.android_devices[0]
42        req_params = ("wifi_network", "remote_server", "server_ssh_config")
43        self.unpack_userparams(req_params)
44
45        wutils.wifi_connect(self.dut, self.wifi_network)
46
47        self.ip_a = self.dut.droid.connectivityGetIPv4Addresses(WLAN)[0]
48        self.ip_b = self.remote_server
49        self.log.info("DUT IP addr: %s" % self.ip_a)
50        self.log.info("Remote server IP addr: %s" % self.ip_b)
51        self.tcpdump_pid = None
52
53    def teardown_class(self):
54        wutils.reset_wifi(self.dut)
55
56    def setup_test(self):
57        self.tcpdump_pid = start_tcpdump(self.dut, self.test_name)
58
59    def teardown_test(self):
60        stop_tcpdump(self.dut, self.tcpdump_pid, self.test_name)
61
62    def on_fail(self, test_name, begin_time):
63        self.dut.take_bug_report(test_name, begin_time)
64
65    """ Helper functions """
66
67    def _verify_time_interval(self, time_interval, cmd_out):
68        """ Verify time diff between packets is equal to time interval """
69
70        self.log.info("Packet Info: \n%s\n" % cmd_out)
71        pkts = cmd_out.rstrip().decode("utf-8", "ignore").split("\n")
72
73        prev = 0
74        for i in range(len(pkts)):
75            interval, data = pkts[i][1:-1].split(',')
76
77            # verify data
78            if data.lstrip().rstrip()[1:-1] != KEEPALIVE_DATA:
79                self.log.error("Server received invalid data %s" % data)
80                return False
81
82            # verify time interval
83            curr = float(interval)
84            if i == 0:
85                prev = curr
86                continue
87            diff = int(round(curr-prev))
88            if diff != time_interval:
89                self.log.error("Keepalive time interval is %s, expected %s"
90                               % (diff, time_interval))
91                return False
92            prev = curr
93
94        return True
95
96    """ Tests begin """
97
98    @test_tracker_info(uuid="c9012da2-656f-44ef-bad6-26892335d4bd")
99    def test_natt_keepalive_ipv4(self):
100        """ Test natt keepalive over wifi
101
102        Steps:
103          1. Open a UDP port 4500 on linux host
104          2. Start NATT keepalive packets on DUT and send 5 packets
105          3. Verify that 5 keepalive packets reached host with data '0xff'
106        """
107
108        # set a time interval
109        result = True
110        time_interval = random.randint(10, 60)
111        port = random.randint(8000, 9000)
112        self.log.info("NATT keepalive time interval is %s" % time_interval)
113        self.log.info("Source port is %s" % port)
114        time_out = time_interval * PKTS + 6
115
116        # start NATT keep alive
117        nka_key = cutils.start_natt_keepalive(
118            self.dut, self.ip_a, port, self.ip_b, time_interval)
119        asserts.assert_true(nka_key, "Failed to start NATT keepalive")
120
121        # capture packets on server
122        self.log.info("Capturing keepalive packets on %s" % self.ip_b)
123        cmd_out = utils.exe_cmd("timeout %s %s" %
124                                (time_out, SERVER_UDP_KEEPALIVE))
125
126        # verify packets received
127        result = self._verify_time_interval(time_interval, cmd_out)
128
129        # stop NATT keep alive
130        status = cutils.stop_natt_keepalive(self.dut, nka_key)
131        asserts.assert_true(status, "Failed to stop NATT keepalive")
132
133        return result
134
135    @test_tracker_info(uuid="8ab20733-4a9e-4e4d-a46f-4d32a9f221c5")
136    def test_natt_keepalive_ipv4_invalid_interval(self):
137        """ Test invalid natt keepalive time interval
138
139        Steps:
140          1. Start NATT keepalive with time interval less than 10 seconds
141          2. API should return invalid interval
142        """
143
144        # start NATT keep alive
145        port = random.randint(8000, 9000)
146        nka_key = cutils.start_natt_keepalive(
147            self.dut, self.ip_a, port, self.ip_b, 2)
148        asserts.assert_true(not nka_key,
149                            "Started NATT keepalive with invalid interval")
150
151    """ Tests end """
152