1# Copyright 2013 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# TODO(nduca): Rewrite what some of these tests to use mocks instead of 6# actually talking to the device. This would improve our coverage quite 7# a bit. 8 9import socket 10import tempfile 11import unittest 12 13from telemetry import benchmark 14from telemetry.core import forwarders 15from telemetry.core.platform import cros_interface 16from telemetry.core.forwarders import cros_forwarder 17from telemetry.unittest import options_for_unittests 18 19 20class CrOSInterfaceTest(unittest.TestCase): 21 @benchmark.Enabled('cros-chrome') 22 def testPushContents(self): 23 remote = options_for_unittests.GetCopy().cros_remote 24 with cros_interface.CrOSInterface( 25 remote, 26 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 27 cri.RunCmdOnDevice(['rm', '-rf', '/tmp/testPushContents']) 28 cri.PushContents('hello world', '/tmp/testPushContents') 29 contents = cri.GetFileContents('/tmp/testPushContents') 30 self.assertEquals(contents, 'hello world') 31 32 @benchmark.Enabled('cros-chrome') 33 def testExists(self): 34 remote = options_for_unittests.GetCopy().cros_remote 35 with cros_interface.CrOSInterface( 36 remote, 37 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 38 self.assertTrue(cri.FileExistsOnDevice('/proc/cpuinfo')) 39 self.assertTrue(cri.FileExistsOnDevice('/etc/passwd')) 40 self.assertFalse(cri.FileExistsOnDevice('/etc/sdlfsdjflskfjsflj')) 41 42 @benchmark.Enabled('linux') 43 def testExistsLocal(self): 44 with cros_interface.CrOSInterface() as cri: 45 self.assertTrue(cri.FileExistsOnDevice('/proc/cpuinfo')) 46 self.assertTrue(cri.FileExistsOnDevice('/etc/passwd')) 47 self.assertFalse(cri.FileExistsOnDevice('/etc/sdlfsdjflskfjsflj')) 48 49 @benchmark.Enabled('cros-chrome') 50 def testGetFileContents(self): # pylint: disable=R0201 51 remote = options_for_unittests.GetCopy().cros_remote 52 with cros_interface.CrOSInterface( 53 remote, 54 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 55 hosts = cri.GetFileContents('/etc/lsb-release') 56 self.assertTrue('CHROMEOS' in hosts) 57 58 @benchmark.Enabled('cros-chrome') 59 def testGetFileContentsNonExistent(self): 60 remote = options_for_unittests.GetCopy().cros_remote 61 with cros_interface.CrOSInterface( 62 remote, 63 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 64 f = tempfile.NamedTemporaryFile() 65 cri.PushContents('testGetFileNonExistent', f.name) 66 cri.RmRF(f.name) 67 self.assertRaises( 68 OSError, 69 lambda: cri.GetFileContents(f.name)) 70 71 @benchmark.Enabled('cros-chrome') 72 def testGetFile(self): # pylint: disable=R0201 73 remote = options_for_unittests.GetCopy().cros_remote 74 with cros_interface.CrOSInterface( 75 remote, 76 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 77 f = tempfile.NamedTemporaryFile() 78 cri.GetFile('/etc/lsb-release', f.name) 79 with open(f.name, 'r') as f2: 80 res = f2.read() 81 self.assertTrue('CHROMEOS' in res) 82 83 @benchmark.Enabled('cros-chrome') 84 def testGetFileNonExistent(self): 85 remote = options_for_unittests.GetCopy().cros_remote 86 with cros_interface.CrOSInterface( 87 remote, 88 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 89 f = tempfile.NamedTemporaryFile() 90 cri.PushContents('testGetFileNonExistent', f.name) 91 cri.RmRF(f.name) 92 self.assertRaises( 93 OSError, 94 lambda: cri.GetFile(f.name)) 95 96 @benchmark.Enabled('cros-chrome') 97 def testIsServiceRunning(self): 98 remote = options_for_unittests.GetCopy().cros_remote 99 with cros_interface.CrOSInterface( 100 remote, 101 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 102 self.assertTrue(cri.IsServiceRunning('openssh-server')) 103 104 @benchmark.Enabled('linux') 105 def testIsServiceRunningLocal(self): 106 with cros_interface.CrOSInterface() as cri: 107 self.assertTrue(cri.IsServiceRunning('dbus')) 108 109 @benchmark.Enabled('cros-chrome') 110 def testGetRemotePortAndIsHTTPServerRunningOnPort(self): 111 remote = options_for_unittests.GetCopy().cros_remote 112 with cros_interface.CrOSInterface( 113 remote, 114 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 115 116 # Create local server. 117 sock = socket.socket() 118 sock.bind(('', 0)) 119 port = sock.getsockname()[1] 120 sock.listen(0) 121 122 # Get remote port and ensure that it was unused. 123 remote_port = cri.GetRemotePort() 124 self.assertFalse(cri.IsHTTPServerRunningOnPort(remote_port)) 125 126 # Forward local server's port to remote device's remote_port. 127 forwarder = cros_forwarder.CrOsForwarderFactory(cri).Create( 128 forwarders.PortPairs(http=forwarders.PortPair(port, remote_port), 129 https=None, dns=None)) 130 131 # At this point, remote device should be able to connect to local server. 132 self.assertTrue(cri.IsHTTPServerRunningOnPort(remote_port)) 133 134 # Next remote port shouldn't be the same as remote_port, since remote_port 135 # is now in use. 136 self.assertTrue(cri.GetRemotePort() != remote_port) 137 138 139 # Close forwarder and local server ports. 140 forwarder.Close() 141 sock.close() 142 143 # Device should no longer be able to connect to remote_port since it is no 144 # longer in use. 145 self.assertFalse(cri.IsHTTPServerRunningOnPort(remote_port)) 146 147 @benchmark.Enabled('cros-chrome') 148 def testGetRemotePortReservedPorts(self): 149 remote = options_for_unittests.GetCopy().cros_remote 150 with cros_interface.CrOSInterface( 151 remote, 152 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 153 154 # Should return 2 separate ports even though the first one isn't 155 # technically being used yet. 156 remote_port_1 = cri.GetRemotePort() 157 remote_port_2 = cri.GetRemotePort() 158 159 self.assertTrue(remote_port_1 != remote_port_2) 160 161 # TODO(tengs): It would be best if we can filter this test and other tests 162 # that need to be run locally based on the platform of the system browser. 163 @benchmark.Enabled('linux') 164 def testEscapeCmdArguments(self): 165 ''' Commands and their arguments that are executed through the cros 166 interface should follow bash syntax. This test needs to run on remotely 167 and locally on the device to check for consistency. 168 ''' 169 with cros_interface.CrOSInterface( 170 options_for_unittests.GetCopy().cros_remote, 171 options_for_unittests.GetCopy().cros_ssh_identity) as cri: 172 173 # Check arguments with no special characters 174 stdout, _ = cri.RunCmdOnDevice(['echo', '--arg1=value1', '--arg2=value2', 175 '--arg3="value3"']) 176 assert(stdout.strip() == '--arg1=value1 --arg2=value2 --arg3=value3') 177 178 # Check argument with special characters escaped 179 stdout, _ = cri.RunCmdOnDevice(['echo', '--arg=A\\; echo \\"B\\"']) 180 assert(stdout.strip() == '--arg=A; echo "B"') 181 182 # Check argument with special characters in quotes 183 stdout, _ = cri.RunCmdOnDevice(['echo', "--arg='$HOME;;$PATH'"]) 184 assert(stdout.strip() == "--arg=$HOME;;$PATH") 185