| # 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 |
| |
| """Tests for the user service.""" |
| |
| import unittest |
| |
| import mox |
| |
| from google.appengine.ext import testbed |
| |
| from framework import sql |
| from proto import user_pb2 |
| from services import user_svc |
| from testing import fake |
| |
| |
| def SetUpGetUsers(user_service, cnxn): |
| """Set up expected calls to SQL tables.""" |
| user_service.user_tbl.Select( |
| cnxn, cols=user_svc.USER_COLS, user_id=[333L]).AndReturn( |
| [(333L, 'c@example.com', False, False, False, 'Spammer', |
| 'stay_same_issue', False, False, False, True)]) |
| user_service.actionlimit_tbl.Select( |
| cnxn, cols=user_svc.ACTIONLIMIT_COLS, user_id=[333L]).AndReturn([]) |
| user_service.dismissedcues_tbl.Select( |
| cnxn, cols=user_svc.DISMISSEDCUES_COLS, user_id=[333L]).AndReturn([]) |
| |
| |
| def MakeUserService(cache_manager, my_mox): |
| user_service = user_svc.UserService(cache_manager) |
| user_service.user_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| user_service.actionlimit_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| user_service.dismissedcues_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| return user_service |
| |
| |
| class UserTwoLevelCacheTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.testbed = testbed.Testbed() |
| self.testbed.activate() |
| self.testbed.init_memcache_stub() |
| |
| self.mox = mox.Mox() |
| self.cnxn = fake.MonorailConnection() |
| self.cache_manager = fake.CacheManager() |
| self.user_service = MakeUserService(self.cache_manager, self.mox) |
| |
| def testDeserializeUsersByID(self): |
| user_rows = [ |
| (111L, 'a@example.com', False, False, False, '', |
| 'stay_same_issue', False, False, False, True), |
| (222L, 'b@example.com', False, False, False, '', |
| 'next_in_list', False, False, False, True), |
| ] |
| actionlimit_rows = [] |
| dismissedcues_rows = [] |
| user_dict = self.user_service.user_2lc._DeserializeUsersByID( |
| user_rows, actionlimit_rows, dismissedcues_rows) |
| self.assertEqual(2, len(user_dict)) |
| self.assertEqual('a@example.com', user_dict[111L].email) |
| self.assertFalse(user_dict[111L].is_site_admin) |
| self.assertEqual('', user_dict[111L].banned) |
| self.assertFalse(user_dict[111L].notify_issue_change) |
| self.assertEqual('b@example.com', user_dict[222L].email) |
| |
| def testFetchItems(self): |
| SetUpGetUsers(self.user_service, self.cnxn) |
| self.mox.ReplayAll() |
| user_dict = self.user_service.user_2lc.FetchItems(self.cnxn, [333L]) |
| self.mox.VerifyAll() |
| self.assertEqual([333L], user_dict.keys()) |
| self.assertEqual('c@example.com', user_dict[333L].email) |
| self.assertFalse(user_dict[333L].is_site_admin) |
| self.assertEqual('Spammer', user_dict[333L].banned) |
| |
| |
| class UserServiceTest(unittest.TestCase): |
| |
| def setUp(self): |
| self.testbed = testbed.Testbed() |
| self.testbed.activate() |
| self.testbed.init_memcache_stub() |
| |
| self.mox = mox.Mox() |
| self.cnxn = fake.MonorailConnection() |
| self.cache_manager = fake.CacheManager() |
| self.user_service = MakeUserService(self.cache_manager, self.mox) |
| |
| def tearDown(self): |
| self.testbed.deactivate() |
| self.mox.UnsetStubs() |
| self.mox.ResetAll() |
| |
| def SetUpCreateUsers(self): |
| self.user_service.user_tbl.InsertRows( |
| self.cnxn, |
| ['user_id', 'email', 'obscure_email'], |
| [(3035911623, 'a@example.com', True), |
| (2996997680, 'b@example.com', True)] |
| ).AndReturn(None) |
| |
| def testCreateUsers(self): |
| self.SetUpCreateUsers() |
| self.mox.ReplayAll() |
| self.user_service._CreateUsers( |
| self.cnxn, ['a@example.com', 'b@example.com']) |
| self.mox.VerifyAll() |
| |
| def SetUpLookupUserEmails(self): |
| self.user_service.user_tbl.Select( |
| self.cnxn, cols=['user_id', 'email'], user_id=[222L]).AndReturn( |
| [(222L, 'b@example.com')]) |
| |
| def testLookupUserEmails(self): |
| self.SetUpLookupUserEmails() |
| self.user_service.email_cache.CacheItem( |
| 111L, 'a@example.com') |
| self.mox.ReplayAll() |
| emails_dict = self.user_service.LookupUserEmails( |
| self.cnxn, [111L, 222L]) |
| self.mox.VerifyAll() |
| self.assertEqual( |
| {111L: 'a@example.com', 222L: 'b@example.com'}, |
| emails_dict) |
| |
| def testLookupUserEmail(self): |
| self.SetUpLookupUserEmails() # Same as testLookupUserEmails() |
| self.mox.ReplayAll() |
| email_addr = self.user_service.LookupUserEmail(self.cnxn, 222L) |
| self.mox.VerifyAll() |
| self.assertEqual('b@example.com', email_addr) |
| |
| def SetUpLookupUserIDs(self): |
| self.user_service.user_tbl.Select( |
| self.cnxn, cols=['email', 'user_id'], |
| email=['b@example.com']).AndReturn([('b@example.com', 222L)]) |
| |
| def testLookupUserIDs(self): |
| self.SetUpLookupUserIDs() |
| self.user_service.user_id_cache.CacheItem( |
| 'a@example.com', 111L) |
| self.mox.ReplayAll() |
| user_id_dict = self.user_service.LookupUserIDs( |
| self.cnxn, ['a@example.com', 'b@example.com']) |
| self.mox.VerifyAll() |
| self.assertEqual( |
| {'a@example.com': 111L, 'b@example.com': 222L}, |
| user_id_dict) |
| |
| def testLookupUserIDs_InvalidEmail(self): |
| self.user_service.user_tbl.Select( |
| self.cnxn, cols=['email', 'user_id'], email=['abc']).AndReturn([]) |
| self.mox.ReplayAll() |
| user_id_dict = self.user_service.LookupUserIDs( |
| self.cnxn, ['abc'], autocreate=True) |
| self.mox.VerifyAll() |
| self.assertEqual({}, user_id_dict) |
| |
| def testLookupUserID(self): |
| self.SetUpLookupUserIDs() # Same as testLookupUserIDs() |
| self.user_service.user_id_cache.CacheItem('a@example.com', 111L) |
| self.mox.ReplayAll() |
| user_id = self.user_service.LookupUserID(self.cnxn, 'b@example.com') |
| self.mox.VerifyAll() |
| self.assertEqual(222, user_id) |
| |
| def testGetUsersByIDs(self): |
| SetUpGetUsers(self.user_service, self.cnxn) |
| user_a = user_pb2.User(email='a@example.com') |
| self.user_service.user_2lc.CacheItem(111L, user_a) |
| self.mox.ReplayAll() |
| user_dict = self.user_service.GetUsersByIDs( |
| self.cnxn, [111L, 333L]) |
| self.mox.VerifyAll() |
| self.assertEqual(2, len(user_dict)) |
| self.assertEqual('a@example.com', user_dict[111L].email) |
| self.assertFalse(user_dict[111L].is_site_admin) |
| self.assertFalse(user_dict[111L].banned) |
| self.assertTrue(user_dict[111L].notify_issue_change) |
| self.assertEqual('c@example.com', user_dict[333L].email) |
| |
| def testGetUser(self): |
| SetUpGetUsers(self.user_service, self.cnxn) |
| user_a = user_pb2.User(email='a@example.com') |
| self.user_service.user_2lc.CacheItem(111L, user_a) |
| self.mox.ReplayAll() |
| user = self.user_service.GetUser(self.cnxn, 333L) |
| self.mox.VerifyAll() |
| self.assertEqual('c@example.com', user.email) |
| |
| def SetUpUpdateUser(self): |
| delta = { |
| 'keep_people_perms_open': False, |
| 'preview_on_hover': True, |
| 'ignore_action_limits': False, |
| 'notify_issue_change': True, |
| 'after_issue_update': 'STAY_SAME_ISSUE', |
| 'notify_starred_issue_change': True, |
| 'is_site_admin': False, |
| 'banned': 'Turned spammer', |
| 'obscure_email': True, |
| } |
| self.user_service.user_tbl.Update( |
| self.cnxn, delta, user_id=111L, commit=False) |
| |
| self.user_service.actionlimit_tbl.Delete( |
| self.cnxn, user_id=111L, commit=False) |
| self.user_service.actionlimit_tbl.InsertRows( |
| self.cnxn, user_svc.ACTIONLIMIT_COLS, [], commit=False) |
| |
| self.user_service.dismissedcues_tbl.Delete( |
| self.cnxn, user_id=111L, commit=False) |
| self.user_service.dismissedcues_tbl.InsertRows( |
| self.cnxn, user_svc.DISMISSEDCUES_COLS, [], commit=False) |
| |
| def testUpdateUser(self): |
| self.SetUpUpdateUser() |
| user_a = user_pb2.User( |
| email='a@example.com', banned='Turned spammer') |
| self.mox.ReplayAll() |
| self.user_service.UpdateUser(self.cnxn, 111L, user_a) |
| self.mox.VerifyAll() |
| self.assertFalse(self.user_service.user_2lc.HasItem(111L)) |
| |
| def testUpdateUserSettings(self): |
| self.SetUpUpdateUser() |
| user_a = user_pb2.User(email='a@example.com') |
| self.mox.ReplayAll() |
| self.user_service.UpdateUserSettings( |
| self.cnxn, 111L, user_a, is_banned=True, |
| banned_reason='Turned spammer') |
| self.mox.VerifyAll() |
| |
| |
| class UserServiceFunctionsTest(unittest.TestCase): |
| |
| def testActionLimitToRow(self): |
| al = user_pb2.ActionLimit( |
| recent_count=1, reset_timestamp=123456, lifetime_count=9, |
| lifetime_limit=10, period_soft_limit=2, period_hard_limit=5) |
| action_kind = 3 |
| row = user_svc._ActionLimitToRow( |
| 111, action_kind, al) |
| self.assertEqual((111, action_kind, 1, 123456, 9, 10, 2, 5), row) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
