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