• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/python2
2# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
3# Use of this source code is governed by a BSD-style license that can be
4# found in the LICENSE file.
5
6import mox
7import socket
8import unittest
9
10from config import rpm_config
11import rpm_dispatcher
12
13DUT_SAME_RPM1 = 'chromeos-rack8e-hostbs1'
14DUT_SAME_RPM2 = 'chromeos-rack8e-hostbs2'
15RPM_HOSTNAME = 'chromeos-rack8e-rpm1'
16DUT_DIFFERENT_RPM = 'chromeos-rack1-hostbs1'
17FAKE_DISPATCHER_URI = 'fake-dispatcher'
18FAKE_DISPATCHER_PORT = 9999
19FRONT_END_PORT = rpm_config.getint('RPM_INFRASTRUCTURE', 'frontend_port')
20PROPER_URI_FORMAT = 'http://%s:%d'
21
22
23class TestRPMDispatcher(mox.MoxTestBase):
24    """
25    Simple unit tests to verify that the RPM Dispatcher properly registers with
26    the frontend server, and also initializes and reuses the same RPM Controller
27    for DUT requests on the same RPM.
28
29    queue_request is the only public method of RPM Dispatcher, however its logic
30    is simple and relies mostly on the private methods; therefore, I am testing
31    primarily RPMDispatcher initialization and _get_rpm_controller (which calls
32    _create_rpm_controller) to verify correct implementation.
33    """
34
35    def setUp(self):
36        super(TestRPMDispatcher, self).setUp()
37        self.frontend_mock = self.mox.CreateMockAnything()
38        expected_uri = PROPER_URI_FORMAT % (FAKE_DISPATCHER_URI,
39                                            FAKE_DISPATCHER_PORT)
40        self.frontend_mock.register_dispatcher(expected_uri)
41        rpm_dispatcher.xmlrpclib.ServerProxy = self.mox.CreateMockAnything()
42        frontend_uri = 'http://%s:%d' % (socket.gethostname(), FRONT_END_PORT)
43        rpm_dispatcher.xmlrpclib.ServerProxy(frontend_uri).AndReturn(
44                self.frontend_mock)
45        rpm_dispatcher.atexit = self.mox.CreateMockAnything()
46        rpm_dispatcher.atexit.register(mox.IgnoreArg())
47        self.mox.ReplayAll()
48        self.dispatcher = rpm_dispatcher.RPMDispatcher(FAKE_DISPATCHER_URI,
49                                                       FAKE_DISPATCHER_PORT)
50
51
52    def testRegistration(self):
53        """
54        Make sure that as a dispatcher is initialized it properly registered
55        with the frontend server.
56        """
57        self.mox.VerifyAll()
58
59
60    def testGetSameRPMController(self):
61        """
62        Make sure that calls to _get_rpm_controller with DUT hostnames that
63        belong to the same RPM device create and retrieve the same RPMController
64        instance.
65        """
66        controller1 = self.dispatcher._get_rpm_controller(RPM_HOSTNAME)
67        controller2 = self.dispatcher._get_rpm_controller(RPM_HOSTNAME)
68        self.assertEquals(controller1, controller2)
69
70
71    def testGetDifferentRPMController(self):
72        """
73        Make sure that calls to _get_rpm_controller with DUT hostnames that
74        belong to the different RPM device create and retrieve different
75        RPMController instances.
76        """
77        controller1 = self.dispatcher._get_rpm_controller(DUT_SAME_RPM1)
78        controller2 = self.dispatcher._get_rpm_controller(DUT_DIFFERENT_RPM)
79        self.assertNotEquals(controller1, controller2)
80
81
82if __name__ == '__main__':
83    unittest.main()
84