slave/tests/run_slave_test.py

#!/usr/bin/env vpython
# Copyright 2016 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Starts run_slave and verify if it starts successfully.
"""

import os
import re
import runpy
import subprocess
import sys
import unittest

import mock

BASE_DIR = os.path.abspath(
  os.path.join(os.path.dirname(__file__), '..', '..'))

sys.path.insert(0, os.path.join(BASE_DIR, 'scripts'))
import common.env
common.env.Install(with_third_party=True)

RUN_SLAVE_PATH = os.path.join(BASE_DIR, 'slave')
sys.path.insert(0, RUN_SLAVE_PATH)

import run_slave

# import twisted to mock run()
import twisted.scripts.twistd


class ExitInvoked(Exception):
  # Raises when a mocked sys.exit() is invoked.
  # With mocking sys.exit() with this exception, the execution flow doesn't
  # continue after sys.exit(), but interrupted and the control is passed back to
  # a unit test.
  pass


def _GetCallArgumentFromMock(mock_call, position, keyword=None):
  args, kwargs = mock_call.call_args
  return kwargs[keyword] if keyword and keyword in kwargs else args[position]


class RunSlaveTest(unittest.TestCase):
  def setUp(self):
    super(RunSlaveTest, self).setUp()
    os.environ['BUILDBOT_TEST_PASSWORD'] = 'hot bananas'

  def tearDown(self):
    super(RunSlaveTest, self).tearDown()
    os.environ.pop('BUILDBOT_TEST_PASSWORD', None)

  @mock.patch('subprocess.call')
  @mock.patch('subprocess.Popen')
  @mock.patch('sys.exit', mock.Mock(side_effect=ExitInvoked))
  def test_run_slave_restart_after_gclient_sync(self, popen_call,
                                                subprocess_call):
    """Tests if run_slave restarts itself after gclient sync."""
    with self.assertRaises(ExitInvoked):
      runpy.run_module("run_slave", run_name="__main__", alter_sys=True)

    # verify that gclient sync has been executed
    self.assertTrue(subprocess_call.called)
    call_cmd_args = _GetCallArgumentFromMock(subprocess_call, 0, 'args')
    self.assertEqual(call_cmd_args[0], run_slave.GetGClientPath())
    self.assertEqual(call_cmd_args[1], 'sync')

    # verify that run_slave was re-executed by Popen() with --no-gclient-sync
    run_slave_py_path = re.sub(
        r'pyc$', 'py', os.path.abspath(run_slave.__file__))

    popen_cmd_args = _GetCallArgumentFromMock(popen_call, 0, 'args')

    # run_slave.py should be placed either at the beinning of the args
    # after the path python executable
    # : i.e., python run_slave.py --foo, or run_slave.py --foo
    self.assertIn(run_slave_py_path, popen_cmd_args[:2])
    self.assertIn('--no-gclient-sync', popen_cmd_args[1:])

  @mock.patch('subprocess.call')
  @mock.patch('subprocess.check_call')
  @mock.patch('subprocess.check_output')
  @mock.patch('twisted.scripts.twistd.run')
  @mock.patch('sys.argv', [run_slave.__file__, '--no-gclient-sync'])
  def test_run_slave_with_no_gclient_sync(self, twistd_run,
                                          subprocess_check_output,
                                          subprocess_check_call,
                                          subprocess_call):
    """Tests if twistd.run() gets invoked when --no-gclient-sync is given."""
    os.environ['TESTING_MASTER'] = 'Master1'
    os.environ['TESTING_SLAVE'] = 'Slave1'
    runpy.run_module("run_slave", run_name="__main__", alter_sys=True)

    gclient_path = run_slave.GetGClientPath()
    for mock_call in (subprocess_call, subprocess_check_call,
                      subprocess_check_output):
      if mock_call.called:
        cmd_args = _GetCallArgumentFromMock(mock_call, 0, 'args')

        # verify that gclient with sync option has not been called within
        # run_slave.
        if gclient_path == cmd_args[0]:
          self.assertNotEqual(cmd_args[1], 'sync')

    self.assertTrue(twistd_run.called)


if __name__ == '__main__':
  unittest.main()