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 5 import os 6 import time 7 8 from autotest_lib.client.bin import test, utils 9 from autotest_lib.client.common_lib import error 10 from autotest_lib.client.common_lib.cros import chrome 11 from autotest_lib.client.common_lib.cros.network import interface 12 13 _CLIENT_COMPLETE_FLAG = '/tmp/network_FirewallHolePunch' 14 15 class 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
