python/google/appengine/tools/devappserver2/admin/cron_handler_test.py

   1 #!/usr/bin/env python
   2 #
   3 # Copyright 2007 Google Inc.
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the "License");
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an "AS IS" BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17 """Tests for devappserver2.admin.cron_handler."""
  18
  19 import datetime
  20 import traceback
  21 import unittest
  22
  23 import google
  24
  25 import mox
  26 import webapp2
  27
  28 from google.appengine.api import croninfo
  29 from google.appengine.api import yaml_errors
  30 from google.appengine.tools.devappserver2 import dispatcher
  31 from google.appengine.tools.devappserver2.admin import cron_handler
  32
  33 CRON_INFO_EXTERNAL = croninfo.CronInfoExternal(cron=[
  34     croninfo.CronEntry(
  35         url='/foo',
  36         schedule='every day 00:00',
  37         timezone='UTC',
  38         description='description',
  39         target='version'),
  40     croninfo.CronEntry(
  41         url='/bar',
  42         schedule='every day 00:00',
  43         description='description'),
  44     ])
  45
  46
  47 class CronHandlerTest(unittest.TestCase):
  48
  49   def setUp(self):
  50     self.mox = mox.Mox()
  51     self._pytz = cron_handler.pytz
  52
  53   def tearDown(self):
  54     cron_handler.pytz = self._pytz
  55     self.mox.UnsetStubs()
  56
  57   def test_post(self):
  58     self.mox.StubOutWithMock(cron_handler.CronHandler, 'dispatcher')
  59     request = webapp2.Request.blank('/cron', POST={'url': '/url'})
  60     response = webapp2.Response()
  61     handler = cron_handler.CronHandler(request, response)
  62     handler.dispatcher = self.mox.CreateMock(dispatcher.Dispatcher)
  63     handler.dispatcher.add_request(
  64         method='GET',
  65         relative_url='/url',
  66         headers=[('X-AppEngine-Cron', 'true')],
  67         body='',
  68         source_ip='0.1.0.1').AndReturn(
  69             dispatcher.ResponseTuple('500 Internal Server Error', [], ''))
  70     self.mox.ReplayAll()
  71     handler.post()
  72     self.mox.VerifyAll()
  73     self.assertEqual(500, response.status_int)
  74
  75   def test_get_with_pytz(self, pytz=object()):
  76     cron_handler.pytz = pytz
  77     jobs = object()
  78     request = webapp2.Request.blank('/cron')
  79     response = webapp2.Response()
  80     handler = cron_handler.CronHandler(request, response)
  81     self.mox.StubOutWithMock(handler, '_get_cron_jobs')
  82     self.mox.StubOutWithMock(handler, 'render')
  83     handler._get_cron_jobs().AndReturn(jobs)
  84     handler.render('cron.html',
  85                    {'has_pytz': bool(pytz), 'cronjobs': jobs}).AndReturn(
  86                        'template')
  87     self.mox.ReplayAll()
  88     handler.get()
  89     self.mox.VerifyAll()
  90     self.assertEqual('template', response.body)
  91
  92   def test_get_without_pytz(self):
  93     self.test_get_with_pytz(pytz=None)
  94
  95   def test_get_with_invalid_cron_yaml(self):
  96     cron_handler.pytz = None
  97     request = webapp2.Request.blank('/cron')
  98     response = webapp2.Response()
  99     handler = cron_handler.CronHandler(request, response)
 100     self.mox.StubOutWithMock(handler, '_get_cron_jobs')
 101     self.mox.StubOutWithMock(handler, 'render')
 102     self.mox.StubOutWithMock(traceback, 'format_exc')
 103     handler._get_cron_jobs().AndRaise(yaml_errors.Error)
 104     traceback.format_exc().AndReturn('traceback')
 105     handler.render(
 106         'cron.html',
 107         {'has_pytz': False, 'cron_error': 'traceback'}).AndReturn('template')
 108     self.mox.ReplayAll()
 109     handler.get()
 110     self.mox.VerifyAll()
 111     self.assertEqual('template', response.body)
 112
 113   @unittest.skipIf(cron_handler.pytz is None, 'requires pytz')
 114   def test_get_cron_jobs_with_pytz(self):
 115     handler = cron_handler.CronHandler(None, None)
 116     self.mox.StubOutWithMock(handler, '_parse_cron_yaml')
 117     handler._parse_cron_yaml().AndReturn(CRON_INFO_EXTERNAL)
 118     self.mox.ReplayAll()
 119     cron_jobs = handler._get_cron_jobs()
 120     self.mox.VerifyAll()
 121     self.assertEqual(2, len(cron_jobs))
 122     next_runs = [cron_jobs[0].pop('times'), cron_jobs[1].pop('times')]
 123     self.assertEqual(CRON_INFO_EXTERNAL.cron[0].ToDict(), cron_jobs[0])
 124     self.assertEqual(CRON_INFO_EXTERNAL.cron[1].ToDict(), cron_jobs[1])
 125     for next_run in next_runs:
 126       self.assertEqual(3, len(next_run))
 127       for run in next_run:
 128         datetime.datetime.strptime(run['runtime'], '%Y-%m-%d %H:%M:%SZ')
 129
 130   def test_get_cron_jobs_without_pytz(self):
 131     cron_handler.pytz = None
 132     handler = cron_handler.CronHandler(None, None)
 133     self.mox.StubOutWithMock(handler, '_parse_cron_yaml')
 134     handler._parse_cron_yaml().AndReturn(CRON_INFO_EXTERNAL)
 135     self.mox.ReplayAll()
 136     cron_jobs = handler._get_cron_jobs()
 137     self.mox.VerifyAll()
 138     self.assertEqual(2, len(cron_jobs))
 139     self.assertEqual(CRON_INFO_EXTERNAL.cron[0].ToDict(), cron_jobs[0])
 140     next_run = cron_jobs[1].pop('times')
 141     self.assertEqual(CRON_INFO_EXTERNAL.cron[1].ToDict(), cron_jobs[1])
 142     self.assertEqual(3, len(next_run))
 143     for run in next_run:
 144       datetime.datetime.strptime(run['runtime'], '%Y-%m-%d %H:%M:%SZ')
 145
 146   def test_get_cron_jobs_no_cron_yaml(self):
 147     cron_handler.pytz = None
 148     handler = cron_handler.CronHandler(None, None)
 149     self.mox.StubOutWithMock(handler, '_parse_cron_yaml')
 150     handler._parse_cron_yaml().AndReturn(None)
 151     self.mox.ReplayAll()
 152     self.assertEqual([], handler._get_cron_jobs())
 153     self.mox.VerifyAll()
 154
 155 if __name__ == '__main__':
 156   unittest.main()