Home | History | Annotate | Download | only in fake_device_server 
      1 #! /usr/bin/python
      2 
      3 # Copyright 2014 The Chromium OS Authors. All rights reserved.
      4 # Use of this source code is governed by a BSD-style license that can be
      5 # found in the LICENSE file.
      6 
      7 """Unit tests for commands.py."""
      8 
      9 import copy
     10 import mox
     11 import unittest
     12 
     13 import common
     14 from fake_device_server import commands
     15 from fake_device_server import fake_oauth
     16 from fake_device_server import fail_control
     17 from fake_device_server import server_errors
     18 
     19 
     20 class CommandsTest(mox.MoxTestBase):
     21     """Tests for the Commands class.
     22 
     23     Note unlike other unittests in this project, I set the api_key for all
     24     tests. This makes the logic easier to read because of the additional
     25     dictionary mapping of
     26     # commands.devices_commands[(id, api_key)] = dict of commands by command id.
     27     """
     28 
     29     def setUp(self):
     30         """Sets up mox and a ticket / registration objects."""
     31         mox.MoxTestBase.setUp(self)
     32         # Use a fake OAuth module to work around the hack that this
     33         # module bypass cherrypy by directly invoking commands.GET.
     34         self.oauth = fake_oauth.FakeOAuth()
     35         self.fail_control = fail_control.FailControl()
     36         self.commands = commands.Commands(self.oauth, self.fail_control)
     37 
     38 
     39     def testCreateCommand(self):
     40         """Tests that we can create a new command."""
     41         DEVICE_ID = '1234awesomeDevice'
     42         GOOD_COMMAND = {
     43             'deviceId': DEVICE_ID,
     44             'name': 'base._vendorCommand',
     45             'base': {
     46                 '_vendorCommand': {
     47                     'name': 'specialCommand',
     48                     'kind': 'buffetSpecialCommand',
     49                 }
     50             }
     51         }
     52 
     53         self.commands.new_device(DEVICE_ID)
     54         new_command = self.commands.create_command(GOOD_COMMAND)
     55         self.assertTrue('id' in new_command)
     56         command_id = new_command['id']
     57         self.assertEqual(new_command['state'], 'queued')
     58         self.assertEqual(
     59                 self.commands.device_commands[DEVICE_ID][command_id],
     60                 new_command)
     61 
     62         # Test command without necessary nesting.
     63         bad_command = {'base': {}}
     64         self.assertRaises(server_errors.HTTPError,
     65                           self.commands.create_command, bad_command)
     66 
     67         # Test adding a good command to an unknown device.
     68         BAD_COMMAND = copy.deepcopy(GOOD_COMMAND)
     69         BAD_COMMAND['deviceId'] = 'not_a_real_device'
     70         self.assertRaises(server_errors.HTTPError,
     71                           self.commands.create_command, BAD_COMMAND)
     72 
     73 
     74     def testGet(self):
     75         """Tests that we can retrieve a command correctly."""
     76         DEVICE_ID = 'device_id'
     77         COMMAND_ID = 'command_id'
     78         COMMAND_RESOURCE = {'faked': 'out'}
     79         self.commands.new_device(DEVICE_ID)
     80         self.commands.device_commands[DEVICE_ID][COMMAND_ID] = COMMAND_RESOURCE
     81         returned_json = self.commands.GET(COMMAND_ID, deviceId=DEVICE_ID)
     82         self.assertEquals(returned_json, COMMAND_RESOURCE)
     83 
     84         BAD_COMMAND_ID = 'fubar'
     85         # Non-existing command.
     86         self.assertRaises(server_errors.HTTPError,
     87                           self.commands.GET, BAD_COMMAND_ID)
     88 
     89 
     90     def testListing(self):
     91         """Tests that we can get a listing back correctly using the GET method.
     92         """
     93         DEVICE_ID = 'device_id'
     94         COMMAND = {
     95             'name': 'base.reboot',
     96             'deviceId': DEVICE_ID,
     97         }
     98         self.commands.new_device(DEVICE_ID)
     99         command1 = self.commands.create_command(copy.deepcopy(COMMAND))
    100         command2 = self.commands.create_command(copy.deepcopy(COMMAND))
    101         command1_id = command1['id']
    102         command2_id = command2['id']
    103         self.commands.device_commands[DEVICE_ID][command1_id]['state'] = \
    104                 'inProgress'
    105 
    106         # Without state should return all commands.
    107         def check_has_commands(expected_ids, state=None):
    108             """Check that we get all the commands we expect given a state.
    109 
    110             @param expected_ids: list of string command ids.
    111             @param state: Optional state to filter on (a string like 'queued'
    112 
    113             """
    114             returned_json = self.commands.GET(deviceId=DEVICE_ID, state=state)
    115             self.assertEqual('clouddevices#commandsListResponse',
    116                              returned_json['kind'])
    117             self.assertTrue('commands' in returned_json)
    118             returned_command_ids = [command['id']
    119                                     for command in returned_json['commands']]
    120             self.assertEqual(sorted(returned_command_ids), sorted(expected_ids))
    121 
    122         check_has_commands([command1_id, command2_id])
    123         check_has_commands([command1_id], state='inProgress')
    124 
    125 
    126 if __name__ == '__main__':
    127     unittest.main()
    128