server/frontend_unittest.py

   1 #!/usr/bin/python
   2 #
   3 # Copyright Gregory P. Smith, Google Inc 2008
   4 # Released under the GPL v2
   5
   6 """Tests for server.frontend."""
   7
   8 from cStringIO import StringIO
   9 import os, sys, unittest
  10 import common
  11 from autotest_lib.client.common_lib import global_config
  12 from autotest_lib.client.common_lib import utils
  13 from autotest_lib.client.common_lib.test_utils import mock
  14 from autotest_lib.frontend.afe import rpc_client_lib
  15 from autotest_lib.server import frontend
  16
  17 GLOBAL_CONFIG = global_config.global_config
  18
  19
  20 class BaseRpcClientTest(unittest.TestCase):
  21     def setUp(self):
  22         self.god = mock.mock_god()
  23         self.god.mock_up(rpc_client_lib, 'rpc_client_lib')
  24         self.god.stub_function(utils, 'send_email')
  25         self._saved_environ = dict(os.environ)
  26         if 'AUTOTEST_WEB' in os.environ:
  27             del os.environ['AUTOTEST_WEB']
  28
  29
  30     def tearDown(self):
  31         self.god.unstub_all()
  32         os.environ.clear()
  33         os.environ.update(self._saved_environ)
  34
  35
  36 class RpcClientTest(BaseRpcClientTest):
  37     def test_init(self):
  38         os.environ['LOGNAME'] = 'unittest-user'
  39         GLOBAL_CONFIG.override_config_value('SERVER', 'hostname', 'test-host')
  40         rpc_client_lib.authorization_headers.expect_call(
  41                 'unittest-user', 'http://test-host').and_return(
  42                         {'AUTHORIZATION': 'unittest-user'})
  43         rpc_client_lib.get_proxy.expect_call(
  44                 'http://test-host/path',
  45                 headers={'AUTHORIZATION': 'unittest-user'})
  46         frontend.RpcClient('/path', None, None, None, None, None)
  47         self.god.check_playback()
  48
  49
  50 class AFETest(BaseRpcClientTest):
  51     def test_result_notify(self):
  52         class fake_job(object):
  53             result = True
  54             name = 'nameFoo'
  55             id = 'idFoo'
  56             results_platform_map = {'NORAD' : {'Seeking_Joshua': ['WOPR']}}
  57         GLOBAL_CONFIG.override_config_value('SERVER', 'hostname', 'chess')
  58         rpc_client_lib.authorization_headers.expect_call(
  59                 'david', 'http://chess').and_return(
  60                         {'AUTHORIZATION': 'david'})
  61         rpc_client_lib.get_proxy.expect_call(
  62                 'http://chess/afe/server/rpc/',
  63                 headers={'AUTHORIZATION': 'david'})
  64         self.god.stub_function(utils, 'send_email')
  65         utils.send_email.expect_any_call()
  66
  67         my_afe = frontend.AFE(user='david')
  68
  69         fake_stdout = StringIO()
  70         self.god.stub_with(sys, 'stdout', fake_stdout)
  71         my_afe.result_notify(fake_job, 'userA', 'userB')
  72         self.god.unstub(sys, 'stdout')
  73         fake_stdout = fake_stdout.getvalue()
  74
  75         self.god.check_playback()
  76
  77         self.assert_('PASSED' in fake_stdout)
  78         self.assert_('WOPR' in fake_stdout)
  79         self.assert_('http://chess/tko/compose_query.cgi?' in fake_stdout)
  80         self.assert_('columns=test' in fake_stdout)
  81         self.assert_('rows=machine_group' in fake_stdout)
  82         self.assert_("condition=tag~'idFoo-%25'" in fake_stdout)
  83         self.assert_('title=Report' in fake_stdout)
  84         self.assert_('Sending email' in fake_stdout)
  85
  86
  87
  88 if __name__ == '__main__':
  89     unittest.main()