| OLD | NEW |
|---|
| (Empty) |
| 1 # Copyright (c) 2012 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 import unittest | |
| 9 import socket | |
| 10 | |
| 11 from telemetry.core import util | |
| 12 from telemetry.core.chrome import cros_browser_backend | |
| 13 from telemetry.core.chrome import cros_interface | |
| 14 from telemetry.test import options_for_unittests | |
| 15 from telemetry.test import run_tests | |
| 16 | |
| 17 class CrOSInterfaceTest(unittest.TestCase): | |
| 18 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 19 def testDeviceSideProcessFailureToLaunch(self): | |
| 20 remote = options_for_unittests.GetCopy().cros_remote | |
| 21 cri = cros_interface.CrOSInterface( | |
| 22 remote, | |
| 23 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 24 | |
| 25 def WillFail(): | |
| 26 dsp = cros_interface.DeviceSideProcess( | |
| 27 cri, | |
| 28 ['sfsdfskjflwejfweoij']) | |
| 29 dsp.Close() | |
| 30 self.assertRaises(OSError, WillFail) | |
| 31 | |
| 32 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 33 def testDeviceSideProcessCloseDoesClose(self): | |
| 34 remote = options_for_unittests.GetCopy().cros_remote | |
| 35 cri = cros_interface.CrOSInterface( | |
| 36 remote, | |
| 37 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 38 | |
| 39 with cros_interface.DeviceSideProcess( | |
| 40 cri, | |
| 41 ['sleep', '111']) as dsp: | |
| 42 procs = cri.ListProcesses() | |
| 43 sleeps = [x for x in procs | |
| 44 if x[1] == 'sleep 111'] | |
| 45 assert dsp.IsAlive() | |
| 46 procs = cri.ListProcesses() | |
| 47 sleeps = [x for x in procs | |
| 48 if x[1] == 'sleep 111'] | |
| 49 self.assertEquals(len(sleeps), 0) | |
| 50 | |
| 51 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 52 def testPushContents(self): | |
| 53 remote = options_for_unittests.GetCopy().cros_remote | |
| 54 cri = cros_interface.CrOSInterface( | |
| 55 remote, | |
| 56 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 57 cri.GetCmdOutput(['rm', '-rf', '/tmp/testPushContents']) | |
| 58 cri.PushContents('hello world', '/tmp/testPushContents') | |
| 59 contents = cri.GetFileContents('/tmp/testPushContents') | |
| 60 self.assertEquals(contents, 'hello world') | |
| 61 | |
| 62 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 63 def testExists(self): | |
| 64 remote = options_for_unittests.GetCopy().cros_remote | |
| 65 cri = cros_interface.CrOSInterface( | |
| 66 remote, | |
| 67 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 68 self.assertTrue(cri.FileExistsOnDevice('/proc/cpuinfo')) | |
| 69 self.assertTrue(cri.FileExistsOnDevice('/etc/passwd')) | |
| 70 self.assertFalse(cri.FileExistsOnDevice('/etc/sdlfsdjflskfjsflj')) | |
| 71 | |
| 72 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 73 def testGetFileContents(self): # pylint: disable=R0201 | |
| 74 remote = options_for_unittests.GetCopy().cros_remote | |
| 75 cri = cros_interface.CrOSInterface( | |
| 76 remote, | |
| 77 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 78 hosts = cri.GetFileContents('/etc/hosts') | |
| 79 assert hosts.startswith('# /etc/hosts') | |
| 80 | |
| 81 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 82 def testGetFileContentsForSomethingThatDoesntExist(self): | |
| 83 remote = options_for_unittests.GetCopy().cros_remote | |
| 84 cri = cros_interface.CrOSInterface( | |
| 85 remote, | |
| 86 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 87 self.assertRaises( | |
| 88 OSError, | |
| 89 lambda: cri.GetFileContents('/tmp/209fuslfskjf/dfsfsf')) | |
| 90 | |
| 91 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 92 def testListProcesses(self): # pylint: disable=R0201 | |
| 93 remote = options_for_unittests.GetCopy().cros_remote | |
| 94 cri = cros_interface.CrOSInterface( | |
| 95 remote, | |
| 96 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 97 with cros_interface.DeviceSideProcess( | |
| 98 cri, | |
| 99 ['sleep', '11']): | |
| 100 procs = cri.ListProcesses() | |
| 101 sleeps = [x for x in procs | |
| 102 if x[1] == 'sleep 11'] | |
| 103 | |
| 104 assert len(sleeps) == 1 | |
| 105 | |
| 106 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 107 def testIsServiceRunning(self): | |
| 108 remote = options_for_unittests.GetCopy().cros_remote | |
| 109 cri = cros_interface.CrOSInterface( | |
| 110 remote, | |
| 111 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 112 | |
| 113 self.assertTrue(cri.IsServiceRunning('openssh-server')) | |
| 114 | |
| 115 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 116 def testGetRemotePortAndIsHTTPServerRunningOnPort(self): | |
| 117 remote = options_for_unittests.GetCopy().cros_remote | |
| 118 cri = cros_interface.CrOSInterface( | |
| 119 remote, | |
| 120 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 121 | |
| 122 # Create local server. | |
| 123 sock = socket.socket() | |
| 124 sock.bind(('', 0)) | |
| 125 port = sock.getsockname()[1] | |
| 126 sock.listen(0) | |
| 127 | |
| 128 # Get remote port and ensure that it was unused. | |
| 129 remote_port = cri.GetRemotePort() | |
| 130 self.assertFalse(cri.IsHTTPServerRunningOnPort(remote_port)) | |
| 131 | |
| 132 # Forward local server's port to remote device's remote_port. | |
| 133 forwarder = cros_browser_backend.SSHForwarder( | |
| 134 cri, 'R', util.PortPair(port, remote_port)) | |
| 135 | |
| 136 # At this point, remote device should be able to connect to local server. | |
| 137 self.assertTrue(cri.IsHTTPServerRunningOnPort(remote_port)) | |
| 138 | |
| 139 # Next remote port shouldn't be the same as remote_port, since remote_port | |
| 140 # is now in use. | |
| 141 self.assertTrue(cri.GetRemotePort() != remote_port) | |
| 142 | |
| 143 # Close forwarder and local server ports. | |
| 144 forwarder.Close() | |
| 145 sock.close() | |
| 146 | |
| 147 # Device should no longer be able to connect to remote_port since it is no | |
| 148 # longer in use. | |
| 149 self.assertFalse(cri.IsHTTPServerRunningOnPort(remote_port)) | |
| 150 | |
| 151 @run_tests.RequiresBrowserOfType('cros-chrome') | |
| 152 def testGetRemotePortReservedPorts(self): | |
| 153 remote = options_for_unittests.GetCopy().cros_remote | |
| 154 cri = cros_interface.CrOSInterface( | |
| 155 remote, | |
| 156 options_for_unittests.GetCopy().cros_ssh_identity) | |
| 157 | |
| 158 # Should return 2 separate ports even though the first one isn't technically | |
| 159 # being used yet. | |
| 160 remote_port_1 = cri.GetRemotePort() | |
| 161 remote_port_2 = cri.GetRemotePort() | |
| 162 | |
| 163 self.assertTrue(remote_port_1 != remote_port_2) | |
| OLD | NEW |
|---|
Chromium Code Reviews
