client/common_lib/cros/fake_device_server/commands_unittest.py

#!/usr/bin/python2

# Copyright 2014 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

"""Unit tests for commands.py."""

import copy
import mox
import unittest

import common
from fake_device_server import commands
from fake_device_server import fake_oauth
from fake_device_server import fail_control
from fake_device_server import server_errors


class CommandsTest(mox.MoxTestBase):
    """Tests for the Commands class.

    Note unlike other unittests in this project, I set the api_key for all
    tests. This makes the logic easier to read because of the additional
    dictionary mapping of
    # commands.devices_commands[(id, api_key)] = dict of commands by command id.
    """

    def setUp(self):
        """Sets up mox and a ticket / registration objects."""
        mox.MoxTestBase.setUp(self)
        # Use a fake OAuth module to work around the hack that this
        # module bypass cherrypy by directly invoking commands.GET.
        self.oauth = fake_oauth.FakeOAuth()
        self.fail_control = fail_control.FailControl()
        self.commands = commands.Commands(self.oauth, self.fail_control)


    def testCreateCommand(self):
        """Tests that we can create a new command."""
        DEVICE_ID = '1234awesomeDevice'
        GOOD_COMMAND = {
            'deviceId': DEVICE_ID,
            'name': 'base._vendorCommand',
            'base': {
                '_vendorCommand': {
                    'name': 'specialCommand',
                    'kind': 'buffetSpecialCommand',
                }
            }
        }

        self.commands.new_device(DEVICE_ID)
        new_command = self.commands.create_command(GOOD_COMMAND)
        self.assertTrue('id' in new_command)
        command_id = new_command['id']
        self.assertEqual(new_command['state'], 'queued')
        self.assertEqual(
                self.commands.device_commands[DEVICE_ID][command_id],
                new_command)

        # Test command without necessary nesting.
        bad_command = {'base': {}}
        self.assertRaises(server_errors.HTTPError,
                          self.commands.create_command, bad_command)

        # Test adding a good command to an unknown device.
        BAD_COMMAND = copy.deepcopy(GOOD_COMMAND)
        BAD_COMMAND['deviceId'] = 'not_a_real_device'
        self.assertRaises(server_errors.HTTPError,
                          self.commands.create_command, BAD_COMMAND)


    def testGet(self):
        """Tests that we can retrieve a command correctly."""
        DEVICE_ID = 'device_id'
        COMMAND_ID = 'command_id'
        COMMAND_RESOURCE = {'faked': 'out'}
        self.commands.new_device(DEVICE_ID)
        self.commands.device_commands[DEVICE_ID][COMMAND_ID] = COMMAND_RESOURCE
        returned_json = self.commands.GET(COMMAND_ID, deviceId=DEVICE_ID)
        self.assertEquals(returned_json, COMMAND_RESOURCE)

        BAD_COMMAND_ID = 'fubar'
        # Non-existing command.
        self.assertRaises(server_errors.HTTPError,
                          self.commands.GET, BAD_COMMAND_ID)


    def testListing(self):
        """Tests that we can get a listing back correctly using the GET method.
        """
        DEVICE_ID = 'device_id'
        COMMAND = {
            'name': 'base.reboot',
            'deviceId': DEVICE_ID,
        }
        self.commands.new_device(DEVICE_ID)
        command1 = self.commands.create_command(copy.deepcopy(COMMAND))
        command2 = self.commands.create_command(copy.deepcopy(COMMAND))
        command1_id = command1['id']
        command2_id = command2['id']
        self.commands.device_commands[DEVICE_ID][command1_id]['state'] = \
                'inProgress'

        # Without state should return all commands.
        def check_has_commands(expected_ids, state=None):
            """Check that we get all the commands we expect given a state.

            @param expected_ids: list of string command ids.
            @param state: Optional state to filter on (a string like 'queued'

            """
            returned_json = self.commands.GET(deviceId=DEVICE_ID, state=state)
            self.assertEqual('clouddevices#commandsListResponse',
                             returned_json['kind'])
            self.assertTrue('commands' in returned_json)
            returned_command_ids = [command['id']
                                    for command in returned_json['commands']]
            self.assertEqual(sorted(returned_command_ids), sorted(expected_ids))

        check_has_commands([command1_id, command2_id])
        check_has_commands([command1_id], state='inProgress')


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