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 |