From d58989d77739376bd140417f2d90da5b98938c73 Mon Sep 17 00:00:00 2001
From: Tomas Kopecek <tkopecek@redhat.com>
Date: Feb 20 2018 14:39:27 +0000
Subject: tests for notifications


---

diff --git a/hub/kojihub.py b/hub/kojihub.py
index 383c54c..51e9a2d 100644
--- a/hub/kojihub.py
+++ b/hub/kojihub.py
@@ -35,7 +35,6 @@ import errno
 import logging
 import fcntl
 import fnmatch
-import hashlib
 from koji.util import md5_constructor
 from koji.util import sha1_constructor
 from koji.util import dslice
@@ -53,7 +52,6 @@ import tarfile
 import tempfile
 import traceback
 import time
-import types
 import xmlrpclib
 import zipfile
 
@@ -10827,27 +10825,22 @@ class RootExports(object):
 
         # sanitize input
         if package_id is not None:
-            package_id = get_package_id(package_id)
+            package_id = get_package_id(package_id, strict=True)
         if tag_id is not None:
-            tag_id = get_tag_id(tag_id)
+            tag_id = get_tag_id(tag_id, strict=True)
         success_only = bool(success_only)
 
         # check existing notifications to not have same twice
         for notification in get_build_notifications(orig_notif['user_id']):
-            if notification['package_id'] == package_id and \
-               notification['tag_id'] == tag_id and \
-               notification['success_only'] == success_only and \
-               notification['email'] == email:
+            if (notification['package_id'] == package_id and
+                notification['tag_id'] == tag_id and
+                notification['success_only'] == success_only):
                 raise koji.GenericError('notification already exists')
 
-        update = """UPDATE build_notifications
-        SET package_id = %(package_id)s,
-        tag_id = %(tag_id)s,
-        success_only = %(success_only)s
-        WHERE id = %(id)i
-        """
-
-        _dml(update, locals())
+        update = UpdateProcessor('build_notifications',
+                                 clauses=['id = %(id)i'], values=locals())
+        update.set(package_id=package_id, tag_id=tag_id, success_only=success_only)
+        update.execute()
 
     def createNotification(self, user_id, package_id, tag_id, success_only):
         """Create a new notification.  If the user_id does not match the currently logged-in user
@@ -10867,27 +10860,24 @@ class RootExports(object):
         # sanitize input
         user_id = notificationUser['id']
         if package_id is not None:
-            package_id = get_package_id(package_id)
+            package_id = get_package_id(package_id, strict=True)
         if tag_id is not None:
-            tag_id = get_tag_id(tag_id)
+            tag_id = get_tag_id(tag_id, strict=True)
         success_only = bool(success_only)
 
         email = '%s@%s' % (notificationUser['name'], context.opts['EmailDomain'])
 
         # check existing notifications to not have same twice
         for notification in get_build_notifications(user_id):
-            if notification['package_id'] == package_id and \
-               notification['tag_id'] == tag_id and \
-               notification['success_only'] == success_only and \
-               notification['email'] == email:
+            if (notification['package_id'] == package_id and
+                notification['tag_id'] == tag_id and
+                notification['success_only'] == success_only):
                 raise koji.GenericError('notification already exists')
 
-        insert = """INSERT INTO build_notifications
-        (user_id, package_id, tag_id, success_only, email)
-        VALUES
-        (%(user_id)i, %(package_id)s, %(tag_id)s, %(success_only)s, %(email)s)
-        """
-        _dml(insert, locals())
+        insert = InsertProcessor('build_notifications')
+        insert.set(user_id=user_id, package_id=package_id, tag_id=tag_id,
+                   success_only=success_only, email=email)
+        insert.execute()
 
     def deleteNotification(self, id):
         """Delete the notification with the given ID.  If the currently logged-in
diff --git a/tests/test_hub/test_notifications.py b/tests/test_hub/test_notifications.py
index 74573c4..4629fdd 100644
--- a/tests/test_hub/test_notifications.py
+++ b/tests/test_hub/test_notifications.py
@@ -5,14 +5,28 @@ import koji
 import kojihub
 
 QP = kojihub.QueryProcessor
+IP = kojihub.InsertProcessor
+UP = kojihub.UpdateProcessor
+
+class TestNotifications(unittest.TestCase):
+    def getInsert(self, *args, **kwargs):
+        insert = IP(*args, **kwargs)
+        insert.execute = mock.MagicMock()
+        self.inserts.append(insert)
+        return insert
 
-class TestGetNotificationRecipients(unittest.TestCase):
     def getQuery(self, *args, **kwargs):
         query = QP(*args, **kwargs)
         query.execute = mock.MagicMock()
         self.queries.append(query)
         return query
 
+    def getUpdate(self, *args, **kwargs):
+        update = UP(*args, **kwargs)
+        update.execute = mock.MagicMock()
+        self.updates.append(update)
+        return update
+
     def setUp(self):
         self.context = mock.patch('kojihub.context').start()
         self.context.opts = {
@@ -23,6 +37,18 @@ class TestGetNotificationRecipients(unittest.TestCase):
         self.QueryProcessor = mock.patch('kojihub.QueryProcessor',
                 side_effect=self.getQuery).start()
         self.queries = []
+        self.InsertProcessor = mock.patch('kojihub.InsertProcessor',
+                side_effect=self.getInsert).start()
+        self.inserts = []
+        self.UpdateProcessor = mock.patch('kojihub.UpdateProcessor',
+                side_effect=self.getUpdate).start()
+        self.updates = []
+
+        self.exports = kojihub.RootExports()
+        self.exports.getLoggedInUser = mock.MagicMock()
+        self.exports.getUser = mock.MagicMock()
+        self.exports.hasPerm = mock.MagicMock()
+        self.exports.getBuildNotification = mock.MagicMock()
 
     def tearDown(self):
         mock.patch.stopall()
@@ -144,3 +170,352 @@ class TestGetNotificationRecipients(unittest.TestCase):
         }
         emails = kojihub.get_notification_recipients(build, tag_id, state)
         self.assertEqual(emails, ['owner_name@test.domain.com'])
+
+    #####################
+    # Create notification
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.getUser.return_value = {'id': 2, 'name': 'username'}
+        self.exports.hasPerm.return_value = True
+        get_package_id.return_value = package_id
+        get_tag_id.return_value = tag_id
+        get_build_notifications.return_value = []
+
+        r = self.exports.createNotification(user_id, package_id, tag_id, success_only)
+        self.assertEqual(r, None)
+
+        self.exports.getLoggedInUser.assert_called_once()
+        self.exports.getUser.asssert_called_once_with(user_id)
+        self.exports.hasPerm.asssert_called_once_with('admin')
+        get_package_id.assert_called_once_with(package_id, strict=True)
+        get_tag_id.assert_called_once_with(tag_id, strict=True)
+        get_build_notifications.assert_called_once_with(2)
+        self.assertEqual(len(self.inserts), 1)
+        insert = self.inserts[0]
+        self.assertEqual(insert.table, 'build_notifications')
+        self.assertEqual(insert.data, {
+            'package_id': package_id,
+            'user_id': 2,
+            'tag_id': tag_id,
+            'success_only': success_only,
+            'email': 'username@test.domain.com',
+        })
+        self.assertEqual(insert.rawdata, {})
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_unauthentized(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_invalid_user(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 2
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.getUser.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_no_perm(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 2
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1, 'name': 'a'}
+        self.exports.getUser.return_value = {'id': 2, 'name': 'b'}
+        self.exports.hasPerm.return_value = False
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_invalid_pkg(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 2
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
+        self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
+        get_package_id.side_effect = ValueError
+
+        with self.assertRaises(ValueError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_invalid_tag(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 2
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
+        self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
+        get_package_id.return_value = package_id
+        get_tag_id.side_effect = ValueError
+
+        with self.assertRaises(ValueError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_createNotification_exists(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        user_id = 2
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 2, 'name': 'a'}
+        self.exports.getUser.return_value = {'id': 2, 'name': 'a'}
+        get_package_id.return_value = package_id
+        get_tag_id.return_value = tag_id
+        get_build_notifications.return_value = [{
+            'package_id': package_id,
+            'tag_id': tag_id,
+            'success_only': success_only,
+        }]
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.createNotification(user_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+
+    #####################
+    # Delete notification
+    @mock.patch('kojihub._dml')
+    def test_deleteNotification(self, _dml):
+        user_id = 752
+        n_id = 543
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+
+        self.exports.deleteNotification(n_id)
+
+        self.exports.getBuildNotification.assert_called_once_with(n_id)
+        self.exports.getLoggedInUser.assert_called_once_with()
+        _dml.assert_called_once()
+
+    @mock.patch('kojihub._dml')
+    def test_deleteNotification_missing(self, _dml):
+        user_id = 752
+        n_id = 543
+        self.exports.getBuildNotification.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.deleteNotification(n_id)
+
+        self.exports.getBuildNotification.assert_called_once_with(n_id)
+        _dml.assert_not_called()
+
+    @mock.patch('kojihub._dml')
+    def test_deleteNotification_not_logged(self, _dml):
+        user_id = 752
+        n_id = 543
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+        self.exports.getLoggedInUser.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.deleteNotification(n_id)
+
+        self.exports.getBuildNotification.assert_called_once_with(n_id)
+        _dml.assert_not_called()
+
+    @mock.patch('kojihub._dml')
+    def test_deleteNotification_no_perm(self, _dml):
+        user_id = 752
+        n_id = 543
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.hasPerm.return_value = False
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.deleteNotification(n_id)
+
+        self.exports.getBuildNotification.assert_called_once_with(n_id)
+        _dml.assert_not_called()
+
+
+    #####################
+    # Update notification
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.hasPerm.return_value = True
+        get_package_id.return_value = package_id
+        get_tag_id.return_value = tag_id
+        get_build_notifications.return_value = [{
+            'tag_id': tag_id,
+            'user_id': user_id,
+            'package_id': package_id,
+            'success_only': not success_only,
+        }]
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+
+        r = self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+        self.assertEqual(r, None)
+
+        self.exports.getLoggedInUser.assert_called_once()
+        self.exports.hasPerm.asssert_called_once_with('admin')
+        get_package_id.assert_called_once_with(package_id, strict=True)
+        get_tag_id.assert_called_once_with(tag_id, strict=True)
+        get_build_notifications.assert_called_once_with(user_id)
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 1)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification_not_logged(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification_missing(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.getBuildNotification.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification_no_perm(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 132}
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+        self.exports.hasPerm.return_value = False
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification_exists(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = {'id': 1}
+        self.exports.hasPerm.return_value = True
+        get_package_id.return_value = package_id
+        get_tag_id.return_value = tag_id
+        get_build_notifications.return_value = [{
+            'tag_id': tag_id,
+            'user_id': user_id,
+            'package_id': package_id,
+            'success_only': success_only,
+        }]
+        self.exports.getBuildNotification.return_value = {'user_id': user_id}
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+
+        self.exports.getLoggedInUser.assert_called_once()
+        self.exports.hasPerm.asssert_called_once_with('admin')
+        get_package_id.assert_called_once_with(package_id, strict=True)
+        get_tag_id.assert_called_once_with(tag_id, strict=True)
+        get_build_notifications.assert_called_once_with(user_id)
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 0)
+
+    @mock.patch('kojihub.get_build_notifications')
+    @mock.patch('kojihub.get_tag_id')
+    @mock.patch('kojihub.get_package_id')
+    def test_updateNotification_not_logged(self, get_package_id, get_tag_id,
+            get_build_notifications):
+        n_id = 5432
+        user_id = 1
+        package_id = 234
+        tag_id = 345
+        success_only = True
+        self.exports.getLoggedInUser.return_value = None
+
+        with self.assertRaises(koji.GenericError):
+            self.exports.updateNotification(n_id, package_id, tag_id, success_only)
+
+        self.assertEqual(len(self.inserts), 0)
+        self.assertEqual(len(self.updates), 0)