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