particle-os/debian-koji
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

notification optout tests

Browse source 
Related: https://pagure.io/koji/issue/1204
This commit is contained in:
Tomas Kopecek 2019-05-07 11:41:27 +02:00 • committed by Mike McLean
parent 6ae165f77b
commit e4c37dfffb
3 changed files with 131 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

31
hub/kojihub.py
View file

@ -7597,21 +7597,24 @@ def get_notification_recipients(build, tag_id, state):
return None
# apply the out outs
user_ids = [r['user_id'] for r in recipients]
clauses = ['user_id in %(user_ids)s']
if build:
package_id = build['package_id']
query.clauses.append('package_id = %(package_id)i OR package_id IS NULL')
user_ids = set([r['user_id'] for r in recipients])
if user_ids:
clauses = ['user_id IN %(user_ids)s']
if build:
package_id = build['package_id']
clauses.append('package_id = %(package_id)i OR package_id IS NULL')
else:
clauses.append('package_id IS NULL')
if tag_id:
clauses.append('tag_id = %(tag_id)i OR tag_id IS NULL')
else:
clauses.append('tag_id IS NULL')
query = QueryProcessor(columns=['user_id'], clauses=clauses,
tables=['build_notifications_block'], values=locals())
optouts = [r['user_id'] for r in query.execute()]
optouts = set(optouts)
else:
query.clauses.append('package_id IS NULL')
if tag_id:
query.clauses.append('tag_id = %(tag_id)i OR tag_id IS NULL')
else:
query.clauses.append('tag_id IS NULL')
query = QueryProcessor(columns=['user_id'], clauses=clauses,
tables=['build_notifications_block'], values=locals())
optouts = [r['user_id'] for r in query.execute()]
optouts = set(optouts)
optouts = set()
emails = [r['email'] for r in recipients if r['user_id'] not in optouts]
return list(set(emails))

2
tests/test_hub/test_get_build_notifications.py
View file

@ -8,8 +8,6 @@ import kojihub
class TestGetBuildNotifications(unittest.TestCase):
@mock.patch('kojihub.get_user', return_value={'id': 1})
@mock.patch('kojihub.get_build_notifications')
def test_loggedin_user(self, get_build_notifications, get_user):

124
tests/test_hub/test_notifications.py
View file

@ -60,7 +60,7 @@ class TestNotifications(unittest.TestCase):
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients(self, readPackageList, get_user):
def test_get_notification_recipients_watchers(self, readPackageList, get_user):
# without build / tag_id
build = None
tag_id = None
@ -72,7 +72,7 @@ class TestNotifications(unittest.TestCase):
# only query to watchers
self.assertEqual(len(self.queries), 1)
q = self.queries[0]
self.assertEqual(q.columns, ['email'])
self.assertEqual(q.columns, ['email', 'user_id'])
self.assertEqual(q.tables, ['build_notifications'])
self.assertEqual(q.clauses, ['package_id IS NULL',
'status = %(users_status)i',
@ -83,20 +83,38 @@ class TestNotifications(unittest.TestCase):
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
'''
q = self.queries[1]
self.assertEqual(q.columns, ['user_id'])
self.assertEqual(q.tables, ['build_notifications_block'])
self.assertEqual(q.clauses, ['user_id IN %(user_ids)s'])
self.assertEqual(q.joins, [])
self.assertEqual(q.values['user_ids'], None)
'''
readPackageList.assert_not_called()
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_build_without_tag(self, readPackageList, get_user):
### with build without tag
build = {'package_id': 12345, 'owner_name': 'owner_name'}
tag_id = None
state = koji.BUILD_STATES['CANCELED']
build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
self.queries = []
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[]
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(emails, ['owner_name@test.domain.com'])
# there should be only query to watchers
self.assertEqual(len(self.queries), 1)
self.assertEqual(len(self.queries), 2)
q = self.queries[0]
self.assertEqual(q.columns, ['email'])
self.assertEqual(q.columns, ['email', 'user_id'])
self.assertEqual(q.tables, ['build_notifications'])
self.assertEqual(q.clauses, ['package_id = %(package_id)i OR package_id IS NULL',
'status = %(users_status)i',
@ -108,11 +126,27 @@ class TestNotifications(unittest.TestCase):
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
q = self.queries[1]
self.assertEqual(q.columns, ['user_id'])
self.assertEqual(q.tables, ['build_notifications_block'])
self.assertEqual(q.clauses, [
'package_id = %(package_id)i OR package_id IS NULL',
'tag_id IS NULL',
'user_id IN %(user_ids)s',
])
self.assertEqual(q.joins, None)
self.assertEqual(q.values['user_ids'], set([5]))
readPackageList.assert_not_called()
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_tag_without_build(self, readPackageList, get_user):
### with tag without build makes no sense
build = None
tag_id = 123
state = koji.BUILD_STATES['CANCELED']
self.queries = []
with self.assertRaises(koji.GenericError):
@ -120,11 +154,24 @@ class TestNotifications(unittest.TestCase):
self.assertEqual(self.queries, [])
readPackageList.assert_not_called()
def set_queries(self, return_values):
self.query_returns = return_values
self.query_returns.reverse()
def getQuery(*args, **kwargs):
q = QP(*args, **kwargs)
q.execute = mock.MagicMock()
q.execute.return_value = self.query_returns.pop()
self.queries.append(q)
return q
self.QueryProcessor.side_effect = getQuery
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_tag_with_build(self, readPackageList, get_user):
### with tag and build
build = {'package_id': 12345, 'owner_name': 'owner_name'}
build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
tag_id = 123
self.queries = []
state = koji.BUILD_STATES['CANCELED']
readPackageList.return_value = {12345: {'blocked': False, 'owner_id': 'owner_id'}}
get_user.return_value = {
'id': 'owner_id',
@ -132,46 +179,103 @@ class TestNotifications(unittest.TestCase):
'status': koji.USER_STATUS['NORMAL'],
'usertype': koji.USERTYPES['NORMAL']
}
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[]
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(sorted(emails), ['owner_name@test.domain.com', 'pkg_owner_name@test.domain.com'])
# there should be only query to watchers
self.assertEqual(len(self.queries), 1)
self.assertEqual(len(self.queries), 2)
q = self.queries[0]
self.assertEqual(q.columns, ['email'])
self.assertEqual(q.columns, ['email', 'user_id'])
self.assertEqual(q.tables, ['build_notifications'])
self.assertEqual(q.clauses, ['package_id = %(package_id)i OR package_id IS NULL',
'status = %(users_status)i',
'success_only = FALSE',
'tag_id = %(tag_id)i OR tag_id IS NULL',
'usertype IN %(users_usertypes)s'])
'usertype IN %(users_usertypes)s',
])
self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.id'])
self.assertEqual(q.values['package_id'], build['package_id'])
self.assertEqual(q.values['state'], state)
self.assertEqual(q.values['build'], build)
self.assertEqual(q.values['tag_id'], tag_id)
q = self.queries[1]
self.assertEqual(q.columns, ['user_id'])
self.assertEqual(q.tables, ['build_notifications_block'])
self.assertEqual(q.clauses, [
'package_id = %(package_id)i OR package_id IS NULL',
'tag_id = %(tag_id)i OR tag_id IS NULL',
'user_id IN %(user_ids)s',
])
self.assertEqual(q.joins, None)
self.assertEqual(q.values['user_ids'], set([5, 'owner_id']))
readPackageList.assert_called_once_with(pkgID=build['package_id'], tagID=tag_id, inherit=True)
get_user.asssert_called_once_with('owner_id', strict=True)
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_blocked_pkg_owner(self, readPackageList, get_user):
# blocked package owner
build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
tag_id = 123
state = koji.BUILD_STATES['CANCELED']
get_user.return_value = {
'id': 'owner_id',
'name': 'pkg_owner_name',
'status': koji.USER_STATUS['BLOCKED'],
'usertype': koji.USERTYPES['NORMAL']
}
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[]
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(emails, ['owner_name@test.domain.com'])
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_optout(self, readPackageList, get_user):
# blocked package owner
build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
tag_id = 123
state = koji.BUILD_STATES['CANCELED']
get_user.return_value = {
'id': 'owner_id',
'name': 'pkg_owner_name',
'status': koji.USER_STATUS['NORMAL'],
'usertype': koji.USERTYPES['NORMAL']
}
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[{'user_id': 5}]
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(emails, [])
@mock.patch('kojihub.get_user')
@mock.patch('kojihub.readPackageList')
def test_get_notification_recipients_machine(self, readPackageList, get_user):
# package owner is machine
build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
tag_id = 123
state = koji.BUILD_STATES['CANCELED']
get_user.return_value = {
'id': 'owner_id',
'name': 'pkg_owner_name',
'status': koji.USER_STATUS['NORMAL'],
'usertype': koji.USERTYPES['HOST']
}
self.set_queries([
[{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
[]
])
emails = kojihub.get_notification_recipients(build, tag_id, state)
self.assertEqual(emails, ['owner_name@test.domain.com'])