blob: 87ed5bca59d5799e8f24063f57fed4c367d868a6 [file] [log] [blame]
# Copyright 2019 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
"""Unit tests for deleteusers classes."""
from __future__ import print_function
from __future__ import division
from __future__ import absolute_import
import logging
import mock
import unittest
from six.moves import urllib
from framework import cloud_tasks_helpers
from framework import deleteusers
from framework import framework_constants
from framework import urls
from services import service_manager
from testing import fake
from testing import testing_helpers
class TestWipeoutSyncCron(unittest.TestCase):
def setUp(self):
self.services = service_manager.Services(user=fake.UserService())
self.task = deleteusers.WipeoutSyncCron(services=self.services)
self.user_1 = self.services.user.TestAddUser('user1@example.com', 111)
self.user_2 = self.services.user.TestAddUser('user2@example.com', 222)
self.user_3 = self.services.user.TestAddUser('user3@example.com', 333)
def generate_simple_task(self, url, body):
return {
'app_engine_http_request':
{
'relative_uri': url,
'body': body,
'headers': {
'Content-type': 'application/x-www-form-urlencoded'
}
}
}
@mock.patch('framework.cloud_tasks_helpers._get_client')
def testHandleRequest(self, get_client_mock):
mr = testing_helpers.MakeMonorailRequest(
path='url/url?batchsize=2',
services=self.services)
self.task.HandleRequest(mr)
self.assertEqual(get_client_mock().create_task.call_count, 3)
expected_task = self.generate_simple_task(
urls.SEND_WIPEOUT_USER_LISTS_TASK + '.do', 'limit=2&offset=0')
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
expected_task = self.generate_simple_task(
urls.SEND_WIPEOUT_USER_LISTS_TASK + '.do', 'limit=2&offset=2')
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
expected_task = self.generate_simple_task(
urls.DELETE_WIPEOUT_USERS_TASK + '.do', '')
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
@mock.patch('framework.cloud_tasks_helpers._get_client')
def testHandleRequest_NoBatchSizeParam(self, get_client_mock):
mr = testing_helpers.MakeMonorailRequest(services=self.services)
self.task.HandleRequest(mr)
expected_task = self.generate_simple_task(
urls.SEND_WIPEOUT_USER_LISTS_TASK + '.do',
'limit={}&offset=0'.format(deleteusers.MAX_BATCH_SIZE))
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
@mock.patch('framework.cloud_tasks_helpers._get_client')
def testHandleRequest_NoUsers(self, get_client_mock):
mr = testing_helpers.MakeMonorailRequest()
self.services.user.users_by_id = {}
self.task.HandleRequest(mr)
calls = get_client_mock().create_task.call_args_list
self.assertEqual(len(calls), 0)
class SendWipeoutUserListsTaskTest(unittest.TestCase):
def setUp(self):
self.services = service_manager.Services(user=fake.UserService())
self.task = deleteusers.SendWipeoutUserListsTask(services=self.services)
self.task.sendUserLists = mock.Mock()
deleteusers.authorize = mock.Mock(return_value='service')
self.user_1 = self.services.user.TestAddUser('user1@example.com', 111)
self.user_2 = self.services.user.TestAddUser('user2@example.com', 222)
self.user_3 = self.services.user.TestAddUser('user3@example.com', 333)
def testHandleRequest_NoBatchSizeParam(self):
mr = testing_helpers.MakeMonorailRequest(path='url/url?limit=2&offset=1')
self.task.HandleRequest(mr)
deleteusers.authorize.assert_called_once_with()
self.task.sendUserLists.assert_called_once_with(
'service', [
{'id': self.user_2.email},
{'id': self.user_3.email}])
def testHandleRequest_NoLimit(self):
mr = testing_helpers.MakeMonorailRequest()
self.services.user.users_by_id = {}
with self.assertRaisesRegexp(AssertionError, 'Missing param limit'):
self.task.HandleRequest(mr)
def testHandleRequest_NoOffset(self):
mr = testing_helpers.MakeMonorailRequest(path='url/url?limit=3')
self.services.user.users_by_id = {}
with self.assertRaisesRegexp(AssertionError, 'Missing param offset'):
self.task.HandleRequest(mr)
def testHandleRequest_ZeroOffset(self):
mr = testing_helpers.MakeMonorailRequest(path='url/url?limit=2&offset=0')
self.task.HandleRequest(mr)
self.task.sendUserLists.assert_called_once_with(
'service', [
{'id': self.user_1.email},
{'id': self.user_2.email}])
class DeleteWipeoutUsersTaskTest(unittest.TestCase):
def setUp(self):
self.services = service_manager.Services()
deleteusers.authorize = mock.Mock(return_value='service')
self.task = deleteusers.DeleteWipeoutUsersTask(services=self.services)
deleted_users = [
{'id': 'user1@gmail.com'}, {'id': 'user2@gmail.com'},
{'id': 'user3@gmail.com'}, {'id': 'user4@gmail.com'}]
self.task.fetchDeletedUsers = mock.Mock(return_value=deleted_users)
def generate_simple_task(self, url, body):
return {
'app_engine_http_request':
{
'relative_uri': url,
'body': body,
'headers': {
'Content-type': 'application/x-www-form-urlencoded'
}
}
}
@mock.patch('framework.cloud_tasks_helpers._get_client')
def testHandleRequest(self, get_client_mock):
mr = testing_helpers.MakeMonorailRequest(path='url/url?limit=3')
self.task.HandleRequest(mr)
deleteusers.authorize.assert_called_once_with()
self.task.fetchDeletedUsers.assert_called_once_with('service')
((_app_id, _region, queue),
_kwargs) = get_client_mock().queue_path.call_args
self.assertEqual(queue, framework_constants.QUEUE_DELETE_USERS)
self.assertEqual(get_client_mock().create_task.call_count, 2)
query = urllib.parse.urlencode(
{'emails': 'user1@gmail.com,user2@gmail.com,user3@gmail.com'})
expected_task = self.generate_simple_task(
urls.DELETE_USERS_TASK + '.do', query)
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
query = urllib.parse.urlencode({'emails': 'user4@gmail.com'})
expected_task = self.generate_simple_task(
urls.DELETE_USERS_TASK + '.do', query)
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)
@mock.patch('framework.cloud_tasks_helpers._get_client')
def testHandleRequest_DefaultMax(self, get_client_mock):
mr = testing_helpers.MakeMonorailRequest(path='url/url')
self.task.HandleRequest(mr)
deleteusers.authorize.assert_called_once_with()
self.task.fetchDeletedUsers.assert_called_once_with('service')
self.assertEqual(get_client_mock().create_task.call_count, 1)
emails = 'user1@gmail.com,user2@gmail.com,user3@gmail.com,user4@gmail.com'
query = urllib.parse.urlencode({'emails': emails})
expected_task = self.generate_simple_task(
urls.DELETE_USERS_TASK + '.do', query)
get_client_mock().create_task.assert_any_call(
get_client_mock().queue_path(),
expected_task,
retry=cloud_tasks_helpers._DEFAULT_RETRY)