Open Source Monorail

appengine/monorail/services/test/features_svc_test.py

+# Copyright 2016 The Chromium Authors. All rights reserved.
+# Use of this source code is govered by a BSD-style
+# license that can be found in the LICENSE file or at
+# https://developers.google.com/open-source/licenses/bsd
+
+"""Unit tests for features_svc module."""
+
+import unittest
+
+import mox
+
+from features import filterrules_helpers
+from framework import sql
+from services import features_svc
+from testing import fake
+from tracker import tracker_bizobj
+from tracker import tracker_constants
+
+
+class FeaturesServiceTest(unittest.TestCase):
+
+  def MakeMockTable(self):
+    return self.mox.CreateMock(sql.SQLTableManager)
+
+  def setUp(self):
+    self.mox = mox.Mox()
+    self.cnxn = self.mox.CreateMock(sql.MonorailConnection)
+    self.cache_manager = fake.CacheManager()
+
+    self.features_service = features_svc.FeaturesService(self.cache_manager)
+
+    for table_var in [
+        'user2savedquery_tbl', 'quickedithistory_tbl',
+        'quickeditmostrecent_tbl', 'savedquery_tbl',
+        'savedqueryexecutesinproject_tbl', 'project2savedquery_tbl',
+        'filterrule_tbl']:
+      setattr(self.features_service, table_var, self.MakeMockTable())
+
+  def tearDown(self):
+    self.mox.UnsetStubs()
+    self.mox.ResetAll()
+
+  ### quickedit command history
+
+  def testGetRecentCommands(self):
+    self.features_service.quickedithistory_tbl.Select(
+        self.cnxn, cols=['slot_num', 'command', 'comment'],
+        user_id=1, project_id=12345).AndReturn(
+        [(1, 'status=New', 'Brand new issue')])
+    self.features_service.quickeditmostrecent_tbl.SelectValue(
+        self.cnxn, 'slot_num', default=1, user_id=1, project_id=12345
+        ).AndReturn(1)
+    self.mox.ReplayAll()
+    slots, recent_slot_num = self.features_service.GetRecentCommands(
+        self.cnxn, 1, 12345)
+    self.mox.VerifyAll()
+
+    self.assertEqual(1, recent_slot_num)
+    self.assertEqual(
+        len(tracker_constants.DEFAULT_RECENT_COMMANDS), len(slots))
+    self.assertEqual('status=New', slots[0][1])
+
+  def testStoreRecentCommand(self):
+    self.features_service.quickedithistory_tbl.InsertRow(
+        self.cnxn, replace=True, user_id=1, project_id=12345,
+        slot_num=1, command='status=New', comment='Brand new issue')
+    self.features_service.quickeditmostrecent_tbl.InsertRow(
+        self.cnxn, replace=True, user_id=1, project_id=12345,
+        slot_num=1)
+    self.mox.ReplayAll()
+    self.features_service.StoreRecentCommand(
+        self.cnxn, 1, 12345, 1, 'status=New', 'Brand new issue')
+    self.mox.VerifyAll()
+
+  def testExpungeQuickEditHistory(self):
+    self.features_service.quickeditmostrecent_tbl.Delete(
+        self.cnxn, project_id=12345)
+    self.features_service.quickedithistory_tbl.Delete(
+        self.cnxn, project_id=12345)
+    self.mox.ReplayAll()
+    self.features_service.ExpungeQuickEditHistory(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+
+  ### Saved User and Project Queries
+
+  def testGetSavedQuery(self):
+    self.features_service.savedquery_tbl.Select(
+        self.cnxn, cols=features_svc.SAVEDQUERY_COLS, id=[1]).AndReturn(
+        [(1, 'query1', 100, 'owner:me')])
+    self.features_service.savedqueryexecutesinproject_tbl.Select(
+        self.cnxn, cols=features_svc.SAVEDQUERYEXECUTESINPROJECT_COLS,
+        query_id=[1]).AndReturn([(1, 12345)])
+    self.mox.ReplayAll()
+    saved_query = self.features_service.GetSavedQuery(
+        self.cnxn, 1)
+    self.mox.VerifyAll()
+    self.assertEqual(1, saved_query.query_id)
+    self.assertEqual('query1', saved_query.name)
+    self.assertEqual(100, saved_query.base_query_id)
+    self.assertEqual('owner:me', saved_query.query)
+    self.assertEqual([12345], saved_query.executes_in_project_ids)
+
+  def SetUpUsersSavedQueries(self):
+    query = tracker_bizobj.MakeSavedQuery(1, 'query1', 100, 'owner:me')
+    self.features_service.saved_query_cache.CacheItem(1, [query])
+    self.features_service.user2savedquery_tbl.Select(
+        self.cnxn,
+        cols=features_svc.SAVEDQUERY_COLS + ['user_id', 'subscription_mode'],
+        left_joins=[('SavedQuery ON query_id = id', [])],
+        order_by=[('rank', [])], user_id=[2]).AndReturn(
+        [(2, 'query2', 100, 'status:New', 2, 'Sub_Mode')])
+    self.features_service.savedqueryexecutesinproject_tbl.Select(
+          self.cnxn, cols=features_svc.SAVEDQUERYEXECUTESINPROJECT_COLS,
+          query_id=set([2])).AndReturn([(2, 12345)])
+
+  def testGetUsersSavedQueriesDict(self):
+    self.SetUpUsersSavedQueries()
+    self.mox.ReplayAll()
+    results_dict = self.features_service._GetUsersSavedQueriesDict(
+        self.cnxn, [1, 2])
+    self.mox.VerifyAll()
+    self.assertIn(1, results_dict)
+    self.assertIn(2, results_dict)
+
+  def testGetSavedQueriesByUserID(self):
+    self.SetUpUsersSavedQueries()
+    self.mox.ReplayAll()
+    saved_queries = self.features_service.GetSavedQueriesByUserID(
+        self.cnxn, 2)
+    self.mox.VerifyAll()
+    self.assertEqual(1, len(saved_queries))
+    self.assertEqual(2, saved_queries[0].query_id)
+
+  def SetUpCannedQueriesForProjects(self):
+    self.features_service.project2savedquery_tbl.Select(
+        self.cnxn, cols=['project_id'] + features_svc.SAVEDQUERY_COLS,
+        left_joins=[('SavedQuery ON query_id = id', [])],
+        order_by=[('rank', [])], project_id=[12345]).AndReturn(
+        [(12345, 1, 'query1', 100, 'owner:me')])
+
+  def testGetCannedQueriesForProjects(self):
+    self.SetUpCannedQueriesForProjects()
+    self.mox.ReplayAll()
+    results_dict = self.features_service.GetCannedQueriesForProjects(
+        self.cnxn, [12345])
+    self.mox.VerifyAll()
+    self.assertIn(12345, results_dict)
+
+  def testGetCannedQueriesByProjectID(self):
+    self.SetUpCannedQueriesForProjects()
+    self.mox.ReplayAll()
+    result = self.features_service.GetCannedQueriesByProjectID(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+    self.assertEqual(1, len(result))
+    self.assertEqual(1, result[0].query_id)
+
+  def SetUpUpdateSavedQueries(self, commit=True):
+    query1 = tracker_bizobj.MakeSavedQuery(1, 'query1', 100, 'owner:me')
+    query2 = tracker_bizobj.MakeSavedQuery(None, 'query2', 100, 'status:New')
+    saved_queries = [query1, query2]
+    savedquery_rows = [
+        (sq.query_id or None, sq.name, sq.base_query_id, sq.query)
+        for sq in saved_queries]
+    self.features_service.savedquery_tbl.Delete(
+        self.cnxn, id=[1], commit=commit)
+    self.features_service.savedquery_tbl.InsertRows(
+        self.cnxn, features_svc.SAVEDQUERY_COLS, savedquery_rows, commit=commit,
+        return_generated_ids=True).AndReturn([11, 12])
+    return saved_queries
+
+  def testUpdateSavedQueries(self):
+    saved_queries = self.SetUpUpdateSavedQueries()
+    self.mox.ReplayAll()
+    self.features_service._UpdateSavedQueries(
+        self.cnxn, saved_queries, True)
+    self.mox.VerifyAll()
+
+  def testUpdateCannedQueries(self):
+    self.features_service.project2savedquery_tbl.Delete(
+        self.cnxn, project_id=12345, commit=False)
+    canned_queries = self.SetUpUpdateSavedQueries(False)
+    project2savedquery_rows = [(12345, 0, 1), (12345, 1, 12)]
+    self.features_service.project2savedquery_tbl.InsertRows(
+        self.cnxn, features_svc.PROJECT2SAVEDQUERY_COLS,
+        project2savedquery_rows, commit=False)
+    self.cnxn.Commit()
+    self.mox.ReplayAll()
+    self.features_service.UpdateCannedQueries(
+        self.cnxn, 12345, canned_queries)
+    self.mox.VerifyAll()
+
+  def testUpdateUserSavedQueries(self):
+    saved_queries = self.SetUpUpdateSavedQueries(False)
+    self.features_service.savedqueryexecutesinproject_tbl.Delete(
+        self.cnxn, query_id=[1], commit=False)
+    self.features_service.user2savedquery_tbl.Delete(
+        self.cnxn, user_id=1, commit=False)
+    user2savedquery_rows = [
+      (1, 0, 1, 'noemail'), (1, 1, 12, 'noemail')]
+    self.features_service.user2savedquery_tbl.InsertRows(
+        self.cnxn, features_svc.USER2SAVEDQUERY_COLS,
+        user2savedquery_rows, commit=False)
+    self.features_service.savedqueryexecutesinproject_tbl.InsertRows(
+        self.cnxn, features_svc.SAVEDQUERYEXECUTESINPROJECT_COLS, [],
+        commit=False)
+    self.cnxn.Commit()
+    self.mox.ReplayAll()
+    self.features_service.UpdateUserSavedQueries(
+        self.cnxn, 1, saved_queries)
+    self.mox.VerifyAll()
+
+  ### Subscriptions
+
+  def testGetSubscriptionsInProjects(self):
+    join_str = (
+        'SavedQueryExecutesInProject ON '
+        'SavedQueryExecutesInProject.query_id = User2SavedQuery.query_id')
+    self.features_service.user2savedquery_tbl.Select(
+        self.cnxn, cols=['user_id'], distinct=True,
+        joins=[(join_str, [])],
+        subscription_mode='immediate', project_id=12345).AndReturn(
+        [(1, 'asd'), (2, 'efg')])
+    self.SetUpUsersSavedQueries()
+    self.mox.ReplayAll()
+    result = self.features_service.GetSubscriptionsInProjects(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+    self.assertIn(1, result)
+    self.assertIn(2, result)
+
+  def testExpungeSavedQueriesExecuteInProject(self):
+    self.features_service.savedqueryexecutesinproject_tbl.Delete(
+        self.cnxn, project_id=12345)
+    self.features_service.project2savedquery_tbl.Select(
+        self.cnxn, cols=['query_id'], project_id=12345).AndReturn(
+        [(1, 'asd'), (2, 'efg')])
+    self.features_service.project2savedquery_tbl.Delete(
+        self.cnxn, project_id=12345)
+    self.features_service.savedquery_tbl.Delete(
+        self.cnxn, id=[1, 2])
+    self.mox.ReplayAll()
+    self.features_service.ExpungeSavedQueriesExecuteInProject(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+
+  ### Filter Rules
+
+  def testDeserializeFilterRules(self):
+    filterrule_rows = [
+        (12345, 0, 'predicate1', 'default_status:New'),
+        (12345, 1, 'predicate2', 'default_owner_id:1 add_cc_id:2'),
+    ]
+    result_dict = self.features_service._DeserializeFilterRules(
+        filterrule_rows)
+    self.assertIn(12345, result_dict)
+    self.assertEqual(2, len(result_dict[12345]))
+    self.assertEqual('New', result_dict[12345][0].default_status)
+    self.assertEqual(1, result_dict[12345][1].default_owner_id)
+    self.assertEqual([2], result_dict[12345][1].add_cc_ids)
+
+  def testDeserializeRuleConsequence(self):
+    consequence = ('default_status:New default_owner_id:1 add_cc_id:2'
+                   ' add_label:label1 add_label:label2 add_notify:admin')
+    (default_status, default_owner_id, add_cc_ids, add_labels,
+     add_notify) = self.features_service._DeserializeRuleConsequence(
+        consequence)
+    self.assertEqual('New', default_status)
+    self.assertEqual(1, default_owner_id)
+    self.assertEqual([2], add_cc_ids)
+    self.assertEqual(['label1', 'label2'], add_labels)
+    self.assertEqual(['admin'], add_notify)
+
+  def SetUpGetFilterRulesByProjectIDs(self):
+    filterrule_rows = [
+        (12345, 0, 'predicate1', 'default_status:New'),
+        (12345, 1, 'predicate2', 'default_owner_id:1 add_cc_id:2'),
+    ]
+
+    self.features_service.filterrule_tbl.Select(
+        self.cnxn, cols=features_svc.FILTERRULE_COLS,
+        project_id=[12345]).AndReturn(filterrule_rows)
+
+  def testGetFilterRulesByProjectIDs(self):
+    self.SetUpGetFilterRulesByProjectIDs()
+    self.mox.ReplayAll()
+    result = self.features_service._GetFilterRulesByProjectIDs(
+        self.cnxn, [12345])
+    self.mox.VerifyAll()
+    self.assertIn(12345, result)
+    self.assertEqual(2, len(result[12345]))
+
+  def testGetFilterRules(self):
+    self.SetUpGetFilterRulesByProjectIDs()
+    self.mox.ReplayAll()
+    result = self.features_service.GetFilterRules(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+    self.assertEqual(2, len(result))
+
+  def testSerializeRuleConsequence(self):
+    rule = filterrules_helpers.MakeRule(
+        'predicate', 'New', 1, [1, 2], ['label1', 'label2'], ['admin'])
+    result = self.features_service._SerializeRuleConsequence(rule)
+    self.assertEqual('add_label:label1 add_label:label2 default_status:New'
+                     ' default_owner_id:1 add_cc_id:1 add_cc_id:2'
+                     ' add_notify:admin', result)
+
+  def testUpdateFilterRules(self):
+    self.features_service.filterrule_tbl.Delete(self.cnxn, project_id=12345)
+    rows = [
+        (12345, 0, 'predicate1', 'add_label:label1 add_label:label2'
+                                 ' default_status:New default_owner_id:1'
+                                 ' add_cc_id:1 add_cc_id:2 add_notify:admin'),
+        (12345, 1, 'predicate2', 'add_label:label2 add_label:label3'
+                                 ' default_status:Fixed default_owner_id:2'
+                                 ' add_cc_id:1 add_cc_id:2 add_notify:admin2')
+    ]
+    self.features_service.filterrule_tbl.InsertRows(
+        self.cnxn, features_svc.FILTERRULE_COLS, rows)
+    rule1 = filterrules_helpers.MakeRule(
+        'predicate1', 'New', 1, [1, 2], ['label1', 'label2'], ['admin'])
+    rule2 = filterrules_helpers.MakeRule(
+        'predicate2', 'Fixed', 2, [1, 2], ['label2', 'label3'], ['admin2'])
+    self.mox.ReplayAll()
+    self.features_service.UpdateFilterRules(
+        self.cnxn, 12345, [rule1, rule2])
+    self.mox.VerifyAll()
+
+  def testExpungeFilterRules(self):
+    self.features_service.filterrule_tbl.Delete(self.cnxn, project_id=12345)
+    self.mox.ReplayAll()
+    self.features_service.ExpungeFilterRules(
+        self.cnxn, 12345)
+    self.mox.VerifyAll()
+
+
+if __name__ == '__main__':
+  unittest.main()