src/tools/telemetry/telemetry/core/backends/chrome/cros_interface_unittest.py

   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
   9 import socket
  10 import tempfile
  11 import unittest
  12
  13 from telemetry import benchmark
  14 from telemetry.core import forwarders
  15 from telemetry.core.backends.chrome import cros_interface
  16 from telemetry.core.forwarders import cros_forwarder
  17 from telemetry.unittest import options_for_unittests
  18
  19
  20 class 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")