| # Copyright 2016 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 or at |
| # https://developers.google.com/open-source/licenses/bsd |
| |
| """Tests for the user service.""" |
| from __future__ import print_function |
| from __future__ import division |
| from __future__ import absolute_import |
| |
| import unittest |
| |
| import mock |
| try: |
| from mox3 import mox |
| except ImportError: |
| import mox |
| import time |
| |
| from google.appengine.ext import testbed |
| |
| from framework import exceptions |
| from framework import framework_constants |
| 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=[333]).AndReturn( |
| [(333, 'c@example.com', False, False, False, False, True, |
| False, 'Spammer', |
| 'stay_same_issue', False, False, True, 0, 0, None)]) |
| user_service.linkedaccount_tbl.Select( |
| cnxn, cols=user_svc.LINKEDACCOUNT_COLS, parent_id=[333], child_id=[333], |
| or_where_conds=True).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.hotlistvisithistory_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| user_service.linkedaccount_tbl = my_mox.CreateMock(sql.SQLTableManager) |
| # Account linking invites are done with patch(). |
| 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 tearDown(self): |
| self.testbed.deactivate() |
| |
| def testDeserializeUsersByID(self): |
| user_rows = [ |
| (111, 'a@example.com', False, False, False, False, True, False, '', |
| 'stay_same_issue', False, False, True, 0, 0, None), |
| (222, 'b@example.com', False, False, False, False, True, False, '', |
| 'next_in_list', False, False, True, 0, 0, None), |
| ] |
| linkedaccount_rows = [] |
| user_dict = self.user_service.user_2lc._DeserializeUsersByID( |
| user_rows, linkedaccount_rows) |
| self.assertEqual(2, len(user_dict)) |
| self.assertEqual('a@example.com', user_dict[111].email) |
| self.assertFalse(user_dict[111].is_site_admin) |
| self.assertEqual('', user_dict[111].banned) |
| self.assertFalse(user_dict[111].notify_issue_change) |
| self.assertEqual('b@example.com', user_dict[222].email) |
| self.assertIsNone(user_dict[111].linked_parent_id) |
| self.assertEqual([], user_dict[111].linked_child_ids) |
| self.assertIsNone(user_dict[222].linked_parent_id) |
| self.assertEqual([], user_dict[222].linked_child_ids) |
| |
| def testDeserializeUsersByID_LinkedAccounts(self): |
| user_rows = [ |
| (111, 'a@example.com', False, False, False, False, True, False, '', |
| 'stay_same_issue', False, False, True, 0, 0, None), |
| ] |
| linkedaccount_rows = [(111, 222), (111, 333), (444, 111)] |
| user_dict = self.user_service.user_2lc._DeserializeUsersByID( |
| user_rows, linkedaccount_rows) |
| self.assertEqual(1, len(user_dict)) |
| user_pb = user_dict[111] |
| self.assertEqual('a@example.com', user_pb.email) |
| self.assertEqual(444, user_pb.linked_parent_id) |
| self.assertEqual([222, 333], user_pb.linked_child_ids) |
| |
| def testFetchItems(self): |
| SetUpGetUsers(self.user_service, self.cnxn) |
| self.mox.ReplayAll() |
| user_dict = self.user_service.user_2lc.FetchItems(self.cnxn, [333]) |
| self.mox.VerifyAll() |
| self.assertEqual([333], list(user_dict.keys())) |
| self.assertEqual('c@example.com', user_dict[333].email) |
| self.assertFalse(user_dict[333].is_site_admin) |
| self.assertEqual('Spammer', user_dict[333].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=[222]).AndReturn( |
| [(222, 'b@example.com')]) |
| |
| def testLookupUserEmails(self): |
| self.SetUpLookupUserEmails() |
| self.user_service.email_cache.CacheItem( |
| 111, 'a@example.com') |
| self.mox.ReplayAll() |
| emails_dict = self.user_service.LookupUserEmails( |
| self.cnxn, [111, 222]) |
| self.mox.VerifyAll() |
| self.assertEqual( |
| {111: 'a@example.com', 222: 'b@example.com'}, |
| emails_dict) |
| |
| def SetUpLookupUserEmails_Missed(self): |
| self.user_service.user_tbl.Select( |
| self.cnxn, cols=['user_id', 'email'], user_id=[222]).AndReturn([]) |
| self.user_service.email_cache.CacheItem( |
| 111, 'a@example.com') |
| |
| def testLookupUserEmails_Missed(self): |
| self.SetUpLookupUserEmails_Missed() |
| self.mox.ReplayAll() |
| with self.assertRaises(exceptions.NoSuchUserException): |
| self.user_service.LookupUserEmails(self.cnxn, [111, 222]) |
| self.mox.VerifyAll() |
| |
| def testLookUpUserEmails_IgnoreMissed(self): |
| self.SetUpLookupUserEmails_Missed() |
| self.mox.ReplayAll() |
| emails_dict = self.user_service.LookupUserEmails( |
| self.cnxn, [111, 222], ignore_missed=True) |
| self.mox.VerifyAll() |
| self.assertEqual({111: 'a@example.com'}, emails_dict) |
| |
| def testLookupUserEmail(self): |
| self.SetUpLookupUserEmails() # Same as testLookupUserEmails() |
| self.mox.ReplayAll() |
| email_addr = self.user_service.LookupUserEmail(self.cnxn, 222) |
| 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', 222)]) |
| |
| def testLookupUserIDs(self): |
| self.SetUpLookupUserIDs() |
| self.user_service.user_id_cache.CacheItem( |
| 'a@example.com', 111) |
| 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': 111, 'b@example.com': 222}, |
| 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 testLookupUserIDs_NoUserValue(self): |
| self.user_service.user_tbl.Select = mock.Mock( |
| return_value=[('b@example.com', 222)]) |
| user_id_dict = self.user_service.LookupUserIDs( |
| self.cnxn, [framework_constants.NO_VALUES, '', 'b@example.com']) |
| self.assertEqual({'b@example.com': 222}, user_id_dict) |
| self.user_service.user_tbl.Select.assert_called_once_with( |
| self.cnxn, cols=['email', 'user_id'], email=['b@example.com']) |
| |
| def testLookupUserID(self): |
| self.SetUpLookupUserIDs() # Same as testLookupUserIDs() |
| self.user_service.user_id_cache.CacheItem('a@example.com', 111) |
| self.mox.ReplayAll() |
| user_id = self.user_service.LookupUserID(self.cnxn, 'b@example.com') |
| self.mox.VerifyAll() |
| self.assertEqual(222, user_id) |
| |
| def SetUpGetUsersByIDs(self): |
| self.user_service.user_tbl.Select( |
| self.cnxn, cols=user_svc.USER_COLS, user_id=[333, 444]).AndReturn( |
| [ |
| ( |
| 333, 'c@example.com', False, False, False, False, True, |
| False, 'Spammer', 'stay_same_issue', False, False, True, 0, |
| 0, None) |
| ]) |
| self.user_service.linkedaccount_tbl.Select( |
| self.cnxn, |
| cols=user_svc.LINKEDACCOUNT_COLS, |
| parent_id=[333, 444], |
| child_id=[333, 444], |
| or_where_conds=True).AndReturn([]) |
| |
| |
| def testGetUsersByIDs(self): |
| self.SetUpGetUsersByIDs() |
| user_a = user_pb2.User(email='a@example.com') |
| self.user_service.user_2lc.CacheItem(111, user_a) |
| self.mox.ReplayAll() |
| # 444 user does not exist. |
| user_dict = self.user_service.GetUsersByIDs(self.cnxn, [111, 333, 444]) |
| self.mox.VerifyAll() |
| self.assertEqual(3, len(user_dict)) |
| self.assertEqual('a@example.com', user_dict[111].email) |
| self.assertFalse(user_dict[111].is_site_admin) |
| self.assertFalse(user_dict[111].banned) |
| self.assertTrue(user_dict[111].notify_issue_change) |
| self.assertEqual('c@example.com', user_dict[333].email) |
| self.assertEqual(user_dict[444], user_pb2.MakeUser(444)) |
| |
| def testGetUsersByIDs_SkipMissed(self): |
| self.SetUpGetUsersByIDs() |
| user_a = user_pb2.User(email='a@example.com') |
| self.user_service.user_2lc.CacheItem(111, user_a) |
| self.mox.ReplayAll() |
| # 444 user does not exist |
| user_dict = self.user_service.GetUsersByIDs( |
| self.cnxn, [111, 333, 444], skip_missed=True) |
| self.mox.VerifyAll() |
| self.assertEqual(2, len(user_dict)) |
| self.assertEqual('a@example.com', user_dict[111].email) |
| self.assertFalse(user_dict[111].is_site_admin) |
| self.assertFalse(user_dict[111].banned) |
| self.assertTrue(user_dict[111].notify_issue_change) |
| self.assertEqual('c@example.com', user_dict[333].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(111, user_a) |
| self.mox.ReplayAll() |
| user = self.user_service.GetUser(self.cnxn, 333) |
| self.mox.VerifyAll() |
| self.assertEqual('c@example.com', user.email) |
| |
| def SetUpUpdateUser(self): |
| delta = { |
| 'keep_people_perms_open': False, |
| 'preview_on_hover': True, |
| 'notify_issue_change': True, |
| 'after_issue_update': 'STAY_SAME_ISSUE', |
| 'notify_starred_issue_change': True, |
| 'notify_starred_ping': False, |
| 'is_site_admin': False, |
| 'banned': 'Turned spammer', |
| 'obscure_email': True, |
| 'email_compact_subject': False, |
| 'email_view_widget': True, |
| 'last_visit_timestamp': 0, |
| 'email_bounce_timestamp': 0, |
| 'vacation_message': None, |
| } |
| self.user_service.user_tbl.Update( |
| self.cnxn, delta, user_id=111, 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, 111, user_a) |
| self.mox.VerifyAll() |
| self.assertFalse(self.user_service.user_2lc.HasItem(111)) |
| |
| def SetUpGetRecentlyVisitedHotlists(self): |
| self.user_service.hotlistvisithistory_tbl.Select( |
| self.cnxn, cols=['hotlist_id'], user_id=[111], |
| order_by=[('viewed DESC', [])], limit=10).AndReturn( |
| ((123,), (234,))) |
| |
| def testGetRecentlyVisitedHotlists(self): |
| self.SetUpGetRecentlyVisitedHotlists() |
| self.mox.ReplayAll() |
| recent_hotlist_rows = self.user_service.GetRecentlyVisitedHotlists( |
| self.cnxn, 111) |
| self.mox.VerifyAll() |
| self.assertEqual(recent_hotlist_rows, [123, 234]) |
| |
| def SetUpAddVisitedHotlist(self, ts): |
| self.user_service.hotlistvisithistory_tbl.Delete( |
| self.cnxn, hotlist_id=123, user_id=111, commit=False) |
| self.user_service.hotlistvisithistory_tbl.InsertRows( |
| self.cnxn, user_svc.HOTLISTVISITHISTORY_COLS, |
| [(123, 111, ts)], |
| commit=False) |
| |
| @mock.patch('time.time') |
| def testAddVisitedHotlist(self, mockTime): |
| ts = 122333 |
| mockTime.return_value = ts |
| self.SetUpAddVisitedHotlist(ts) |
| self.mox.ReplayAll() |
| self.user_service.AddVisitedHotlist(self.cnxn, 111, 123, commit=False) |
| self.mox.VerifyAll() |
| |
| def testExpungeHotlistsFromHistory(self): |
| self.user_service.hotlistvisithistory_tbl.Delete = mock.Mock() |
| hotlist_ids = [123, 223] |
| self.user_service.ExpungeHotlistsFromHistory( |
| self.cnxn, hotlist_ids, commit=False) |
| self.user_service.hotlistvisithistory_tbl.Delete.assert_called_once_with( |
| self.cnxn, hotlist_id=hotlist_ids, commit=False) |
| |
| def testExpungeUsersHotlistsHistory(self): |
| self.user_service.hotlistvisithistory_tbl.Delete = mock.Mock() |
| user_ids = [111, 222] |
| self.user_service.ExpungeUsersHotlistsHistory( |
| self.cnxn, user_ids, commit=False) |
| self.user_service.hotlistvisithistory_tbl.Delete.assert_called_once_with( |
| self.cnxn, user_id=user_ids, commit=False) |
| |
| def SetUpTrimUserVisitedHotlists(self, user_ids, ts): |
| self.user_service.hotlistvisithistory_tbl.Select( |
| self.cnxn, cols=['user_id'], group_by=['user_id'], |
| having=[('COUNT(*) > %s', [10])], limit=1000).AndReturn(( |
| (111,), (222,), (333,))) |
| for user_id in user_ids: |
| self.user_service.hotlistvisithistory_tbl.Select( |
| self.cnxn, cols=['viewed'], user_id=user_id, |
| order_by=[('viewed DESC', [])]).AndReturn([ |
| (ts,), (ts,), (ts,), (ts,), (ts,), (ts,), |
| (ts,), (ts,), (ts,), (ts,), (ts+1,)]) |
| self.user_service.hotlistvisithistory_tbl.Delete( |
| self.cnxn, user_id=user_id, where=[('viewed < %s', [ts])], |
| commit=False) |
| |
| @mock.patch('time.time') |
| def testTrimUserVisitedHotlists(self, mockTime): |
| ts = 122333 |
| mockTime.return_value = ts |
| self.SetUpTrimUserVisitedHotlists([111, 222, 333], ts) |
| self.mox.ReplayAll() |
| self.user_service.TrimUserVisitedHotlists(self.cnxn, commit=False) |
| self.mox.VerifyAll() |
| |
| def testGetPendingLinkedInvites_Anon(self): |
| """An Anon user never has invites to link accounts.""" |
| as_parent, as_child = self.user_service.GetPendingLinkedInvites( |
| self.cnxn, 0) |
| self.assertEqual([], as_parent) |
| self.assertEqual([], as_child) |
| |
| def testGetPendingLinkedInvites_None(self): |
| """A user who has no link invites gets empty lists.""" |
| self.user_service.linkedaccountinvite_tbl = mock.Mock() |
| self.user_service.linkedaccountinvite_tbl.Select.return_value = [] |
| as_parent, as_child = self.user_service.GetPendingLinkedInvites( |
| self.cnxn, 111) |
| self.assertEqual([], as_parent) |
| self.assertEqual([], as_child) |
| |
| def testGetPendingLinkedInvites_Some(self): |
| """A user who has link invites can get them.""" |
| self.user_service.linkedaccountinvite_tbl = mock.Mock() |
| self.user_service.linkedaccountinvite_tbl.Select.return_value = [ |
| (111, 222), (111, 333), (888, 999), (333, 111)] |
| as_parent, as_child = self.user_service.GetPendingLinkedInvites( |
| self.cnxn, 111) |
| self.assertEqual([222, 333], as_parent) |
| self.assertEqual([333], as_child) |
| |
| def testAssertNotAlreadyLinked_NotLinked(self): |
| """No exception is raised when accounts are not already linked.""" |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.linkedaccount_tbl.Select.return_value = [] |
| self.user_service._AssertNotAlreadyLinked(self.cnxn, 111, 222) |
| |
| def testAssertNotAlreadyLinked_AlreadyLinked(self): |
| """Reject attempt to link any account that is already linked.""" |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.linkedaccount_tbl.Select.return_value = [ |
| (111, 222)] |
| with self.assertRaises(exceptions.InputException): |
| self.user_service._AssertNotAlreadyLinked(self.cnxn, 111, 333) |
| |
| def testInviteLinkedParent_Anon(self): |
| """Anon cannot invite anyone to link accounts.""" |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.InviteLinkedParent(self.cnxn, 0, 0) |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.InviteLinkedParent(self.cnxn, 111, 0) |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.InviteLinkedParent(self.cnxn, 0, 111) |
| |
| def testInviteLinkedParent_Normal(self): |
| """One account can invite another to link.""" |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.linkedaccount_tbl.Select.return_value = [] |
| self.user_service.linkedaccountinvite_tbl = mock.Mock() |
| self.user_service.InviteLinkedParent( |
| self.cnxn, 111, 222) |
| self.user_service.linkedaccountinvite_tbl.InsertRow.assert_called_once_with( |
| self.cnxn, parent_id=111, child_id=222) |
| |
| def testAcceptLinkedChild_Anon(self): |
| """Reject attempts for anon to accept any invite.""" |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.AcceptLinkedChild(self.cnxn, 0, 333) |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.AcceptLinkedChild(self.cnxn, 333, 0) |
| |
| def testAcceptLinkedChild_Missing(self): |
| """Reject attempts to link without a matching invite.""" |
| self.user_service.linkedaccountinvite_tbl = mock.Mock() |
| self.user_service.linkedaccountinvite_tbl.Select.return_value = [] |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.linkedaccount_tbl.Select.return_value = [] |
| with self.assertRaises(exceptions.InputException) as cm: |
| self.user_service.AcceptLinkedChild(self.cnxn, 111, 333) |
| self.assertEqual('No such invite', cm.exception.message) |
| |
| def testAcceptLinkedChild_Normal(self): |
| """Create linkage between accounts and remove invite.""" |
| self.user_service.linkedaccountinvite_tbl = mock.Mock() |
| self.user_service.linkedaccountinvite_tbl.Select.return_value = [ |
| (111, 222), (333, 444)] |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.linkedaccount_tbl.Select.return_value = [] |
| |
| self.user_service.AcceptLinkedChild(self.cnxn, 111, 222) |
| self.user_service.linkedaccount_tbl.InsertRow.assert_called_once_with( |
| self.cnxn, parent_id=111, child_id=222) |
| self.user_service.linkedaccountinvite_tbl.Delete.assert_called_once_with( |
| self.cnxn, parent_id=111, child_id=222) |
| |
| def testUnlinkAccounts_MissingIDs(self): |
| """Reject an attempt to unlink anon.""" |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.UnlinkAccounts(self.cnxn, 0, 0) |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.UnlinkAccounts(self.cnxn, 0, 111) |
| with self.assertRaises(exceptions.InputException): |
| self.user_service.UnlinkAccounts(self.cnxn, 111, 0) |
| |
| def testUnlinkAccounts_Normal(self): |
| """We can unlink accounts.""" |
| self.user_service.linkedaccount_tbl = mock.Mock() |
| self.user_service.UnlinkAccounts(self.cnxn, 111, 222) |
| self.user_service.linkedaccount_tbl.Delete.assert_called_once_with( |
| self.cnxn, parent_id=111, child_id=222) |
| |
| def testUpdateUserSettings(self): |
| self.SetUpUpdateUser() |
| user_a = user_pb2.User(email='a@example.com') |
| self.mox.ReplayAll() |
| self.user_service.UpdateUserSettings( |
| self.cnxn, 111, user_a, is_banned=True, |
| banned_reason='Turned spammer') |
| self.mox.VerifyAll() |
| |
| def testGetUsersPrefs(self): |
| self.user_service.userprefs_tbl = mock.Mock() |
| self.user_service.userprefs_tbl.Select.return_value = [ |
| (111, 'code_font', 'true'), |
| (111, 'keep_perms_open', 'true'), |
| # Note: user 222 has not set any prefs. |
| (333, 'code_font', 'false')] |
| |
| prefs_dict = self.user_service.GetUsersPrefs(self.cnxn, [111, 222, 333]) |
| |
| expected = { |
| 111: user_pb2.UserPrefs( |
| user_id=111, |
| prefs=[user_pb2.UserPrefValue(name='code_font', value='true'), |
| user_pb2.UserPrefValue(name='keep_perms_open', value='true')]), |
| 222: user_pb2.UserPrefs(user_id=222), |
| 333: user_pb2.UserPrefs( |
| user_id=333, |
| prefs=[user_pb2.UserPrefValue(name='code_font', value='false')]), |
| } |
| self.assertEqual(expected, prefs_dict) |
| |
| def testGetUserPrefs(self): |
| self.user_service.userprefs_tbl = mock.Mock() |
| self.user_service.userprefs_tbl.Select.return_value = [ |
| (111, 'code_font', 'true'), |
| (111, 'keep_perms_open', 'true'), |
| # Note: user 222 has not set any prefs. |
| (333, 'code_font', 'false')] |
| |
| userprefs = self.user_service.GetUserPrefs(self.cnxn, 111) |
| expected = user_pb2.UserPrefs( |
| user_id=111, |
| prefs=[user_pb2.UserPrefValue(name='code_font', value='true'), |
| user_pb2.UserPrefValue(name='keep_perms_open', value='true')]) |
| self.assertEqual(expected, userprefs) |
| |
| userprefs = self.user_service.GetUserPrefs(self.cnxn, 222) |
| expected = user_pb2.UserPrefs(user_id=222) |
| self.assertEqual(expected, userprefs) |
| |
| def testSetUserPrefs(self): |
| self.user_service.userprefs_tbl = mock.Mock() |
| pref_values = [user_pb2.UserPrefValue(name='code_font', value='true'), |
| user_pb2.UserPrefValue(name='keep_perms_open', value='true')] |
| self.user_service.SetUserPrefs(self.cnxn, 111, pref_values) |
| self.user_service.userprefs_tbl.InsertRows.assert_called_once_with( |
| self.cnxn, user_svc.USERPREFS_COLS, |
| [(111, 'code_font', 'true'), |
| (111, 'keep_perms_open', 'true')], |
| replace=True) |
| |
| def testExpungeUsers(self): |
| self.user_service.linkedaccount_tbl.Delete = mock.Mock() |
| self.user_service.linkedaccountinvite_tbl.Delete = mock.Mock() |
| self.user_service.userprefs_tbl.Delete = mock.Mock() |
| self.user_service.user_tbl.Delete = mock.Mock() |
| |
| user_ids = [222, 444] |
| self.user_service.ExpungeUsers(self.cnxn, user_ids) |
| |
| linked_account_calls = [ |
| mock.call(self.cnxn, parent_id=user_ids, commit=False), |
| mock.call(self.cnxn, child_id=user_ids, commit=False)] |
| self.user_service.linkedaccount_tbl.Delete.has_calls(linked_account_calls) |
| self.user_service.linkedaccountinvite_tbl.Delete.has_calls( |
| linked_account_calls) |
| user_calls = [mock.call(self.cnxn, user_id=user_ids, commit=False)] |
| self.user_service.userprefs_tbl.Delete.has_calls(user_calls) |
| self.user_service.user_tbl.Delete.has_calls(user_calls) |
| |
| def testTotalUsersCount(self): |
| self.user_service.user_tbl.SelectValue = mock.Mock(return_value=10) |
| self.assertEqual(self.user_service.TotalUsersCount(self.cnxn), 9) |
| self.user_service.user_tbl.SelectValue.assert_called_once_with( |
| self.cnxn, col='COUNT(*)') |
| |
| def testGetAllUserEmailsBatch(self): |
| rows = [('cow@test.com',), ('pig@test.com',), ('fox@test.com',)] |
| self.user_service.user_tbl.Select = mock.Mock(return_value=rows) |
| emails = self.user_service.GetAllUserEmailsBatch(self.cnxn) |
| self.user_service.user_tbl.Select.assert_called_once_with( |
| self.cnxn, cols=['email'], limit=1000, offset=0, |
| where=[('user_id != %s', [framework_constants.DELETED_USER_ID])], |
| order_by=[('user_id ASC', [])]) |
| self.assertItemsEqual( |
| emails, ['cow@test.com', 'pig@test.com', 'fox@test.com']) |
| |
| def testGetAllUserEmailsBatch_CustomLimit(self): |
| rows = [('cow@test.com',), ('pig@test.com',), ('fox@test.com',)] |
| self.user_service.user_tbl.Select = mock.Mock(return_value=rows) |
| emails = self.user_service.GetAllUserEmailsBatch( |
| self.cnxn, limit=30, offset=60) |
| self.user_service.user_tbl.Select.assert_called_once_with( |
| self.cnxn, cols=['email'], limit=30, offset=60, |
| where=[('user_id != %s', [framework_constants.DELETED_USER_ID])], |
| order_by=[('user_id ASC', [])]) |
| self.assertItemsEqual( |
| emails, ['cow@test.com', 'pig@test.com', 'fox@test.com']) |