compute_page_test.py

# Copyright 2015 Google Inc. All Rights Reserved.

import unittest
import webtest

import apprtc
import compute_page
from test_util import CapturingFunction
from test_util import ReplaceFunction

from google.appengine.ext import testbed


class FakeComputeService(object):
  """Handles Compute Engine service calls to the Google API client."""

  def __init__(self):
    # This is a bit hacky but the Google API is lots of function
    # chaining. This at least makes things easier to reason about as
    # opposed to making a new variable for each level.
    self.instances = CapturingFunction(lambda: self.instances)

    self.instances.start = CapturingFunction(lambda: self.instances.start)
    self.instances.start.execute = CapturingFunction()

    self.instances.stop = CapturingFunction(lambda: self.instances.stop)
    self.instances.stop.execute = CapturingFunction()

    # Change the status of an instance by setting the return value.
    self.get_return_value = None
    self.instances.get = CapturingFunction(lambda: self.instances.get)
    self.instances.get.execute = CapturingFunction(
        lambda: self.get_return_value)


class ComputePageHandlerTest(unittest.TestCase):
  def setUp(self):
    # First, create an instance of the Testbed class.
    self.testbed = testbed.Testbed()

    # Then activate the testbed, which prepares the service stubs for use.
    self.testbed.activate()

    # Next, declare which service stubs you want to use.
    self.testbed.init_taskqueue_stub()

    self.test_app = webtest.TestApp(apprtc.app)

    # Fake out the compute service.
    self.build_compute_service_replacement = ReplaceFunction(
        compute_page.ComputePage,
        '_build_compute_service',
        self.fake_build_compute_service)

    # Note that '_build_compute_service' is called at request
    # time. Tests need to be able to setup test data in the fake
    # before ComputePage is initialize. Thus compute_service is
    # created here and it is assumed that _build_compute_service is
    # only called once.
    self.compute_service = FakeComputeService()

    # Default testbed app name.
    self.app_id = 'testbed-test'

  def fake_build_compute_service(self, *args):
    return self.compute_service

  def tearDown(self):
    self.testbed.deactivate()
    del self.build_compute_service_replacement

  # TODO(decurtis): Extract these functions to our own test base class.
  def makePostRequest(self, path, body=''):
    return self.test_app.post(path, body, headers={'User-Agent': 'Safari'})

  def makeGetRequest(self, path):
    # PhantomJS uses WebKit, so Safari is closest to the thruth.
    return self.test_app.get(path, headers={'User-Agent': 'Safari'})

  def testGetStatus(self):
    instance = 'test-instance'
    zone = 'test-zone'

    # Fake instance has status of RUNNING
    instance_status = compute_page.COMPUTE_STATUS_RUNNING
    self.compute_service.get_return_value = {
        compute_page.COMPUTE_STATUS: instance_status
    }

    response = self.makeGetRequest('/compute/%s/%s/%s' %
                                   (compute_page.ACTION_STATUS, instance, zone))

    get_dict = {'project': self.app_id,
                'instance': instance,
                'zone': zone}

    actual_get_dict = self.compute_service.instances.get.last_kwargs
    self.assertEqual(get_dict, actual_get_dict)
    self.assertEqual(instance_status, response.body)

  def testPostStartWhenTerminated(self):
    """Test start action with instance in the TERMINATED state."""
    instance = 'test-instance'
    zone = 'test-zone'
    start_url = '/compute/%s/%s/%s' % (compute_page.ACTION_START,
                                       instance, zone)

    # Suppose that instance is TERMINATED.
    instance_status = compute_page.COMPUTE_STATUS_TERMINATED
    self.compute_service.get_return_value = {
        compute_page.COMPUTE_STATUS: instance_status
    }

    # makePostRequest will check for 200 success.
    self.makePostRequest(start_url)

    # If the state is TERMINATED then we should have a new start task.
    tasks = self.testbed.get_stub('taskqueue').GetTasks('default')
    self.assertEqual(1, len(tasks))

    # Verify start() API called only once.
    self.assertEqual(1, self.compute_service.instances.start.num_calls)

    # Check start() API called when in the TERMINATED state.
    get_dict = {'project': self.app_id,
                'instance': instance,
                'zone': zone}
    actual_get_dict = self.compute_service.instances.start.last_kwargs
    self.assertEqual(get_dict, actual_get_dict)

  def testPostStartWhenRunning(self):
    """Test start action when instance in the RUNNING state."""
    instance = 'test-instance'
    zone = 'test-zone'
    start_url = '/compute/%s/%s/%s' % (compute_page.ACTION_START,
                                       instance, zone)

    instance_status = compute_page.COMPUTE_STATUS_RUNNING
    self.compute_service.get_return_value = {
        compute_page.COMPUTE_STATUS: instance_status
    }

    self.makePostRequest(start_url)

    # No further tasks needed.
    tasks = self.testbed.get_stub('taskqueue').GetTasks('default')
    self.assertEqual(0, len(tasks))

    # Verify start() API not called.
    self.assertEqual(0, self.compute_service.instances.start.num_calls)

  def testPostStartNotRunningOrTerminated(self):
    """Test start action in an intermediate state."""
    compute_instances = self.compute_service.instances
    taskqueue = self.testbed.get_stub('taskqueue')

    instance = 'test-instance'
    zone = 'test-zone'
    start_url = '/compute/%s/%s/%s' % (compute_page.ACTION_START,
                                       instance, zone)

    # Check all intermediate states.
    for instance_status in ['PROVISIONING', 'STAGING', 'STOPPING']:
      self.compute_service.get_return_value = {
          compute_page.COMPUTE_STATUS: instance_status
      }

      self.makePostRequest(start_url)

      # Since if the instance neither RUNNING nor TERMINATED then a
      # new task should be queued.
      tasks = taskqueue.GetTasks('default')
      self.assertEqual(1, len(tasks))
      task = tasks[-1]
      self.assertEqual(start_url, task['url'])

      # Verify start() API not called.
      self.assertEqual(0, self.compute_service.instances.start.num_calls)

      # Simulate the start task running AGAIN but instance is RUNNING.
      taskqueue.FlushQueue('default')

  def testPostRestartWhenRunning(self):
    """Test restart action in a running state."""
    compute_instances = self.compute_service.instances
    taskqueue = self.testbed.get_stub('taskqueue')
    instance = 'test-instance'
    zone = 'test-zone'
    restart_url = '/compute/%s/%s/%s' % (compute_page.ACTION_RESTART,
                                       instance, zone)

    self.compute_service.get_return_value = {
        compute_page.COMPUTE_STATUS: 'RUNNING'
    }
    self.makePostRequest(restart_url)

    # A new task should be queued.
    tasks = taskqueue.GetTasks('default')
    self.assertEqual(1, len(tasks))
    task = tasks[-1]
    start_url = '/compute/%s/%s/%s' % (compute_page.ACTION_START,
                                       instance, zone)
    self.assertEqual(start_url, task['url'])

    # Verify stop() API called.
    self.assertEqual(1, self.compute_service.instances.stop.num_calls)

  def testPostRestartWhenNotRunning(self):
    """Test restart action in a non-running state."""
    compute_instances = self.compute_service.instances
    taskqueue = self.testbed.get_stub('taskqueue')
    instance = 'test-instance'
    zone = 'test-zone'
    restart_url = '/compute/%s/%s/%s' % (compute_page.ACTION_RESTART,
                                         instance, zone)

    for status in ['PROVISIONING', 'STAGING', 'STOPPING', 'TERMINATED']:
      self.compute_service.get_return_value = {
          compute_page.COMPUTE_STATUS: status
      }
      self.makePostRequest(restart_url)

      # No task should be queued.
      tasks = taskqueue.GetTasks('default')
      self.assertEqual(0, len(tasks))

      # Verify stop() API not called.
      self.assertEqual(0, self.compute_service.instances.stop.num_calls)