cron_servlet_test.py revision b2df76ea8fec9e32f6f3718986dba0d95315b29c 
1#!/usr/bin/env python
2# Copyright 2013 The Chromium 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 unittest
7
8from cron_servlet import CronServlet
9from empty_dir_file_system import EmptyDirFileSystem
10from local_file_system import LocalFileSystem
11from mock_file_system import MockFileSystem
12from servlet import Request
13from test_branch_utility import TestBranchUtility
14from test_file_system import TestFileSystem
15from test_util import EnableLogging
16
17# NOTE(kalman): The ObjectStore created by the CronServlet is backed onto our
18# fake AppEngine memcache/datastore, so the tests aren't isolated.
19class _TestDelegate(CronServlet.Delegate):
20  def __init__(self):
21    self.host_file_systems = []
22
23  def CreateBranchUtility(self, object_store_creator):
24    return TestBranchUtility()
25
26  def CreateHostFileSystemForBranch(self, branch):
27    host_file_system = MockFileSystem(LocalFileSystem.Create())
28    self.host_file_systems.append(host_file_system)
29    return host_file_system
30
31  def CreateAppSamplesFileSystem(self, object_store_creator):
32    return EmptyDirFileSystem()
33
34class CronServletTest(unittest.TestCase):
35  @EnableLogging('info')
36  def testEverything(self):
37    # All these tests are dependent (see above comment) so lump everything in
38    # the one test.
39    delegate = _TestDelegate()
40
41    # Test that the cron runs successfully.
42    response = CronServlet(Request.ForTest('trunk'),
43                           delegate_for_test=delegate).Get()
44    self.assertEqual(1, len(delegate.host_file_systems))
45    self.assertEqual(200, response.status)
46
47    # When re-running, all file systems should be Stat()d the same number of
48    # times, but the second round shouldn't have been re-Read() since the
49    # Stats haven't changed.
50    response = CronServlet(Request.ForTest('trunk'),
51                           delegate_for_test=delegate).Get()
52    self.assertEqual(2, len(delegate.host_file_systems))
53    self.assertTrue(*delegate.host_file_systems[1].CheckAndReset(
54        read_count=0,
55        stat_count=delegate.host_file_systems[0].GetStatCount()))
56
57
58if __name__ == '__main__':
59  unittest.main()
60