1# Copyright 2015 The Chromium Authors. All rights reserved.
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5import os
6import time
7
8from autotest_lib.client.bin import test, utils
9from autotest_lib.client.common_lib import error
10from autotest_lib.client.common_lib.cros import chrome
11from autotest_lib.client.common_lib.cros.network import interface
12
13_CLIENT_COMPLETE_FLAG = '/tmp/network_FirewallHolePunch'
14
15class network_FirewallHolePunch(test.test):
16    """Tests that controls an app that can open ports."""
17    version = 1
18    preserve_srcdir = True
19
20
21    def setup(self):
22        """Sets the current directory so the app can be accessed."""
23        os.chdir(self.srcdir)
24        self.extension_path = os.path.join(os.path.dirname(__file__),
25                                           'src/tcpserver')
26
27
28    def wait_for_server_command(self):
29        """Waits for the server to send a command.
30
31        @returns True if the server responds to the request; False otherwise.
32
33        """
34        for i in range(30):
35            result = utils.run('ls %s' %  _CLIENT_COMPLETE_FLAG,
36                               ignore_status=True)
37            if result.exit_status != 0:
38                return True
39            time.sleep(1)
40        return False
41
42
43    def interpret_command(self, command, test_error):
44        """Takes the string command and performs the appropriate action.
45
46        @param command: the command string
47        @param test_error: string of the test error message
48
49
50        @raises TestError if the server does not set the flag or if an
51                invalid command is passed.
52        """
53        if self.wait_for_server_command() is False:
54            raise error.TestError(test_error)
55
56        if command == 'launch app':
57            self.extension = self.cr.get_extension(self.extension_path)
58            self.extension.ExecuteJavaScript('tcpUI.create();')
59        elif command == 'start server':
60            script = str('tcpUI.startServer("%s", %d);' %
61                         (self.ip_address, self.port))
62            self.extension.ExecuteJavaScript(script)
63        elif command == 'stop server':
64            self.extension.ExecuteJavaScript('tcpUI.stopServer();')
65        elif command == 'exit app':
66            script = 'commandWindow.contentWindow.close();'
67            self.extension.ExecuteJavaScript(script)
68            self.extension.ExecuteJavaScript('close();')
69        elif command == 'logout':
70            self.cr.browser.Close()
71        elif command == 'login':
72            flag = '--enable-firewall-hole-punching'
73            self.cr = chrome.Chrome(extension_paths=[self.extension_path],
74                                    extra_browser_args=flag)
75        else:
76            raise error.TestError('Invalid client command passed.')
77
78        utils.run('touch %s' % _CLIENT_COMPLETE_FLAG)
79
80
81    def run_once(self, test_sequence=None, port=8888):
82        """Runs the integration test."""
83
84        # Throw if no test sequence
85        if not test_sequence:
86            raise error.TestError('No test sequence was passed to client.')
87
88        # Get the IP Address of the DUT
89        ethernet = interface.Interface.get_connected_ethernet_interface()
90        self.ip_address = ethernet.ipv4_address
91        self.port = port
92
93        self.cr = None
94        self.extension = None
95
96        for command in test_sequence:
97            self.interpret_command(command['client_command'],
98                                   command['client_error'])
99
100