chromium-cq-status: StatsTest utility class for testing CQ stats

appengine/chromium_cq_status/tests/stats_test.py

Index: appengine/chromium_cq_status/tests/stats_test.py
diff --git a/appengine/chromium_cq_status/tests/stats_test.py b/appengine/chromium_cq_status/tests/stats_test.py
new file mode 100644
index 0000000000000000000000000000000000000000..998436defe50364214f373a4c2fcc25cb62bc345
--- /dev/null
+++ b/appengine/chromium_cq_status/tests/stats_test.py
@@ -0,0 +1,65 @@
+# Copyright 2014 The Chromium Authors. All rights reserved.
+# Use of this source code is governed by a BSD-style license that can be
+# found in the LICENSE file.
+
+from datetime import datetime
+import json
+import os
+
+from appengine.utils import testing
+
+from appengine.path_mangler_hack import PathMangler
+with PathMangler(os.path.dirname(os.path.dirname(__file__))):
+  from appengine.chromium_cq_status import main
+  from appengine.chromium_cq_status.model.record import Record
+
+class StatsTest(testing.AppengineTestCase): # pragma: no cover
+  '''Utility class for stats tests that want to load/clear test Record data'''
+  app_module = main.app
+
+  def load_records(self, json_filename):
+    json_path = os.path.join(os.path.dirname(__file__), 'resources',
+        json_filename)
+    with open(json_path) as json_file:
+      for record in json.load(json_file):
+        self.mock_now(datetime.utcfromtimestamp(record['timestamp']))
+        Record(
+          id=record['key'],
+          tags=record['tags'],
+          fields=record['fields'],
+        ).put()
+
+  @staticmethod
+  def clear_records():
+    for record in Record.query():
+      record.key.delete()
+    assert Record.query().count() == 0
+
+  def test_load_records(self):
+    self.clear_records()
+    self.load_records('test_load_records.json')
+    self.assertEquals([{
+        'key': None,
+        'timestamp': 100,
+        'tags': ['tag_a', 'tag_b', 'tag_c'],
+        'fields': {}
+      }, {
+        'key': 'keyed_record',
+        'timestamp': 200,
+        'tags': [],
+        'fields': {
+          'field_a': 'value_a',
+          'field_b': 'value_b',
+          'field_c': 'value_c',
+        }
+      }],
+      [record.to_dict() for record in Record.query().order(Record.timestamp)])
+
+  def test_clear_records(self):
+    self.clear_records()
+    self.assertEquals(0, Record.query().count())
+    for _ in range(10):
+      Record().put()
+    self.assertEquals(10, Record.query().count())
+    self.clear_records()
+    self.assertEquals(0, Record.query().count())