• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
9import copy
10import mox
11import unittest
12
13import common
14from fake_device_server import commands
15from fake_device_server import fake_oauth
16from fake_device_server import fail_control
17from fake_device_server import server_errors
18
19
20class 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
126if __name__ == '__main__':
127    unittest.main()
128