OLD | NEW |
(Empty) | |
| 1 # Copyright 2016 The Chromium Authors. All rights reserved. |
| 2 # Use of this source code is govered by a BSD-style |
| 3 # license that can be found in the LICENSE file or at |
| 4 # https://developers.google.com/open-source/licenses/bsd |
| 5 |
| 6 """Tests for the user service.""" |
| 7 |
| 8 import unittest |
| 9 |
| 10 import mox |
| 11 |
| 12 from google.appengine.ext import testbed |
| 13 |
| 14 from framework import sql |
| 15 from proto import user_pb2 |
| 16 from services import user_svc |
| 17 from testing import fake |
| 18 |
| 19 |
| 20 def SetUpGetUsers(user_service, cnxn): |
| 21 """Set up expected calls to SQL tables.""" |
| 22 user_service.user_tbl.Select( |
| 23 cnxn, cols=user_svc.USER_COLS, user_id=[333L]).AndReturn( |
| 24 [(333L, 'c@example.com', False, False, False, 'Spammer', |
| 25 'stay_same_issue', False, False, False, True)]) |
| 26 user_service.actionlimit_tbl.Select( |
| 27 cnxn, cols=user_svc.ACTIONLIMIT_COLS, user_id=[333L]).AndReturn([]) |
| 28 user_service.dismissedcues_tbl.Select( |
| 29 cnxn, cols=user_svc.DISMISSEDCUES_COLS, user_id=[333L]).AndReturn([]) |
| 30 |
| 31 |
| 32 def MakeUserService(cache_manager, my_mox): |
| 33 user_service = user_svc.UserService(cache_manager) |
| 34 user_service.user_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| 35 user_service.actionlimit_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| 36 user_service.dismissedcues_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| 37 return user_service |
| 38 |
| 39 |
| 40 class UserTwoLevelCacheTest(unittest.TestCase): |
| 41 |
| 42 def setUp(self): |
| 43 self.testbed = testbed.Testbed() |
| 44 self.testbed.activate() |
| 45 self.testbed.init_memcache_stub() |
| 46 |
| 47 self.mox = mox.Mox() |
| 48 self.cnxn = fake.MonorailConnection() |
| 49 self.cache_manager = fake.CacheManager() |
| 50 self.user_service = MakeUserService(self.cache_manager, self.mox) |
| 51 |
| 52 def testDeserializeUsersByID(self): |
| 53 user_rows = [ |
| 54 (111L, 'a@example.com', False, False, False, '', |
| 55 'stay_same_issue', False, False, False, True), |
| 56 (222L, 'b@example.com', False, False, False, '', |
| 57 'next_in_list', False, False, False, True), |
| 58 ] |
| 59 actionlimit_rows = [] |
| 60 dismissedcues_rows = [] |
| 61 user_dict = self.user_service.user_2lc._DeserializeUsersByID( |
| 62 user_rows, actionlimit_rows, dismissedcues_rows) |
| 63 self.assertEqual(2, len(user_dict)) |
| 64 self.assertEqual('a@example.com', user_dict[111L].email) |
| 65 self.assertFalse(user_dict[111L].is_site_admin) |
| 66 self.assertEqual('', user_dict[111L].banned) |
| 67 self.assertFalse(user_dict[111L].notify_issue_change) |
| 68 self.assertEqual('b@example.com', user_dict[222L].email) |
| 69 |
| 70 def testFetchItems(self): |
| 71 SetUpGetUsers(self.user_service, self.cnxn) |
| 72 self.mox.ReplayAll() |
| 73 user_dict = self.user_service.user_2lc.FetchItems(self.cnxn, [333L]) |
| 74 self.mox.VerifyAll() |
| 75 self.assertEqual([333L], user_dict.keys()) |
| 76 self.assertEqual('c@example.com', user_dict[333L].email) |
| 77 self.assertFalse(user_dict[333L].is_site_admin) |
| 78 self.assertEqual('Spammer', user_dict[333L].banned) |
| 79 |
| 80 |
| 81 class UserServiceTest(unittest.TestCase): |
| 82 |
| 83 def setUp(self): |
| 84 self.testbed = testbed.Testbed() |
| 85 self.testbed.activate() |
| 86 self.testbed.init_memcache_stub() |
| 87 |
| 88 self.mox = mox.Mox() |
| 89 self.cnxn = fake.MonorailConnection() |
| 90 self.cache_manager = fake.CacheManager() |
| 91 self.user_service = MakeUserService(self.cache_manager, self.mox) |
| 92 |
| 93 def tearDown(self): |
| 94 self.testbed.deactivate() |
| 95 self.mox.UnsetStubs() |
| 96 self.mox.ResetAll() |
| 97 |
| 98 def SetUpCreateUsers(self): |
| 99 self.user_service.user_tbl.InsertRows( |
| 100 self.cnxn, |
| 101 ['user_id', 'email', 'obscure_email'], |
| 102 [(3035911623, 'a@example.com', True), |
| 103 (2996997680, 'b@example.com', True)] |
| 104 ).AndReturn(None) |
| 105 |
| 106 def testCreateUsers(self): |
| 107 self.SetUpCreateUsers() |
| 108 self.mox.ReplayAll() |
| 109 self.user_service._CreateUsers( |
| 110 self.cnxn, ['a@example.com', 'b@example.com']) |
| 111 self.mox.VerifyAll() |
| 112 |
| 113 def SetUpLookupUserEmails(self): |
| 114 self.user_service.user_tbl.Select( |
| 115 self.cnxn, cols=['user_id', 'email'], user_id=[222L]).AndReturn( |
| 116 [(222L, 'b@example.com')]) |
| 117 |
| 118 def testLookupUserEmails(self): |
| 119 self.SetUpLookupUserEmails() |
| 120 self.user_service.email_cache.CacheItem( |
| 121 111L, 'a@example.com') |
| 122 self.mox.ReplayAll() |
| 123 emails_dict = self.user_service.LookupUserEmails( |
| 124 self.cnxn, [111L, 222L]) |
| 125 self.mox.VerifyAll() |
| 126 self.assertEqual( |
| 127 {111L: 'a@example.com', 222L: 'b@example.com'}, |
| 128 emails_dict) |
| 129 |
| 130 def testLookupUserEmail(self): |
| 131 self.SetUpLookupUserEmails() # Same as testLookupUserEmails() |
| 132 self.mox.ReplayAll() |
| 133 email_addr = self.user_service.LookupUserEmail(self.cnxn, 222L) |
| 134 self.mox.VerifyAll() |
| 135 self.assertEqual('b@example.com', email_addr) |
| 136 |
| 137 def SetUpLookupUserIDs(self): |
| 138 self.user_service.user_tbl.Select( |
| 139 self.cnxn, cols=['email', 'user_id'], |
| 140 email=['b@example.com']).AndReturn([('b@example.com', 222L)]) |
| 141 |
| 142 def testLookupUserIDs(self): |
| 143 self.SetUpLookupUserIDs() |
| 144 self.user_service.user_id_cache.CacheItem( |
| 145 'a@example.com', 111L) |
| 146 self.mox.ReplayAll() |
| 147 user_id_dict = self.user_service.LookupUserIDs( |
| 148 self.cnxn, ['a@example.com', 'b@example.com']) |
| 149 self.mox.VerifyAll() |
| 150 self.assertEqual( |
| 151 {'a@example.com': 111L, 'b@example.com': 222L}, |
| 152 user_id_dict) |
| 153 |
| 154 def testLookupUserIDs_InvalidEmail(self): |
| 155 self.user_service.user_tbl.Select( |
| 156 self.cnxn, cols=['email', 'user_id'], email=['abc']).AndReturn([]) |
| 157 self.mox.ReplayAll() |
| 158 user_id_dict = self.user_service.LookupUserIDs( |
| 159 self.cnxn, ['abc'], autocreate=True) |
| 160 self.mox.VerifyAll() |
| 161 self.assertEqual({}, user_id_dict) |
| 162 |
| 163 def testLookupUserID(self): |
| 164 self.SetUpLookupUserIDs() # Same as testLookupUserIDs() |
| 165 self.user_service.user_id_cache.CacheItem('a@example.com', 111L) |
| 166 self.mox.ReplayAll() |
| 167 user_id = self.user_service.LookupUserID(self.cnxn, 'b@example.com') |
| 168 self.mox.VerifyAll() |
| 169 self.assertEqual(222, user_id) |
| 170 |
| 171 def testGetUsersByIDs(self): |
| 172 SetUpGetUsers(self.user_service, self.cnxn) |
| 173 user_a = user_pb2.User(email='a@example.com') |
| 174 self.user_service.user_2lc.CacheItem(111L, user_a) |
| 175 self.mox.ReplayAll() |
| 176 user_dict = self.user_service.GetUsersByIDs( |
| 177 self.cnxn, [111L, 333L]) |
| 178 self.mox.VerifyAll() |
| 179 self.assertEqual(2, len(user_dict)) |
| 180 self.assertEqual('a@example.com', user_dict[111L].email) |
| 181 self.assertFalse(user_dict[111L].is_site_admin) |
| 182 self.assertFalse(user_dict[111L].banned) |
| 183 self.assertTrue(user_dict[111L].notify_issue_change) |
| 184 self.assertEqual('c@example.com', user_dict[333L].email) |
| 185 |
| 186 def testGetUser(self): |
| 187 SetUpGetUsers(self.user_service, self.cnxn) |
| 188 user_a = user_pb2.User(email='a@example.com') |
| 189 self.user_service.user_2lc.CacheItem(111L, user_a) |
| 190 self.mox.ReplayAll() |
| 191 user = self.user_service.GetUser(self.cnxn, 333L) |
| 192 self.mox.VerifyAll() |
| 193 self.assertEqual('c@example.com', user.email) |
| 194 |
| 195 def SetUpUpdateUser(self): |
| 196 delta = { |
| 197 'keep_people_perms_open': False, |
| 198 'preview_on_hover': True, |
| 199 'ignore_action_limits': False, |
| 200 'notify_issue_change': True, |
| 201 'after_issue_update': 'STAY_SAME_ISSUE', |
| 202 'notify_starred_issue_change': True, |
| 203 'is_site_admin': False, |
| 204 'banned': 'Turned spammer', |
| 205 'obscure_email': True, |
| 206 } |
| 207 self.user_service.user_tbl.Update( |
| 208 self.cnxn, delta, user_id=111L, commit=False) |
| 209 |
| 210 self.user_service.actionlimit_tbl.Delete( |
| 211 self.cnxn, user_id=111L, commit=False) |
| 212 self.user_service.actionlimit_tbl.InsertRows( |
| 213 self.cnxn, user_svc.ACTIONLIMIT_COLS, [], commit=False) |
| 214 |
| 215 self.user_service.dismissedcues_tbl.Delete( |
| 216 self.cnxn, user_id=111L, commit=False) |
| 217 self.user_service.dismissedcues_tbl.InsertRows( |
| 218 self.cnxn, user_svc.DISMISSEDCUES_COLS, [], commit=False) |
| 219 |
| 220 def testUpdateUser(self): |
| 221 self.SetUpUpdateUser() |
| 222 user_a = user_pb2.User( |
| 223 email='a@example.com', banned='Turned spammer') |
| 224 self.mox.ReplayAll() |
| 225 self.user_service.UpdateUser(self.cnxn, 111L, user_a) |
| 226 self.mox.VerifyAll() |
| 227 self.assertFalse(self.user_service.user_2lc.HasItem(111L)) |
| 228 |
| 229 def testUpdateUserSettings(self): |
| 230 self.SetUpUpdateUser() |
| 231 user_a = user_pb2.User(email='a@example.com') |
| 232 self.mox.ReplayAll() |
| 233 self.user_service.UpdateUserSettings( |
| 234 self.cnxn, 111L, user_a, is_banned=True, |
| 235 banned_reason='Turned spammer') |
| 236 self.mox.VerifyAll() |
| 237 |
| 238 |
| 239 class UserServiceFunctionsTest(unittest.TestCase): |
| 240 |
| 241 def testActionLimitToRow(self): |
| 242 al = user_pb2.ActionLimit( |
| 243 recent_count=1, reset_timestamp=123456, lifetime_count=9, |
| 244 lifetime_limit=10, period_soft_limit=2, period_hard_limit=5) |
| 245 action_kind = 3 |
| 246 row = user_svc._ActionLimitToRow( |
| 247 111, action_kind, al) |
| 248 self.assertEqual((111, action_kind, 1, 123456, 9, 10, 2, 5), row) |
| 249 |
| 250 |
| 251 if __name__ == '__main__': |
| 252 unittest.main() |
OLD | NEW |