debian-koji/tests/test_hub/test_get_notification_recipients.py

from unittest import mock
import unittest

import koji
import kojihub

QP = kojihub.QueryProcessor


class TestGetNotificationRecipients(unittest.TestCase):

    def getQuery(self, *args, **kwargs):
        query = QP(*args, **kwargs)
        query.execute = mock.MagicMock()
        self.queries.append(query)
        return query

    def setUp(self):
        self.context = mock.patch('kojihub.kojihub.context').start()
        self.context.opts = {
            'EmailDomain': 'test.domain.com',
            'NotifyOnSuccess': True,
        }

        self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',
                                         side_effect=self.getQuery).start()
        self.queries = []
        self.readPackageList = mock.patch('kojihub.kojihub.readPackageList').start()
        self.get_user = mock.patch('kojihub.kojihub.get_user').start()

        self.exports = kojihub.RootExports()

    def tearDown(self):
        mock.patch.stopall()

    def test_get_notification_recipients_watchers(self):
        # without build / tag_id
        build = None
        tag_id = None
        state = koji.BUILD_STATES['CANCELED']

        emails = kojihub.get_notification_recipients(build, tag_id, state)
        self.assertEqual(emails, [])

        # only query to watchers
        self.assertEqual(len(self.queries), 1)
        q = self.queries[0]
        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',
                                     'success_only = FALSE',
                                     'tag_id IS NULL',
                                     'usertype IN %(users_usertypes)s'])
        self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.id'])
        self.assertEqual(q.values['state'], state)
        self.assertEqual(q.values['build'], build)
        self.assertEqual(q.values['tag_id'], tag_id)
        self.readPackageList.assert_not_called()

    def test_get_notification_recipients_build_without_tag(self):
        ### with build without tag
        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), 2)
        q = self.queries[0]
        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 IS NULL',
                                     '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 IS NULL',
                                     'user_id IN %(user_ids)s',
                                     ])
        self.assertEqual(q.joins, None)
        self.assertEqual(q.values['user_ids'], [5])

        self.readPackageList.assert_not_called()

    def test_get_notification_not_recipients(self):
        ### without build, without tag, result not recipients
        tag_id = None
        state = koji.BUILD_STATES['CANCELED']
        build = None
        self.queries = []
        self.set_queries([[]])

        result = kojihub.get_notification_recipients(build, tag_id, state)
        self.assertEqual(result, [])

        # there should be only query to watchers
        self.assertEqual(len(self.queries), 1)
        q = self.queries[0]
        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',
                                     'success_only = FALSE',
                                     'tag_id IS NULL',
                                     'usertype IN %(users_usertypes)s'])
        self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.id'])
        self.assertEqual(q.values['state'], state)
        self.assertEqual(q.values['build'], build)
        self.assertEqual(q.values['tag_id'], tag_id)

        self.readPackageList.assert_not_called()

    def test_get_notification_recipients_tag_without_build(self):
        ### with tag without build makes no sense
        build = None
        tag_id = 123
        state = koji.BUILD_STATES['CANCELED']
        self.queries = []

        with self.assertRaises(koji.GenericError) as ex:
            kojihub.get_notification_recipients(build, tag_id, state)
        self.assertEqual('Invalid call', str(ex.exception))
        self.assertEqual(self.queries, [])
        self.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

    def test_get_notification_recipients_tag_with_build(self):
        ### with tag and build
        build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
        tag_id = 123
        state = koji.BUILD_STATES['CANCELED']
        self.readPackageList.return_value = {12345: {'blocked': False, 'owner_id': 'owner_id'}}
        self.get_user.return_value = {
            'id': 342,
            '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']}],
            []
        ])

        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), 2)
        q = self.queries[0]
        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',
                                     ])
        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(sorted(q.values['user_ids']), [5, 342])
        self.readPackageList.assert_called_once_with(
            pkgID=build['package_id'], tagID=tag_id, inherit=True)
        self.get_user.asssert_called_once_with(342, strict=True)

    def test_get_notification_recipients_blocked_pkg_owner(self):
        # blocked package owner
        build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
        tag_id = 123
        state = koji.BUILD_STATES['CANCELED']
        self.get_user.return_value = {
            'id': 342,
            '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'])

    def test_get_notification_recipients_optout(self):
        # blocked package owner
        build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
        tag_id = 123
        state = koji.BUILD_STATES['CANCELED']
        self.get_user.return_value = {
            'id': 342,
            '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, [])

    def test_get_notification_recipients_machine(self):
        # package owner is machine
        build = {'package_id': 12345, 'owner_name': 'owner_name', 'owner_id': 5}
        tag_id = 123
        state = koji.BUILD_STATES['CANCELED']
        self.get_user.return_value = {
            'id': 342,
            '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'])

    def test_get_notification_recipients_without_build(self):
        ### without build, without tag, result not recipients
        tag_id = None
        state = koji.BUILD_STATES['CANCELED']
        build = None
        self.queries = []
        self.set_queries([
            [{'user_id': 5, 'email': 'owner_name@%s' % self.context.opts['EmailDomain']}],
            []
        ])

        result = kojihub.get_notification_recipients(build, tag_id, state)
        self.assertEqual(result, ['owner_name@test.domain.com'])

        # there should be only query to watchers
        self.assertEqual(len(self.queries), 2)
        q = self.queries[0]
        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',
                                     'success_only = FALSE',
                                     'tag_id IS NULL',
                                     'usertype IN %(users_usertypes)s'])
        self.assertEqual(q.joins, ['JOIN users ON build_notifications.user_id = users.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 IS NULL',
                                     'tag_id IS NULL',
                                     'user_id IN %(user_ids)s',
                                     ])
        self.assertEqual(q.joins, None)
        self.assertEqual(q.values['user_ids'], [5])

        self.readPackageList.assert_not_called()