#3752 Increase hub unit tests 03-02
Merged a year ago by tkopecek. Opened a year ago by jcupova.

Increase hub unit tests 03-02
Jana Cupova • a year ago  
@@ -6,12 +6,15 @@ 

  import koji

  import kojihub

  

+ IP = kojihub.InsertProcessor

+ 

  

  class TestCreateExternalRepo(unittest.TestCase):

  

      def setUp(self):

          self.get_external_repos = mock.patch('kojihub.kojihub.get_external_repos').start()

          self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start()

+         self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()

          self.context = mock.patch('kojihub.kojihub.context').start()

          # It seems MagicMock will not automatically handle attributes that

          # start with "assert"
@@ -20,6 +23,17 @@ 

          self.repo_url = 'http://path_to_ext_repo.com'

          self.repo_name = 'test-repo'

          self.repo_info = {'id': 1, 'name': self.repo_name, 'url': self.repo_url}

+         self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',

+                                           side_effect=self.getInsert).start()

+         self.inserts = []

+         self.context_db = mock.patch('kojihub.db.context').start()

+         self.context_db.session.assertLogin = mock.MagicMock()

+ 

+     def getInsert(self, *args, **kwargs):

+         insert = IP(*args, **kwargs)

+         insert.execute = mock.MagicMock()

+         self.inserts.append(insert)

+         return insert

  

      def tearDown(self):

          mock.patch.stopall()
@@ -45,3 +59,24 @@ 

          with self.assertRaises(koji.GenericError) as cm:

              kojihub.create_external_repo(self.repo_name, self.repo_url)

          self.assertEqual(expected, str(cm.exception))

+ 

+     def test_valid(self):

+         self.verify_name_internal.return_value = None

+         self.get_external_repos.return_value = None

+         self.context_db.event_id = 42

+         self.context_db.session.user_id = 23

+         self.get_external_repo_id.return_value = self.repo_info['id']

+         repo_url = self.repo_url + '/'

+         result = kojihub.create_external_repo(self.repo_name, self.repo_url)

+         self.assertEqual(result,

+                          {'id': self.repo_info['id'], 'name': self.repo_name, 'url': repo_url})

+         self.assertEqual(len(self.inserts), 1)

+         insert = self.inserts[0]

+         self.assertEqual(insert.table, 'external_repo_config')

+         self.assertEqual(insert.data, {

+             'create_event': 42,

+             'creator_id': 23,

+             'external_repo_id': self.repo_info['id'],

+             'url': repo_url

+         })

+         self.assertEqual(insert.rawdata, {})

@@ -6,15 +6,46 @@ 

  import kojihub

  

  

+ UP = kojihub.UpdateProcessor

+ 

+ 

  class TestDeleteBuildTarget(unittest.TestCase):

  

+     def getUpdate(self, *args, **kwargs):

+         update = UP(*args, **kwargs)

+         update.execute = mock.MagicMock()

+         self.updates.append(update)

+         return update

+ 

      def setUp(self):

-         self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start()

+         self.lookup_build_target = mock.patch('kojihub.kojihub.lookup_build_target').start()

          self.exports = kojihub.RootExports()

+         self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

+                                           side_effect=self.getUpdate).start()

+         self.updates = []

+         self.context_db = mock.patch('kojihub.db.context').start()

+         # It seems MagicMock will not automatically handle attributes that

+         # start with "assert"

+         self.context_db.session.assertLogin = mock.MagicMock()

  

      def test_non_exist_target(self):

          build_target = 'build-target'

-         self.lookup_name.return_value = None

+         self.lookup_build_target.return_value = None

          with self.assertRaises(koji.GenericError) as cm:

              self.exports.deleteBuildTarget(build_target)

          self.assertEqual("No such build target: %s" % build_target, str(cm.exception))

+         self.assertEqual(len(self.updates), 0)

+ 

+     def test_valid(self):

+         build_target = 'build-target'

+         self.context_db.event_id = 42

+         self.context_db.session.user_id = 23

+         self.lookup_build_target.return_value = {'id': 123}

+         self.exports.deleteBuildTarget(build_target)

+ 

+         self.assertEqual(len(self.updates), 1)

+         update = self.updates[0]

+         self.assertEqual(update.table, 'build_target_config')

+         self.assertEqual(update.clauses, ["build_target_id = %(targetID)i", 'active = TRUE'])

+         self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})

+         self.assertEqual(update.rawdata, {'active': 'NULL'})

@@ -0,0 +1,55 @@ 

+ import unittest

+ 

+ import mock

+ 

+ import kojihub

+ 

+ UP = kojihub.UpdateProcessor

+ 

+ 

+ class TestDeleteExternalRepo(unittest.TestCase):

+     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.kojihub.context').start()

+         self.context.session.assertPerm = mock.MagicMock()

+         self.context_db = mock.patch('kojihub.db.context').start()

+         self.context_db.session.assertLogin = mock.MagicMock()

+         self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

+                                           side_effect=self.getUpdate).start()

+         self.updates = []

+         self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()

+         self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()

+         self.remove_external_repo_from_tag = mock.patch(

+             'kojihub.kojihub.remove_external_repo_from_tag').start()

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     def test_valid(self):

+         tag_ids = [23, 25]

+         repo_id = 123

+         self.context_db.event_id = 42

+         self.context_db.session.user_id = 23

+         self.get_external_repo.return_value = {'id': repo_id}

+         self.get_tag_external_repos.return_value = [{'id': repo_id, 'tag_id': tag_ids[0]},

+                                                     {'id': repo_id, 'tag_id': tag_ids[0]}]

+         self.remove_external_repo_from_tag.side_effect = [None, None]

+         kojihub.delete_external_repo(repo_id)

+ 

+         self.assertEqual(len(self.updates), 1)

+         update = self.updates[0]

+         self.assertEqual(update.table, 'external_repo_config')

+         self.assertEqual(update.clauses, ['external_repo_id = %(repo_id)i', 'active = TRUE'])

+         self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})

+         self.assertEqual(update.rawdata, {'active': 'NULL'})

+ 

+         self.get_external_repo.assert_called_once_with(123, strict=True)

+         self.get_tag_external_repos.assert_called_once_with(repo_info=123)

+         self.remove_external_repo_from_tag.has_calls(

+             [mock.call(tag_info=tag_ids[0], repo_info=repo_id),

+              mock.call(tag_info=tag_ids[1], repo_info=repo_id)])

@@ -6,9 +6,33 @@ 

  import koji

  import kojihub

  

+ IP = kojihub.InsertProcessor

+ QP = kojihub.QueryProcessor

+ UP = kojihub.UpdateProcessor

+ 

  

  class TestEditExternalRepo(unittest.TestCase):

  

+     def getQuery(self, *args, **kwargs):

+         query = QP(*args, **kwargs)

+         query.execute = mock.MagicMock()

+         query.executeOne = mock.MagicMock()

+         query.singleValue = self.query_singleValue

+         self.queries.append(query)

+         return query

+ 

+     def getInsert(self, *args, **kwargs):

+         insert = IP(*args, **kwargs)

+         insert.execute = mock.MagicMock()

+         self.inserts.append(insert)

+         return insert

+ 

+     def getUpdate(self, *args, **kwargs):

+         update = UP(*args, **kwargs)

+         update.execute = mock.MagicMock()

+         self.updates.append(update)

+         return update

+ 

      def setUp(self):

          self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()

          self.verify_name_internal = mock.patch('kojihub.kojihub.verify_name_internal').start()
@@ -20,6 +44,20 @@ 

          self.repo_url = 'http://path_to_ext_repo.com'

          self.repo_name = 'test-repo'

          self.repo_info = {'id': 1, 'name': self.repo_name, 'url': self.repo_url}

+         self.context_db = mock.patch('kojihub.db.context').start()

+         # It seems MagicMock will not automatically handle attributes that

+         # start with "assert"

+         self.context_db.session.assertLogin = mock.MagicMock()

+         self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',

+                                           side_effect=self.getInsert).start()

+         self.inserts = []

+         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

+                                          side_effect=self.getQuery).start()

+         self.queries = []

+         self.query_singleValue = mock.MagicMock()

+         self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

+                                           side_effect=self.getUpdate).start()

+         self.updates = []

  

      def tearDown(self):

          mock.patch.stopall()
@@ -37,3 +75,67 @@ 

          self.verify_name_internal.side_effect = koji.GenericError

          with self.assertRaises(koji.GenericError):

              kojihub.edit_external_repo(self.repo_name, repo_name_new, self.repo_url)

+ 

+     def test_edit_external_repo_existing_name(self):

+         repo_name_new = 'test-repo+'

+         self.get_external_repo.return_value = self.repo_info

+         self.verify_name_internal.side_effect = None

+         self.query_singleValue.return_value = 111

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.edit_external_repo(self.repo_name, repo_name_new, self.repo_url)

+         self.assertEqual(f'name "{repo_name_new}" is already taken by external repo {111}',

+                          str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         self.assertEqual(len(self.inserts), 0)

+         self.assertEqual(len(self.updates), 0)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['external_repo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['name = %(name)s'])

+         self.assertEqual(query.columns, ['id'])

+         self.assertEqual(query.values, {'name': repo_name_new})

+ 

+     def test_edit_valid(self):

+         path_to_new_repo = 'http://path_to_new_ext_repo.com'

+         repo_name_new = 'test-repo+'

+         self.get_external_repo.return_value = self.repo_info

+         self.verify_name_internal.side_effect = None

+         self.query_singleValue.return_value = None

+         self.context_db.event_id = 42

+         self.context_db.session.user_id = 23

+         kojihub.edit_external_repo(self.repo_name, repo_name_new, path_to_new_repo)

+ 

+         self.assertEqual(len(self.queries), 1)

+         self.assertEqual(len(self.inserts), 1)

+         self.assertEqual(len(self.updates), 2)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['external_repo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['name = %(name)s'])

+         self.assertEqual(query.columns, ['id'])

+         self.assertEqual(query.values, {'name': repo_name_new})

+ 

+         insert = self.inserts[0]

+         self.assertEqual(insert.table, 'external_repo_config')

+         self.assertEqual(insert.data, {

+             'create_event': 42,

+             'creator_id': 23,

+             'external_repo_id': self.repo_info['id'],

+             'url': path_to_new_repo + '/'

+         })

+         self.assertEqual(insert.rawdata, {})

+ 

+         update = self.updates[0]

+         self.assertEqual(update.table, 'external_repo')

+         self.assertEqual(update.values, {'repo_id': self.repo_info['id']})

+         self.assertEqual(update.clauses, ['id = %(repo_id)i'])

+         self.assertEqual(update.data, {'name': repo_name_new})

+         self.assertEqual(update.rawdata, {})

+ 

+         update = self.updates[1]

+         self.assertEqual(update.table, 'external_repo_config')

+         self.assertEqual(update.clauses, ['external_repo_id = %(repo_id)i', 'active = TRUE'])

+         self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})

+         self.assertEqual(update.rawdata, {'active': 'NULL'})

@@ -10,7 +10,7 @@ 

      maxDiff = None

  

      def test_getPackageID(self):

-         self.qp_execute_return_value = [{'id': 1}]

+         self.qp_execute_one_return_value = {'id': 1}

          rv = kojihub.RootExports().getPackageID('koji')

          self.assertEqual(len(self.queries), 1)

          self.assertLastQueryEqual(tables=['package'],
@@ -30,8 +30,7 @@ 

                                    values={'name': 'invalidpkg',

                                            'strict': True,

                                            'self': mock.ANY})

-         self.assertEqual(cm.exception.args[0],

-                          'No such package name: invalidpkg')

+         self.assertEqual(cm.exception.args[0], 'No such package name: invalidpkg')

  

      def test_getPackageID_None(self):

          rv = kojihub.RootExports().getPackageID('invalidpkg')

file modified
+46 -17
@@ -1,33 +1,21 @@ 

  import os.path

  import shutil

  import tempfile

- import unittest

  import mock

+ import unittest

  

  import koji

  import kojihub

- 

- QP = kojihub.QueryProcessor

+ from .utils import DBQueryTestCase

  

  

- class TestGetRPM(unittest.TestCase):

+ class TestGetRPM(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetRPM, self).setUp()

          self.exports = kojihub.RootExports()

          self.context = mock.patch('kojihub.kojihub.context').start()

          self.get_external_repo_id = mock.patch('kojihub.kojihub.get_external_repo_id').start()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         self.queries.append(query)

-         return query

- 

-     def tearDown(self):

-         mock.patch.stopall()

  

      def test_wrong_type_rpminfo(self):

          rpminfo = ['test-user']
@@ -37,7 +25,9 @@ 

  

      def test_rpm_info_int(self):

          rpminfo = 123

-         kojihub.get_rpm(rpminfo)

+         self.qp_execute_one_return_value = {'rpminfo.id': 123}

+         result = kojihub.get_rpm(rpminfo)

+         self.assertEqual(result, {'rpminfo.id': 123})

  

          self.assertEqual(len(self.queries), 1)

          query = self.queries[0]
@@ -52,6 +42,45 @@ 

                           ['external_repo ON rpminfo.external_repo_id = external_repo.id'])

          self.assertEqual(query.values, {'id': rpminfo})

  

+     def test_rpm_info_empty_query_result_without_strict(self):

+         rpminfo = 123

+         self.qp_execute_one_return_value = {}

+         result = kojihub.get_rpm(rpminfo)

+         self.assertEqual(result, None)

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         str(query)

+         self.assertEqual(query.tables, ['rpminfo'])

+         columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',

+                    'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',

+                    'size', 'buildtime', 'metadata_only', 'extra']

+         self.assertEqual(set(query.columns), set(columns))

+         self.assertEqual(query.clauses, ["rpminfo.id=%(id)s"])

+         self.assertEqual(query.joins,

+                          ['external_repo ON rpminfo.external_repo_id = external_repo.id'])

+         self.assertEqual(query.values, {'id': rpminfo})

+ 

+     def test_rpm_info_empty_query_result_with_strict(self):

+         rpminfo = 123

+         self.qp_execute_one_return_value = {}

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.get_rpm(rpminfo, strict=True)

+         self.assertEqual("No such rpm: {'id': 123}", str(ex.exception))

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         str(query)

+         self.assertEqual(query.tables, ['rpminfo'])

+         columns = ['rpminfo.id', 'build_id', 'buildroot_id', 'rpminfo.name', 'version', 'release',

+                    'epoch', 'arch', 'external_repo_id', 'external_repo.name', 'payloadhash',

+                    'size', 'buildtime', 'metadata_only', 'extra']

+         self.assertEqual(set(query.columns), set(columns))

+         self.assertEqual(query.clauses, ["rpminfo.id=%(id)s"])

+         self.assertEqual(query.joins,

+                          ['external_repo ON rpminfo.external_repo_id = external_repo.id'])

+         self.assertEqual(query.values, {'id': rpminfo})

+ 

      def test_rpm_info_str(self):

          rpminfo = 'testrpm-1.23-4.x86_64.rpm'

          kojihub.get_rpm(rpminfo, multi=True)

@@ -6,49 +6,50 @@ 

  

  

  class TestGetRPMDeps(unittest.TestCase):

+     def setUp(self):

+         self.exports = kojihub.RootExports()

+         self.get_rpm = mock.patch('kojihub.kojihub.get_rpm').start()

+         self.get_build = mock.patch('kojihub.kojihub.get_build').start()

  

-     @mock.patch('kojihub.kojihub.get_rpm')

-     def test_getRPMDeps_no_rpminfo(self, get_rpm):

+     def test_getRPMDeps_no_rpminfo(self):

          def mock_get_rpm(rpmID, strict=False):

              if strict:

                  raise koji.GenericError('msg')

              else:

                  return None

-         get_rpm.side_effect = mock_get_rpm

-         re = kojihub.RootExports().getRPMDeps(1)

+         self.get_rpm.side_effect = mock_get_rpm

+         re = self.exports.getRPMDeps(1)

          self.assertEqual(re, [])

          with self.assertRaises(koji.GenericError) as cm:

-             kojihub.RootExports().getRPMDeps(1, strict=True)

+             self.exports.getRPMDeps(1, strict=True)

          self.assertEqual(cm.exception.args[0], 'msg')

  

-     @mock.patch('kojihub.kojihub.get_rpm', return_value={'id': 1, 'build_id': None})

-     def test_getRPMDeps_external_rpm(self, get_rpm):

-         re = kojihub.RootExports().getRPMDeps(1)

+     def test_getRPMDeps_external_rpm(self):

+         self.get_rpm.return_value = {'id': 1, 'build_id': None}

+         re = self.exports.getRPMDeps(1)

          self.assertEqual(re, [])

          with self.assertRaises(koji.GenericError) as cm:

-             kojihub.RootExports().getRPMDeps(1, strict=True)

+             self.exports.getRPMDeps(1, strict=True)

          self.assertEqual(cm.exception.args[0],

                           'Can not get dependencies, because RPM: 1 is not internal')

  

-     @mock.patch('kojihub.kojihub.get_rpm', return_value={'id': 1, 'build_id': 1})

-     @mock.patch('kojihub.kojihub.get_build', return_value={'id': 1})

      @mock.patch('koji.pathinfo.build', return_value='fakebuildpath')

      @mock.patch('koji.pathinfo.rpm', return_value='fakerpmrelpath')

      @mock.patch('os.path.exists', return_value=False)

-     def test_getRPMDeps_no_rpmfile(self, ope, pr, pb, get_build, get_rpm):

-         re = kojihub.RootExports().getRPMDeps(1)

+     def test_getRPMDeps_no_rpmfile(self, ope, pr, pb):

+         self.get_rpm.return_value = {'id': 1, 'build_id': 1}

+         self.get_build.return_value = {'id': 1}

+         re = self.exports.getRPMDeps(1)

          self.assertEqual(re, [])

          with self.assertRaises(koji.GenericError) as cm:

-             kojihub.RootExports().getRPMDeps(1, strict=True)

+             self.exports.getRPMDeps(1, strict=True)

          self.assertEqual(cm.exception.args[0], "RPM file of 1 doesn't exist")

  

-     @mock.patch('kojihub.kojihub.get_rpm')

-     @mock.patch('kojihub.kojihub.get_build')

      @mock.patch('koji.pathinfo')

-     def test_getRPMDeps(self, pi, build, rpm):

+     def test_getRPMDeps(self, pi):

          pi.build.return_value = os.path.join(os.path.dirname(__file__), '../test_lib/data/rpms')

          pi.rpm.return_value = 'test-deps-1-1.fc24.x86_64.rpm'

-         getRPMDeps = kojihub.RootExports().getRPMDeps

+         getRPMDeps = self.exports.getRPMDeps

          res = getRPMDeps('')

          # limit test for rpm < 4.12

          if any(koji.SUPPORTED_OPT_DEP_HDRS.values()):

@@ -25,7 +25,7 @@ 

  

      def test_get_active_repos(self):

          # currently not really a lot of parameters to test

-         result = kojihub.get_active_repos()

+         kojihub.get_active_repos()

          self.assertEqual(len(self.queries), 1)

          query = self.queries[0]

          # make sure the following does not error

@@ -1,20 +1,190 @@ 

- import unittest

- 

  import koji

  import kojihub

+ import mock

+ from .utils import DBQueryTestCase

  

  

- class TestGetArchiveType(unittest.TestCase):

+ class TestGetArchiveType(DBQueryTestCase):

+     def setUp(self):

+         super(TestGetArchiveType, self).setUp()

+         self.maxDiff = None

+         self.archive_info = {'id': 1, 'name': 'archive-type-1',

+                              'description': 'archive-desc', 'extensions': 'ext',

+                              'compression_type': 'cmptype'}

  

-     def test_get_archive_wrong_type_filename(self):

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_wrong_type_filename(

+             self, get_archive_type_by_id, get_archive_type_by_name):

          filename = ['test-filename']

-         with self.assertRaises(koji.ParameterError) as cm:

+         with self.assertRaises(koji.ParameterError) as ex:

              kojihub.get_archive_type(filename=filename)

          self.assertEqual(f"Invalid type for value '{filename}': {type(filename)}, "

-                          f"expected type <class 'str'>", str(cm.exception))

+                          f"expected type <class 'str'>", str(ex.exception))

+         self.assertEqual(len(self.queries), 0)

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

  

-     def test_get_archive_without_opt(self):

-         with self.assertRaises(koji.GenericError) as cm:

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_without_opt(self, get_archive_type_by_id, get_archive_type_by_name):

+         with self.assertRaises(koji.GenericError) as ex:

              kojihub.get_archive_type()

          self.assertEqual("one of filename, type_name, or type_id must be specified",

-                          str(cm.exception))

+                          str(ex.exception))

+         self.assertEqual(len(self.queries), 0)

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_type_id(self, get_archive_type_by_id, get_archive_type_by_name):

+         get_archive_type_by_id.return_value = self.archive_info

+         kojihub.get_archive_type(type_id=1)

+         self.assertEqual(len(self.queries), 0)

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_called_once_with(1, False)

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_type_name(self, get_archive_type_by_id, get_archive_type_by_name):

+         get_archive_type_by_name.return_value = self.archive_info

+         kojihub.get_archive_type(type_name='archive-type-1')

+         self.assertEqual(len(self.queries), 0)

+         get_archive_type_by_name.assert_called_once_with('archive-type-1', False)

+         get_archive_type_by_id.assert_not_called()

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_more_files_extension(

+             self, get_archive_type_by_id, get_archive_type_by_name):

+         archive_info = [{'id': 1, 'name': 'archive-type-1', 'extensions': 'ext'},

+                         {'id': 2, 'name': 'archive-type-2', 'extensions': 'ext'}]

+         filename = 'test-filename.ext'

+         self.qp_execute_return_value = archive_info

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.get_archive_type(filename=filename)

+         self.assertEqual("multiple matches for file extension: ext", str(ex.exception))

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_no_extensions_with_strict(

+             self, get_archive_type_by_id, get_archive_type_by_name):

+         archive_info = []

+         filename = 'test-filename.ext'

+         self.qp_execute_return_value = archive_info

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.get_archive_type(filename=filename, strict=True)

+         self.assertEqual(f'unsupported file extension: {filename}', str(ex.exception))

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_no_extensions_without_strict(

+             self, get_archive_type_by_id, get_archive_type_by_name):

+         archive_info = []

+         filename = 'test-filename.ext'

+         self.qp_execute_return_value = archive_info

+         result = kojihub.get_archive_type(filename=filename, strict=False)

+         self.assertEqual(result, None)

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

+ 

+     @mock.patch('kojihub.kojihub._get_archive_type_by_name')

+     @mock.patch('kojihub.kojihub._get_archive_type_by_id')

+     def test_get_archive_valid(self, get_archive_type_by_id, get_archive_type_by_name):

+         filename = 'test-filename.ext'

+         self.qp_execute_return_value = [self.archive_info]

+         result = kojihub.get_archive_type(filename=filename)

+         self.assertEqual(result, self.archive_info)

+ 

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['extensions ~* %(pattern)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         get_archive_type_by_name.assert_not_called()

+         get_archive_type_by_id.assert_not_called()

+ 

+     def test_get_archive_type_by_id_empty_without_strict(self):

+         self.qp_execute_one_return_value = None

+         result = kojihub._get_archive_type_by_id(1, False)

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['id = %(type_id)i'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         self.assertEqual(query.values, {'type_id': 1})

+ 

+     def test_get_archive_type_by_id_empty_with_strict(self):

+         self.qp_execute_one_side_effect = koji.GenericError('query returned no rows')

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub._get_archive_type_by_id(1, True)

+         self.assertEqual('query returned no rows', str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['id = %(type_id)i'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         self.assertEqual(query.values, {'type_id': 1})

+ 

+     def test_get_archive_type_by_name_without_strict(self):

+         self.qp_execute_one_return_value = None

+         result = kojihub._get_archive_type_by_name('archive-type', False)

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['name = %(name)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         self.assertEqual(query.values, {'name': 'archive-type'})

+ 

+     def test_get_archive_type_by_name_with_strict(self):

+         self.qp_execute_one_side_effect = koji.GenericError('query returned no rows')

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub._get_archive_type_by_name('archive-type', True)

+         self.assertEqual('query returned no rows', str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['archivetypes'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['name = %(name)s'])

+         self.assertEqual(query.columns,

+                          ['compression_type', 'description', 'extensions', 'id', 'name'])

+         self.assertEqual(query.values, {'name': 'archive-type'})

@@ -10,22 +10,69 @@ 

      def setUp(self):

          super(TestGetBuild, self).setUp()

          self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()

+         self.lookup_name = mock.patch('kojihub.kojihub.lookup_name').start()

  

-     def test_non_exist_build_string(self):

+     def test_non_exist_build_string_with_strict(self):

          build = 'build-1-23'

          self.find_build_id.side_effect = koji.GenericError('No such build: %s' % build)

          with self.assertRaises(koji.GenericError) as cm:

              kojihub.get_build(build, strict=True)

          self.assertEqual('No such build: %s' % build, str(cm.exception))

+         self.find_build_id.assert_called_once_with(build, strict=True)

+         self.assertEqual(len(self.queries), 0)

  

-     def test_non_exist_build_int(self):

+     def test_non_exist_build_int_without_result_with_strict(self):

          build = 11

          self.find_build_id.return_value = build

          with self.assertRaises(koji.GenericError) as cm:

              kojihub.get_build(build, strict=True)

          self.assertEqual('No such build: %s' % build, str(cm.exception))

+         self.find_build_id.assert_called_once_with(build, strict=True)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build'])

+         self.assertEqual(query.joins, ['events ON build.create_event = events.id',

+                                        'package on build.pkg_id = package.id',

+                                        'volume on build.volume_id = volume.id',

+                                        'users on build.owner = users.id'])

+         self.assertEqual(query.clauses, ['build.id = %(buildID)i'])

+         self.assertEqual(query.columns,

+                          ['build.id', 'build.cg_id', 'build.completion_time',

+                           "date_part('epoch', build.completion_time)", 'events.id', 'events.time',

+                           "date_part('epoch', events.time)", 'build.epoch', 'build.extra',

+                           'build.id', 'package.name',

+                           "package.name || '-' || build.version || '-' || build.release",

+                           'users.id', 'users.name', 'package.id', 'package.name', 'build.release',

+                           'build.source', 'build.start_time',

+                           "date_part('epoch', build.start_time)", 'build.state', 'build.task_id',

+                           'build.version', 'volume.id', 'volume.name'])

  

-     def test_non_exist_build_dict(self):

+     def test_non_exist_build_int_without_result_without_strict(self):

+         build = 11

+         self.find_build_id.return_value = build

+         result = kojihub.get_build(build, strict=False)

+         self.assertEqual(result, None)

+         self.find_build_id.assert_called_once_with(build, strict=False)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build'])

+         self.assertEqual(query.joins, ['events ON build.create_event = events.id',

+                                        'package on build.pkg_id = package.id',

+                                        'volume on build.volume_id = volume.id',

+                                        'users on build.owner = users.id'])

+         self.assertEqual(query.clauses, ['build.id = %(buildID)i'])

+         self.assertEqual(query.columns,

+                          ['build.id', 'build.cg_id', 'build.completion_time',

+                           "date_part('epoch', build.completion_time)", 'events.id', 'events.time',

+                           "date_part('epoch', events.time)", 'build.epoch', 'build.extra',

+                           'build.id', 'package.name',

+                           "package.name || '-' || build.version || '-' || build.release",

+                           'users.id', 'users.name', 'package.id', 'package.name', 'build.release',

+                           'build.source', 'build.start_time',

+                           "date_part('epoch', build.start_time)", 'build.state', 'build.task_id',

+                           'build.version', 'volume.id', 'volume.name'])

+ 

+     def test_non_exist_build_dict_with_strict(self):

          build = {

              'name': 'test_name',

              'version': 'test_version',
@@ -35,3 +82,66 @@ 

          with self.assertRaises(koji.GenericError) as cm:

              kojihub.get_build(build, strict=True)

          self.assertEqual('No such build: %s' % build['name'], str(cm.exception))

+         self.find_build_id.assert_called_once_with(build, strict=True)

+         self.assertEqual(len(self.queries), 0)

+ 

+     def test_build_none_exist_without_strict(self):

+         build = 'build-1-23'

+         self.find_build_id.return_value = None

+         result = kojihub.get_build(build, strict=False)

+         self.assertEqual(result, None)

+         self.find_build_id.assert_called_once_with(build, strict=False)

+         self.assertEqual(len(self.queries), 0)

+ 

+     def test_result_with_cg_id_none(self):

+         build = 11

+         self.find_build_id.return_value = build

+         self.qp_execute_one_return_value = {'cg_id': None}

+         result = kojihub.get_build(build, strict=True)

+         self.assertEqual(result, {'cg_id': None, 'cg_name': None})

+         self.find_build_id.assert_called_once_with(build, strict=True)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build'])

+         self.assertEqual(query.joins, ['events ON build.create_event = events.id',

+                                        'package on build.pkg_id = package.id',

+                                        'volume on build.volume_id = volume.id',

+                                        'users on build.owner = users.id'])

+         self.assertEqual(query.clauses, ['build.id = %(buildID)i'])

+         self.assertEqual(query.columns,

+                          ['build.id', 'build.cg_id', 'build.completion_time',

+                           "date_part('epoch', build.completion_time)", 'events.id', 'events.time',

+                           "date_part('epoch', events.time)", 'build.epoch', 'build.extra',

+                           'build.id', 'package.name',

+                           "package.name || '-' || build.version || '-' || build.release",

+                           'users.id', 'users.name', 'package.id', 'package.name', 'build.release',

+                           'build.source', 'build.start_time',

+                           "date_part('epoch', build.start_time)", 'build.state', 'build.task_id',

+                           'build.version', 'volume.id', 'volume.name'])

+ 

+     def test_result_with_cg_id(self):

+         build = 11

+         self.find_build_id.return_value = build

+         self.lookup_name.return_value = {'name': 'cg_name'}

+         self.qp_execute_one_return_value = {'cg_id': 1}

+         result = kojihub.get_build(build, strict=True)

+         self.assertEqual(result, {'cg_id': 1, 'cg_name': 'cg_name'})

+         self.find_build_id.assert_called_once_with(build, strict=True)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build'])

+         self.assertEqual(query.joins, ['events ON build.create_event = events.id',

+                                        'package on build.pkg_id = package.id',

+                                        'volume on build.volume_id = volume.id',

+                                        'users on build.owner = users.id'])

+         self.assertEqual(query.clauses, ['build.id = %(buildID)i'])

+         self.assertEqual(query.columns,

+                          ['build.id', 'build.cg_id', 'build.completion_time',

+                           "date_part('epoch', build.completion_time)", 'events.id', 'events.time',

+                           "date_part('epoch', events.time)", 'build.epoch', 'build.extra',

+                           'build.id', 'package.name',

+                           "package.name || '-' || build.version || '-' || build.release",

+                           'users.id', 'users.name', 'package.id', 'package.name', 'build.release',

+                           'build.source', 'build.start_time',

+                           "date_part('epoch', build.start_time)", 'build.state', 'build.task_id',

+                           'build.version', 'volume.id', 'volume.name'])

@@ -1,22 +1,52 @@ 

- import mock

- import unittest

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- class TestGetBuildNotification(unittest.TestCase):

+ class TestGetBuildNotification(DBQueryTestCase):

  

      def setUp(self):

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()

-         self.query = self.QueryProcessor.return_value

+         super(TestGetBuildNotification, self).setUp()

          self.exports = kojihub.RootExports()

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

      def test_empty_result_with_strict(self):

          notif_id = 1

-         self.query.executeOne.return_value = None

+         self.qp_execute_one_return_value = None

          with self.assertRaises(koji.GenericError) as cm:

              self.exports.getBuildNotification(notif_id, strict=True)

          self.assertEqual(f"No notification with ID {notif_id} found", str(cm.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build_notifications'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['id = %(id)i'])

+         self.assertEqual(query.columns,

+                          ['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])

+ 

+     def test_empty_result_without_strict(self):

+         notif_id = 1

+         self.qp_execute_one_return_value = None

+         result = self.exports.getBuildNotification(notif_id, strict=False)

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build_notifications'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['id = %(id)i'])

+         self.assertEqual(query.columns,

+                          ['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])

+ 

+     def test_valid_result(self):

+         qp_result = {'id': 1, 'user_id': 2, 'package_id': 3, 'tag_id': 4, 'success_only': True,

+                      'email': 'test-mail'}

+         notif_id = 1

+         self.qp_execute_one_return_value = qp_result

+         result = self.exports.getBuildNotification(notif_id, strict=True)

+         self.assertEqual(result, qp_result)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build_notifications'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['id = %(id)i'])

+         self.assertEqual(query.columns,

+                          ['email', 'id', 'package_id', 'success_only', 'tag_id', 'user_id'])

@@ -1,14 +1,13 @@ 

  import mock

- import unittest

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- class TestGetBuildNotificationBlock(unittest.TestCase):

+ class TestGetBuildNotificationBlock(DBQueryTestCase):

  

      def setUp(self):

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()

-         self.query = self.QueryProcessor.return_value

+         super(TestGetBuildNotificationBlock, self).setUp()

          self.exports = kojihub.RootExports()

  

      def tearDown(self):
@@ -16,7 +15,27 @@ 

  

      def test_empty_result_with_strict(self):

          notif_id = 1

-         self.query.executeOne.return_value = None

+         self.qp_execute_one_return_value = None

          with self.assertRaises(koji.GenericError) as cm:

              self.exports.getBuildNotificationBlock(notif_id, strict=True)

          self.assertEqual(f"No notification block with ID {notif_id} found", str(cm.exception))

+ 

+     def test_empty_result_without_strict(self):

+         notif_id = 1

+         self.qp_execute_one_return_value = None

+         result = self.exports.getBuildNotificationBlock(notif_id, strict=False)

+         self.assertEqual(result, None)

+ 

+     def test_valid_result(self):

+         qp_result = {'id': 1, 'user_id': 2, 'package_id': 3, 'tag_id': 4}

+         notif_id = 1

+         self.qp_execute_one_return_value = qp_result

+         result = self.exports.getBuildNotificationBlock(notif_id, strict=True)

+         self.assertEqual(result, qp_result)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build_notifications_block'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.values, {'id': 1})

+         self.assertEqual(query.clauses, ['id = %(id)i'])

+         self.assertEqual(query.columns, ['id', 'package_id', 'tag_id', 'user_id'])

@@ -1,40 +1,25 @@ 

- import unittest

- 

  import mock

  

  import kojihub

- 

- QP = kojihub.QueryProcessor

+ from .utils import DBQueryTestCase

  

  

- class TestGetBuildTargets(unittest.TestCase):

+ class TestGetBuildTargets(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetBuildTargets, self).setUp()

          self.maxDiff = None

          self.name_or_id_clause = mock.patch('kojihub.kojihub.name_or_id_clause').start()

          self.get_tag_id = mock.patch('kojihub.kojihub.get_tag_id').start()

          self.exports = kojihub.RootExports()

          self.context = mock.patch('kojihub.kojihub.context').start()

          self.cursor = mock.MagicMock()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

          self.build_target = 'build-target'

          self.build_tag_name = 'tag'

          self.dest_tag_name = 'dest-tag'

          self.build_tag_id = 1

          self.dest_tag_id = 2

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         query.executeOne = mock.MagicMock()

-         self.queries.append(query)

-         return query

- 

      def test_get_build_targets_strings(self):

          self.name_or_id_clause.return_value = '(build_target.name = %(build_target_name)s)', \

                                                {'build_target_name': 'build-target-url'}

@@ -1,33 +1,19 @@ 

  import mock

- import unittest

  

  import kojihub

+ from .utils import DBQueryTestCase

  

- QP = kojihub.QueryProcessor

  

- 

- class TestGetBuildType(unittest.TestCase):

+ class TestGetBuildType(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetBuildType, self).setUp()

          self.maxDiff = None

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

-         self.query_execute = mock.MagicMock()

          self.get_build = mock.patch('kojihub.kojihub.get_build').start()

          self.get_maven_build = mock.patch('kojihub.kojihub.get_maven_build').start()

          self.get_win_build = mock.patch('kojihub.kojihub.get_win_build').start()

          self.get_image_build = mock.patch('kojihub.kojihub.get_image_build').start()

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = self.query_execute

-         self.queries.append(query)

-         return query

- 

      def test_no_build(self):

          self.get_build.return_value = None

  
@@ -47,7 +33,7 @@ 

          self.get_win_build.return_value = typeinfo['win']

          self.get_image_build.return_value = typeinfo['image']

  

-         self.query_execute.return_value = [['new_type']]

+         self.qp_execute_return_value = [['new_type']]

  

          ret = kojihub.get_build_type('mytestbuild-1-1', strict=True)

          assert ret == typeinfo

@@ -1,33 +1,17 @@ 

- import unittest

- 

  import mock

  

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- QP = kojihub.QueryProcessor

- 

- 

- class TestGetChannel(unittest.TestCase):

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         query.executeOne = mock.MagicMock()

-         self.queries.append(query)

-         return query

+ class TestGetChannel(DBQueryTestCase):

  

      def setUp(self):

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

+         super(TestGetChannel, self).setUp()

          self.context = mock.patch('kojihub.kojihub.context').start()

          self.exports = kojihub.RootExports()

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

      def test_wrong_type_channelInfo(self):

          # dict

          channel_info = {'channel': 'val'}
@@ -54,7 +38,6 @@ 

          self.assertEqual(set(query.clauses), set(clauses))

          self.assertEqual(query.values, values)

  

- 

      def test_query_by_id(self):

          self.exports.getChannel(12345)

          self.assertEqual(len(self.queries), 1)
@@ -67,7 +50,7 @@ 

          self.assertEqual(query.values, values)

  

      def test_query_by_dict(self):

-         self.exports.getChannel({'id':12345, 'name': 'whatever'})

+         self.exports.getChannel({'id': 12345, 'name': 'whatever'})

          self.assertEqual(len(self.queries), 1)

          query = self.queries[0]

          clauses = ['(channels.id = %(channels_id)s)']

@@ -0,0 +1,34 @@ 

+ import unittest

+ 

+ import mock

+ 

+ import kojihub

+ 

+ 

+ class TestGetExternalRepo(unittest.TestCase):

+ 

+     def setUp(self):

+         self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()

+         self.read_full_inheritance = mock.patch('kojihub.kojihub.readFullInheritance').start()

+         self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     def test_valid(self):

+         self.get_tag.return_value = {'id': 123}

+         self.read_full_inheritance.return_value = [{'parent_id': 111}, {'parent_id': 112}]

+         self.get_tag_external_repos.side_effect = [

+             [{'external_repo_id': 10, 'tag_id': 123}],

+             [{'external_repo_id': 11, 'tag_id': 111}, {'external_repo_id': 12, 'tag_id': 111}],

+             [{'external_repo_id': 13, 'tag_id': 112}]]

+         result = kojihub.get_external_repo_list('test-tag')

+         self.assertEqual(result, [

+             {'external_repo_id': 10, 'tag_id': 123},

+             {'external_repo_id': 11, 'tag_id': 111}, {'external_repo_id': 12, 'tag_id': 111},

+             {'external_repo_id': 13, 'tag_id': 112}])

+         self.get_tag.assert_called_once_with('test-tag', strict=True, event=None)

+         self.read_full_inheritance.assert_called_once_with(123, None)

+         self.get_tag_external_repos.assert_has_calls(

+             [mock.call(tag_info=123, event=None), mock.call(tag_info=111, event=None),

+              mock.call(tag_info=112, event=None)])

@@ -1,16 +1,14 @@ 

- import unittest

- 

  import mock

  

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- class TestGetGroupMembers(unittest.TestCase):

+ class TestGetGroupMembers(DBQueryTestCase):

  

      def setUp(self):

-         self.context = mock.patch('kojihub.kojihub.context').start()

-         self.context.session.assertPerm = mock.MagicMock()

+         super(TestGetGroupMembers, self).setUp()

          self.get_user = mock.patch('kojihub.kojihub.get_user').start()

          self.exports = kojihub.RootExports()

  
@@ -20,3 +18,18 @@ 

          with self.assertRaises(koji.GenericError) as cm:

              self.exports.getGroupMembers(group)

          self.assertEqual("No such group: %s" % group, str(cm.exception))

+         self.assertEqual(len(self.queries), 0)

+ 

+     def test_valid(self):

+         group = 'test-group'

+         self.get_user.return_value = {'id': 23, 'usertype': 2}

+         self.exports.getGroupMembers(group)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['user_groups'])

+         self.assertEqual(query.joins, ['JOIN users ON user_groups.user_id = users.id',

+                                        'LEFT JOIN user_krb_principals'

+                                        ' ON users.id = user_krb_principals.user_id'])

+         self.assertEqual(query.clauses, ['(active = TRUE)', 'group_id = %(group_id)i'])

+         self.assertEqual(query.values, {'group_id': 23})

+         self.assertEqual(query.columns, ['id', 'array_agg(krb_principal)', 'name', 'usertype'])

@@ -0,0 +1,41 @@ 

+ import kojihub

+ from .utils import DBQueryTestCase

+ 

+ 

+ class TestGetImageArchive(DBQueryTestCase):

+ 

+     def setUp(self):

+         super(TestGetImageArchive, self).setUp()

+         self.maxDiff = None

+ 

+     def test_not_exist_image_archive(self):

+         self.qp_execute_one_return_value = {}

+         result = kojihub.get_image_archive(123)

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['image_archives'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])

+         self.assertEqual(query.values, {'archive_id': 123})

+         self.assertEqual(query.columns, ['arch', 'archive_id'])

+ 

+     def test_valid(self):

+         self.qp_execute_one_side_effect = [{'archive_id': 123, 'arch': 'arch'},

+                                            {'rpm_id': 1}]

+         result = kojihub.get_image_archive(123)

+         self.assertEqual(result, {'archive_id': 123, 'arch': 'arch', 'rootid': True})

+         self.assertEqual(len(self.queries), 2)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['image_archives'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])

+         self.assertEqual(query.values, {'archive_id': 123})

+         self.assertEqual(query.columns, ['arch', 'archive_id'])

+ 

+         query = self.queries[1]

+         self.assertEqual(query.tables, ['archive_rpm_components'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])

+         self.assertEqual(query.values, {'archive_id': 123})

+         self.assertEqual(query.columns, ['rpm_id'])

@@ -0,0 +1,64 @@ 

+ import mock

+ import koji

+ import kojihub

+ from .utils import DBQueryTestCase

+ 

+ 

+ class TestGetImageBuild(DBQueryTestCase):

+ 

+     def setUp(self):

+         super(TestGetImageBuild, self).setUp()

+         self.maxDiff = None

+         self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()

+ 

+     def test_build_id_not_found(self):

+         self.find_build_id.return_value = None

+         result = kojihub.get_image_build('test-build.1-23.1')

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 0)

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_valid(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {'build_id': 123}

+         kojihub.get_image_build('test-build.1-23.1')

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['image_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_without_result_without_strict(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {}

+         result = kojihub.get_image_build('test-build.1-23.1')

+         self.assertEqual(result, {})

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['image_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_without_result_with_strict(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {}

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.get_image_build('test-build.1-23.1', strict=True)

+         self.assertEqual('no such image build: test-build.1-23.1', str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['image_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=True)

@@ -1,12 +1,12 @@ 

- import unittest

- 

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- class TestGetLastEvent(unittest.TestCase):

+ class TestGetLastEvent(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetLastEvent, self).setUp()

          self.exports = kojihub.RootExports()

  

      def test_wrong_type_before(self):
@@ -14,3 +14,14 @@ 

          with self.assertRaises(koji.GenericError) as cm:

              self.exports.getLastEvent(before)

          self.assertEqual("Invalid type for before: %s" % type(before), str(cm.exception))

+ 

+     def test_valid(self):

+         before = 123

+         self.exports.getLastEvent(before)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['events'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ["date_part('epoch', time) < %(before)r"])

+         self.assertEqual(query.values, {'before': 123})

+         self.assertEqual(query.columns, ['id', "date_part('epoch', time)"])

@@ -1,26 +1,15 @@ 

- import unittest

- import mock

  import datetime

  import sys

  

  import kojihub

+ from .utils import DBQueryTestCase

  

- QP = kojihub.QueryProcessor

  

- 

- class TestGetLastHostUpdate(unittest.TestCase):

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.singleValue = self.query_singleValue

-         self.queries.append(query)

-         return query

+ class TestGetLastHostUpdate(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetLastHostUpdate, self).setUp()

          self.exports = kojihub.RootExports()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

-         self.query_singleValue = mock.MagicMock()

  

      def test_valid_ts(self):

          expected = 1615875554.862938
@@ -30,9 +19,16 @@ 

          else:

              dt = datetime.datetime.strptime(

                  "2021-03-16T06:19:14.862938+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")

-         self.query_singleValue.return_value = dt

+         self.qp_single_value_return_value = dt

          rv = self.exports.getLastHostUpdate(1, ts=True)

          self.assertEqual(rv, expected)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['sessions'])

+         self.assertEqual(query.joins, ['host ON sessions.user_id = host.user_id'])

+         self.assertEqual(query.clauses, ['host.id = %(hostID)i'])

+         self.assertEqual(query.values, {'hostID': 1})

+         self.assertEqual(query.columns, ['update_time'])

  

      def test_valid_datetime(self):

          if sys.version_info[1] <= 6:
@@ -41,6 +37,6 @@ 

          else:

              dt = datetime.datetime.strptime(

                  "2021-03-16T06:19:14.862938+00:00", "%Y-%m-%dT%H:%M:%S.%f%z")

-         self.query_singleValue.return_value = dt

+         self.qp_single_value_return_value = dt

          rv = self.exports.getLastHostUpdate(1)

          self.assertEqual(rv, dt)

@@ -0,0 +1,19 @@ 

+ import kojihub

+ from .utils import DBQueryTestCase

+ 

+ 

+ class TestGetMavenArchive(DBQueryTestCase):

+ 

+     def setUp(self):

+         super(TestGetMavenArchive, self).setUp()

+         self.maxDiff = None

+ 

+     def test_valid(self):

+         kojihub.get_maven_archive(123)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['maven_archives'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])

+         self.assertEqual(query.values, {'archive_id': 123})

+         self.assertEqual(query.columns, ['archive_id', 'artifact_id', 'group_id', 'version'])

@@ -1,31 +1,37 @@ 

  import mock

- import unittest

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  

- class TestGetNextRelease(unittest.TestCase):

+ class TestGetNextRelease(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetNextRelease, self).setUp()

          self.maxDiff = None

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()

          self.get_build = mock.patch('kojihub.kojihub.get_build').start()

-         self._dml = mock.patch('kojihub.kojihub._dml').start()

-         self.query = self.QueryProcessor.return_value

          self.binfo = {'name': 'name', 'version': 'version'}

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

      def test_get_next_release_new(self):

          # no previous build

-         self.query.executeOne.return_value = None

+         self.qp_execute_one_return_value = None

          result = kojihub.get_next_release(self.binfo)

          self.assertEqual(result, '1')

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['build'])

+         self.assertEqual(query.joins, ['package ON build.pkg_id = package.id'])

+         self.assertEqual(query.clauses,

+                          ['name = %(name)s', 'state in %(states)s', 'version = %(version)s'])

+         self.assertEqual(query.values, {'name': self.binfo['name'],

+                                         'version': self.binfo['version'],

+                                         'states': (1, 2, 0)

+                                         })

+         self.assertEqual(query.columns, ['build.id', 'release'])

  

      def test_get_next_release_int(self):

          for n in [1, 2, 3, 5, 8, 13, 21, 34, 55]:

-             self.query.executeOne.return_value = {'release': str(n)}

+             self.qp_execute_one_return_value = {'release': str(n)}

              result = kojihub.get_next_release(self.binfo)

              self.assertEqual(result, str(n + 1))

  
@@ -38,7 +44,7 @@ 

              ['20211105.nightly.7', '20211105.nightly.8'],

          ]

          for a, b in data:

-             self.query.executeOne.return_value = {'release': a}

+             self.qp_execute_one_return_value = {'release': a}

              result = kojihub.get_next_release(self.binfo)

              self.assertEqual(result, b)

  
@@ -52,9 +58,10 @@ 

              "1.2.fc23",

          ]

          for val in data:

-             self.query.executeOne.return_value = {'release': val}

-             with self.assertRaises(koji.BuildError):

+             self.qp_execute_one_return_value = {'release': val}

+             with self.assertRaises(koji.BuildError) as ex:

                  kojihub.get_next_release(self.binfo)

+             self.assertEqual(f'Unable to increment release value: {val}', str(ex.exception))

  

      def test_get_next_release_bad_incr(self):

          data = [

@@ -5,16 +5,9 @@ 

  import kojihub

  

  QP = kojihub.QueryProcessor

- IP = kojihub.InsertProcessor

- UP = kojihub.UpdateProcessor

  

  

  class TestGetNotificationRecipients(unittest.TestCase):

-     def getInsert(self, *args, **kwargs):

-         insert = IP(*args, **kwargs)

-         insert.execute = mock.MagicMock()

-         self.inserts.append(insert)

-         return insert

  

      def getQuery(self, *args, **kwargs):

          query = QP(*args, **kwargs)
@@ -22,12 +15,6 @@ 

          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.kojihub.context').start()

          self.context.opts = {
@@ -38,12 +25,6 @@ 

          self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

                                           side_effect=self.getQuery).start()

          self.queries = []

-         self.InsertProcessor = mock.patch('kojihub.kojihub.InsertProcessor',

-                                           side_effect=self.getInsert).start()

-         self.inserts = []

-         self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

-                                           side_effect=self.getUpdate).start()

-         self.updates = []

          self.readPackageList = mock.patch('kojihub.kojihub.readPackageList').start()

          self.get_user = mock.patch('kojihub.kojihub.get_user').start()

  
@@ -75,15 +56,6 @@ 

          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)

-         '''

          self.readPackageList.assert_not_called()

  

      def test_get_notification_recipients_build_without_tag(self):
@@ -128,6 +100,34 @@ 

  

          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
@@ -135,8 +135,9 @@ 

          state = koji.BUILD_STATES['CANCELED']

          self.queries = []

  

-         with self.assertRaises(koji.GenericError):

+         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()

  
@@ -256,3 +257,44 @@ 

          ])

          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()

@@ -0,0 +1,64 @@ 

+ import kojihub

+ import mock

+ import unittest

+ 

+ QP = kojihub.QueryProcessor

+ 

+ 

+ class TestGetReadyHosts(unittest.TestCase):

+ 

+     def setUp(self):

+         self.maxDiff = None

+         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

+                                          side_effect=self.getQuery).start()

+         self.queries = []

+         self.query_execute = mock.MagicMock()

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     def getQuery(self, *args, **kwargs):

+         query = QP(*args, **kwargs)

+         query.execute = self.query_execute

+         self.queries.append(query)

+         return query

+ 

+     def test_valid(self):

+         hosts = [{'host.id': 1, 'name': 'hostname', 'arches': 'arch123', 'task_load': 0,

+                   'capacity': 3},

+                  {'host.id': 2, 'name': 'hostname-2', 'arches': 'arch123', 'task_load': 0,

+                   'capacity': 3}]

+         expected_res = [{'host.id': 1, 'name': 'hostname', 'arches': 'arch123', 'task_load': 0,

+                          'capacity': 3, 'channels': [1]},

+                         {'host.id': 2, 'name': 'hostname-2', 'arches': 'arch123', 'task_load': 0,

+                          'capacity': 3, 'channels': [2, 3]}

+                         ]

+         self.query_execute.side_effect = [hosts, [{'channel_id': 1}],

+                                           [{'channel_id': 2}, {'channel_id': 3}]]

+         result = kojihub.get_ready_hosts()

+         self.assertEqual(result, expected_res)

+         self.assertEqual(len(self.queries), 3)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['host'])

+         self.assertEqual(query.joins, ['sessions USING (user_id)',

+                                        'host_config ON host.id = host_config.host_id'])

+         self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'expired IS FALSE',

+                                          'master IS NULL', 'ready IS TRUE',

+                                          "update_time > NOW() - '5 minutes'::interval"])

+         self.assertEqual(query.values, {})

+         self.assertEqual(query.columns, ['arches', 'capacity', 'host.id', 'name', 'task_load'])

+ 

+         query = self.queries[1]

+         self.assertEqual(query.tables, ['host_channels'])

+         self.assertEqual(query.joins, ['channels ON host_channels.channel_id = channels.id'])

+         self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'host_id=%(id)s'])

+         self.assertEqual(query.values, hosts[0])

+         self.assertEqual(query.columns, ['channel_id'])

+ 

+         query = self.queries[2]

+         self.assertEqual(query.tables, ['host_channels'])

+         self.assertEqual(query.joins, ['channels ON host_channels.channel_id = channels.id'])

+         self.assertEqual(query.clauses, ['active IS TRUE', 'enabled IS TRUE', 'host_id=%(id)s'])

+         self.assertEqual(query.values, hosts[1])

+         self.assertEqual(query.columns, ['channel_id'])

@@ -1,34 +1,19 @@ 

- import mock

- import unittest

- 

  import koji

  import kojihub

+ import mock

+ from .utils import DBQueryTestCase

  

- QP = kojihub.QueryProcessor

- 

- 

- class TestGetSessionInfo(unittest.TestCase):

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         self.queries.append(query)

-         return query

  

+ class TestGetSessionInfo(DBQueryTestCase):

      def setUp(self):

+         super(TestGetSessionInfo, self).setUp()

          self.context = mock.patch('kojihub.kojihub.context').start()

          self.exports = kojihub.RootExports()

- 

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

          self.context.session.hasPerm = mock.MagicMock()

          self.get_user = mock.patch('kojihub.kojihub.get_user').start()

          self.userinfo = {'id': 123, 'name': 'testuser'}

          self.exports.getLoggedInUser = mock.MagicMock()

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

      def test_get_session_info_not_logged(self):

          self.context.session.logged_in = False

          result = self.exports.getSessionInfo()
@@ -94,6 +79,7 @@ 

      def test_get_session_info_details(self):

          self.context.session.logged_in = True

          self.context.session.hasPerm.return_value = True

+         self.qp_execute_one_return_value = {'hostip': '10.0.0.0', 'id': 123}

          self.exports.getSessionInfo(details=True)

          self.assertEqual(len(self.queries), 1)

          query = self.queries[0]

@@ -1,39 +1,24 @@ 

- import unittest

- 

  import mock

  

  import kojihub

- 

- QP = kojihub.QueryProcessor

+ from .utils import DBQueryTestCase

  

  

- class TestGetTagExternalRepos(unittest.TestCase):

+ class TestGetTagExternalRepos(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetTagExternalRepos, self).setUp()

          self.maxDiff = None

          self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()

          self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()

          self.exports = kojihub.RootExports()

          self.context = mock.patch('kojihub.kojihub.context').start()

          self.cursor = mock.MagicMock()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

          self.build_tag = 'tag'

          self.repo = 'repo_name'

          self.build_tag_info = {'id': 111, 'name': self.build_tag}

          self.repo_info = {'id': 123, 'name': self.repo}

  

-     def tearDown(self):

-         mock.patch.stopall()

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         query.executeOne = mock.MagicMock()

-         self.queries.append(query)

-         return query

- 

      def test_valid(self):

          self.get_tag.return_value = self.build_tag_info

          self.get_external_repo.return_value = self.repo_info

@@ -1,34 +1,22 @@ 

  import mock

- import unittest

  

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

  QP = kojihub.QueryProcessor

  

  

- class TestGetTaskChildren(unittest.TestCase):

+ class TestGetTaskChildren(DBQueryTestCase):

      def setUp(self):

+         super(TestGetTaskChildren, self).setUp()

          self.exports = kojihub.RootExports()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         query.executeOne = mock.MagicMock()

-         query.singleValue = mock.MagicMock()

-         self.queries.append(query)

-         return query

  

      def tearDown(self):

          mock.patch.stopall()

  

      def test_get_task_children_non_existing(self):

-         q = self.getQuery()

-         q.execute.return_value = []

-         self.QueryProcessor.side_effect = [q]

+         self.qp_execute_return_value = []

  

          r = self.exports.getTaskChildren(1000)

  
@@ -36,18 +24,14 @@ 

  

      def test_get_task_children_non_existing_strict(self):

          # get task info

-         q = self.getQuery()

-         q.executeOne.side_effect = koji.GenericError

-         self.QueryProcessor.side_effect = [q]

+         self.qp_execute_one_side_effect = koji.GenericError

  

          with self.assertRaises(koji.GenericError):

              self.exports.getTaskChildren(1000, strict=True)

  

      def test_get_task_children(self):

          children = [{'id': 1}]

-         q = self.getQuery()

-         q.execute.return_value = children

-         self.QueryProcessor.side_effect = [q]

+         self.qp_execute_return_value = children

  

          r = self.exports.getTaskChildren(1000)

  

file modified
+30 -19
@@ -1,30 +1,19 @@ 

- import unittest

- 

  import mock

+ import unittest

  

  import koji

  import kojihub

+ from .utils import DBQueryTestCase

  

- QP = kojihub.QueryProcessor

  

- 

- class TestGetUser(unittest.TestCase):

+ class TestGetUser(DBQueryTestCase):

  

      def setUp(self):

+         super(TestGetUser, self).setUp()

          self.exports = kojihub.RootExports()

          self.context = mock.patch('kojihub.kojihub.context').start()

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                                          side_effect=self.getQuery).start()

-         self.queries = []

- 

-     def getQuery(self, *args, **kwargs):

-         query = QP(*args, **kwargs)

-         query.execute = mock.MagicMock()

-         self.queries.append(query)

-         return query

- 

-     def tearDown(self):

-         mock.patch.stopall()

+         self.list_user_krb_principals = mock.patch(

+             'kojihub.kojihub.list_user_krb_principals').start()

  

      def test_wrong_format_user_info(self):

          userinfo = ['test-user']
@@ -71,9 +60,14 @@ 

                                         'users.id = user_krb_principals.user_id'])

          self.assertEqual(query.values, {'info': userinfo})

  

-     def test_userinfo_dict(self):

+     def test_userinfo_dict_with_krbs(self):

          userinfo = {'id': 123456, 'name': 'test-user', 'krb_principal': 'test-krb@krb.com'}

-         kojihub.get_user(userinfo, krb_princs=False)

+         self.qp_execute_one_return_value = {'id': 123456, 'name': 'test-user',

+                                             'status': 1, 'usertype': 1}

+         self.list_user_krb_principals.return_value = 'test-krb@krb.com'

+         result = kojihub.get_user(userinfo, krb_princs=True)

+         self.assertEqual(result, {'id': 123456, 'krb_principals': 'test-krb@krb.com',

+                                   'name': 'test-user', 'status': 1, 'usertype': 1})

          self.assertEqual(len(self.queries), 1)

          query = self.queries[0]

          str(query)
@@ -86,6 +80,23 @@ 

                                         'users.id = user_krb_principals.user_id'])

          self.assertEqual(query.values, userinfo)

  

+     def test_userinfo_int_user_not_exist_and_strict(self):

+         userinfo = {'id': 123456}

+         self.qp_execute_one_return_value = {}

+         with self.assertRaises(koji.GenericError) as cm:

+             kojihub.get_user(userinfo['id'], strict=True, krb_princs=False)

+         self.assertEqual(f"No such user: {userinfo['id']}", str(cm.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         str(query)

+         self.assertEqual(query.tables, ['users'])

+         columns = ['id', 'name', 'status', 'usertype']

+         self.assertEqual(set(query.columns), set(columns))

+         self.assertEqual(query.clauses, ['users.id = %(id)i'])

+         self.assertEqual(query.joins, ['LEFT JOIN user_krb_principals ON '

+                                        'users.id = user_krb_principals.user_id'])

+         self.assertEqual(query.values, userinfo)

+ 

  

  class TestGetUserByKrbPrincipal(unittest.TestCase):

      def setUp(self):

@@ -9,8 +9,9 @@ 

  class TestGetVerifyClass(unittest.TestCase):

  

      def test_get_verify_class_generic_error(self):

-         with self.assertRaises(GenericError):

+         with self.assertRaises(GenericError) as ex:

              kojihub.get_verify_class('not_a_real_value')

+         self.assertEqual('Unsupported verify type: not_a_real_value', str(ex.exception))

  

      def test_get_verify_class_is_none(self):

          kojihub.get_verify_class(None) is None
@@ -23,4 +24,3 @@ 

  

      def test_get_verify_class_is_sha256(self):

          kojihub.get_verify_class('sha256') is hashlib.sha256

- 

@@ -18,7 +18,7 @@ 

              self.exports.getVolume(volume, strict=True)

          self.assertEqual("No such volume: %s" % volume, str(cm.exception))

  

-     def test_non_exist_volume_with_strict(self):

+     def test_non_exist_volume_without_strict(self):

          volume = ['test-volume']

          self.lookup_name.return_value = None

          result = self.exports.getVolume(volume)

@@ -0,0 +1,19 @@ 

+ import kojihub

+ from .utils import DBQueryTestCase

+ 

+ 

+ class TestGetWinArchive(DBQueryTestCase):

+ 

+     def setUp(self):

+         super(TestGetWinArchive, self).setUp()

+         self.maxDiff = None

+ 

+     def test_valid(self):

+         kojihub.get_win_archive(123)

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['win_archives'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['archive_id = %(archive_id)i'])

+         self.assertEqual(query.values, {'archive_id': 123})

+         self.assertEqual(query.columns, ['archive_id', 'flags', 'platforms', 'relpath'])

@@ -0,0 +1,64 @@ 

+ import mock

+ import koji

+ import kojihub

+ from .utils import DBQueryTestCase

+ 

+ 

+ class TestGetWinBuild(DBQueryTestCase):

+ 

+     def setUp(self):

+         super(TestGetWinBuild, self).setUp()

+         self.maxDiff = None

+         self.find_build_id = mock.patch('kojihub.kojihub.find_build_id').start()

+ 

+     def test_build_id_not_found(self):

+         self.find_build_id.return_value = None

+         result = kojihub.get_win_build('test-build.1-23.1')

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 0)

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_valid(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {'build_id': 123}

+         kojihub.get_win_build('test-build.1-23.1')

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['win_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id', 'platform'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_without_result_without_strict(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {}

+         result = kojihub.get_win_build('test-build.1-23.1')

+         self.assertEqual(result, {})

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['win_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id', 'platform'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=False)

+ 

+     def test_without_result_with_strict(self):

+         self.find_build_id.return_value = 123

+         self.qp_execute_one_return_value = {}

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.get_win_build('test-build.1-23.1', strict=True)

+         self.assertEqual('no such Windows build: test-build.1-23.1', str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['win_builds'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(build_id)i'])

+         self.assertEqual(query.values, {'build_id': 123})

+         self.assertEqual(query.columns, ['build_id', 'platform'])

+ 

+         self.find_build_id.assert_called_once_with('test-build.1-23.1', strict=True)

@@ -6,18 +6,26 @@ 

  

  QP = kojihub.QueryProcessor

  UP = kojihub.UpdateProcessor

+ DP = kojihub.DeleteProcessor

  

  

  class TestRecycleBuild(unittest.TestCase):

      def setUp(self):

-         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor').start()

+         self.maxDiff = None

+         self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

+                                          side_effect=self.getQuery).start()

+         self.queries = []

+         self.query_execute = mock.MagicMock()

          self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

                                            side_effect=self.getUpdate).start()

-         self._dml = mock.patch('kojihub.kojihub._dml').start()

          self.run_callbacks = mock.patch('koji.plugin.run_callbacks').start()

          self.rmtree = mock.patch('koji.util.rmtree').start()

          self.exists = mock.patch('os.path.exists').start()

          self.updates = []

+         self.DeleteProcessor = mock.patch('kojihub.kojihub.DeleteProcessor',

+                                           side_effect=self.getDelete).start()

+         self.deletes = []

+         self.get_build = mock.patch('kojihub.kojihub.get_build').start()

  

      def tearDown(self):

          mock.patch.stopall()
@@ -28,9 +36,21 @@ 

          self.updates.append(update)

          return update

  

+     def getQuery(self, *args, **kwargs):

+         query = QP(*args, **kwargs)

+         query.execute = self.query_execute

+         self.queries.append(query)

+         return query

+ 

+     def getDelete(self, *args, **kwargs):

+         delete = DP(*args, **kwargs)

+         delete.execute = mock.MagicMock()

+         self.deletes.append(delete)

+         return delete

+ 

      # Basic old and new build infos

      old = {'id': 2,

-            'state': 0,

+            'state': 3,

             'task_id': None,

             'epoch': None,

             'name': 'GConf2',
@@ -44,7 +64,7 @@ 

             'cg_id': None,

             'volume_id': 0,

             'volume_name': 'DEFAULT'}

-     new = {'state': 0,

+     new = {'state': 3,

             'name': 'GConf2',

             'version': '3.2.6',

             'release': '15.fc23',
@@ -58,97 +78,201 @@ 

             'cg_id': None,

             'volume_id': 0}

  

-     def test_recycle_building(self):

+     def test_build_already_in_progress(self):

          new = self.new.copy()

          old = self.old.copy()

          old['state'] = new['state'] = koji.BUILD_STATES['BUILDING']

-         old['task_id'] = new['task_id'] = 137

-         kojihub.recycle_build(old, new)

-         self.UpdateProcessor.assert_not_called()

-         self.QueryProcessor.assert_not_called()

-         self._dml.assert_not_called()

-         self.run_callbacks.assert_not_called()

- 

-     def run_fail(self, old, new):

-         try:

+         old['task_id'] = 137

+         with self.assertRaises(koji.GenericError) as ex:

              kojihub.recycle_build(old, new)

-         except koji.GenericError:

-             pass

-         else:

-             raise Exception("expected koji.GenericError")

-         self.UpdateProcessor.assert_not_called()

-         self._dml.assert_not_called()

+         self.assertEqual(f"Build already in progress (task {old['task_id']})", str(ex.exception))

+         self.assertEqual(len(self.queries), 0)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

+ 

+         self.get_build.assert_not_called()

          self.run_callbacks.assert_not_called()

  

-     def test_recycle_building_bad(self):

+     def test_build_already_in_progress_same_task_id(self):

          new = self.new.copy()

          old = self.old.copy()

          old['state'] = new['state'] = koji.BUILD_STATES['BUILDING']

+         old['task_id'] = new['task_id'] = 137

+         result = kojihub.recycle_build(old, new)

+         self.assertEqual(result, None)

+         self.assertEqual(len(self.queries), 0)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

+ 

+         self.get_build.assert_not_called()

+         self.run_callbacks.assert_not_called()

+ 

+     def test_not_in_failed_or_canceled_state(self):

+         old = self.old.copy()

+         old['state'] = koji.BUILD_STATES['COMPLETE']

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.recycle_build(old, self.new)

+         self.assertEqual(f"Build already exists (id={old['id']}, state=COMPLETE): {self.new}",

+                          str(ex.exception))

+         self.assertEqual(len(self.queries), 0)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

+ 

+         self.get_build.assert_not_called()

+         self.run_callbacks.assert_not_called()

+ 

+     def test_tag_activity_already_exists(self):

+         old = self.old.copy()

          old['task_id'] = 137

-         new['task_id'] = 200

-         self.run_fail(old, new)

-         self.QueryProcessor.assert_not_called()

+         self.query_execute.return_value = [{'tag_id': 123}]

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.recycle_build(old, self.new)

+         self.assertEqual("Build already exists. Unable to recycle, has tag history",

+                          str(ex.exception))

+         self.assertEqual(len(self.queries), 1)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

  

-     def test_recycle_states_good(self):

-         for state in 'FAILED', 'CANCELED':

-             self.check_recycle_states_good, koji.BUILD_STATES[state]

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['tag_listing'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['tag_id'])

  

-     def check_recycle_states_good(self, state):

-         new = self.new.copy()

+         self.get_build.assert_not_called()

+         self.run_callbacks.assert_not_called()

+ 

+     def test_rpm_activity_already_exists(self):

+         old = self.old.copy()

+         old['task_id'] = 137

+         self.query_execute.side_effect = [[], [{'id': 1}]]

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.recycle_build(old, self.new)

+         self.assertEqual("Build already exists. Unable to recycle, has rpm data",

+                          str(ex.exception))

+ 

+         self.assertEqual(len(self.queries), 2)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['tag_listing'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['tag_id'])

+ 

+         query = self.queries[1]

+         self.assertEqual(query.tables, ['rpminfo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['id'])

+ 

+         self.get_build.assert_not_called()

+         self.run_callbacks.assert_not_called()

+ 

+     def test_archive_activity_already_exists(self):

+         old = self.old.copy()

+         old['task_id'] = 137

+         self.query_execute.side_effect = [[], [], [{'id': 11}]]

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.recycle_build(old, self.new)

+         self.assertEqual("Build already exists. Unable to recycle, has archive data",

+                          str(ex.exception))

+ 

+         self.assertEqual(len(self.queries), 3)

+         self.assertEqual(len(self.updates), 0)

+         self.assertEqual(len(self.deletes), 0)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['tag_listing'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['tag_id'])

+ 

+         query = self.queries[1]

+         self.assertEqual(query.tables, ['rpminfo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['id'])

+ 

+         query = self.queries[2]

+         self.assertEqual(query.tables, ['archiveinfo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['id'])

+ 

+         self.get_build.assert_not_called()

+         self.run_callbacks.assert_not_called()

+ 

+     def test_valid(self):

          old = self.old.copy()

-         old['state'] = state

-         new['state'] = koji.BUILD_STATES['BUILDING']

-         old['task_id'] = 99

-         new['task_id'] = 137

-         query = self.QueryProcessor.return_value

-         # for all checks

-         query.execute.return_value = []

-         # for getBuild

-         query.executeOne.return_value = old

-         self.run_pass(old, new)

- 

-     def run_pass(self, old, new):

+         new = self.new.copy()

+         old['task_id'] = new['task_id'] = 137

+         self.query_execute.side_effect = [[], [], []]

+         self.get_build.return_value = {'build_id': 2, 'name': 'GConf2', 'version': '3.2.6',

+                                        'release': '15.fc23'}

+ 

          kojihub.recycle_build(old, new)

-         self.UpdateProcessor.assert_called_once()

+ 

+         self.assertEqual(len(self.queries), 3)

+         self.assertEqual(len(self.updates), 1)

+         self.assertEqual(len(self.deletes), 4)

+ 

+         query = self.queries[0]

+         self.assertEqual(query.tables, ['tag_listing'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['tag_id'])

+ 

+         query = self.queries[1]

+         self.assertEqual(query.tables, ['rpminfo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['id'])

+ 

+         query = self.queries[2]

+         self.assertEqual(query.tables, ['archiveinfo'])

+         self.assertEqual(query.joins, None)

+         self.assertEqual(query.clauses, ['build_id = %(id)s'])

+         self.assertEqual(query.values, old)

+         self.assertEqual(query.columns, ['id'])

+ 

+         delete = self.deletes[0]

+         self.assertEqual(delete.table, 'maven_builds')

+         self.assertEqual(delete.clauses, ['build_id = %(id)i'])

+         self.assertEqual(delete.values, old)

+ 

+         delete = self.deletes[1]

+         self.assertEqual(delete.table, 'win_builds')

+         self.assertEqual(delete.clauses, ['build_id = %(id)i'])

+         self.assertEqual(delete.values, old)

+ 

+         delete = self.deletes[2]

+         self.assertEqual(delete.table, 'image_builds')

+         self.assertEqual(delete.clauses, ['build_id = %(id)i'])

+         self.assertEqual(delete.values, old)

+ 

+         delete = self.deletes[3]

+         self.assertEqual(delete.table, 'build_types')

+         self.assertEqual(delete.clauses, ['build_id = %(id)i'])

+         self.assertEqual(delete.values, old)

+ 

          update = self.updates[0]

-         assert update.table == 'build'

+         self.assertEqual(update.table, 'build')

+         self.assertEqual(update.values, new)

          for key in ['state', 'task_id', 'owner', 'start_time',

                      'completion_time', 'epoch']:

              assert update.data[key] == new[key]

-         assert update.rawdata == {'create_event': 'get_event()'}

-         assert update.clauses == ['id=%(id)s']

-         assert update.values['id'] == old['id']

- 

-     def test_recycle_states_bad(self):

-         for state in 'BUILDING', 'COMPLETE', 'DELETED':

-             self.check_recycle_states_bad, koji.BUILD_STATES[state]

+         self.assertEqual(update.rawdata, {'create_event': 'get_event()'})

+         self.assertEqual(update.clauses, ['id=%(id)s'])

  

-     def check_recycle_states_bad(self, state):

-         new = self.new.copy()

-         old = self.old.copy()

-         old['state'] = state

-         new['state'] = koji.BUILD_STATES['BUILDING']

-         old['task_id'] = 99

-         new['task_id'] = 137

-         self.run_fail(old, new)

-         self.QueryProcessor.assert_not_called()

- 

-     def test_recycle_query_bad(self):

-         vlists = [

-             [[], [], True],

-             [True, [], []],

-             [[], True, []],

-         ]

-         for values in vlists:

-             self.check_recycle_query_bad, values

- 

-     def check_recycle_query_bad(self, values):

-         new = self.new.copy()

-         old = self.old.copy()

-         old['state'] = koji.BUILD_STATES['FAILED']

-         new['state'] = koji.BUILD_STATES['BUILDING']

-         old['task_id'] = 99

-         new['task_id'] = 137

-         query = self.QueryProcessor.return_value

-         query.execute.side_effect = values

-         self.run_fail(old, new)

+         self.get_build.assert_called_once_with(new['id'], strict=True)

+         self.assertEqual(self.run_callbacks.call_count, 2)

@@ -0,0 +1,72 @@ 

+ import unittest

+ import mock

+ 

+ import koji

+ import kojihub

+ 

+ UP = kojihub.UpdateProcessor

+ 

+ 

+ class TestRemoveExternalRepoFromTag(unittest.TestCase):

+     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.kojihub.context').start()

+         self.context.session.assertPerm = mock.MagicMock()

+         self.context_db = mock.patch('kojihub.db.context').start()

+         self.context_db.session.assertLogin = mock.MagicMock()

+         self.UpdateProcessor = mock.patch('kojihub.kojihub.UpdateProcessor',

+                                           side_effect=self.getUpdate).start()

+         self.updates = []

+         self.get_tag = mock.patch('kojihub.kojihub.get_tag').start()

+         self.get_tag_external_repos = mock.patch('kojihub.kojihub.get_tag_external_repos').start()

+         self.get_external_repo = mock.patch('kojihub.kojihub.get_external_repo').start()

+ 

+     def tearDown(self):

+         mock.patch.stopall()

+ 

+     def test_not_exist_tag_external_repo(self):

+         tag_info = {'id': 22, 'name': 'test-tag'}

+         repo_info = {'id': 1, 'name': 'test-repo'}

+         self.get_tag.return_value = tag_info

+         self.get_external_repo.return_value = repo_info

+         self.get_tag_external_repos.return_value = None

+         with self.assertRaises(koji.GenericError) as ex:

+             kojihub.remove_external_repo_from_tag(tag_info['name'], repo_info['name'])

+         self.assertEqual(

+             f'external repo {repo_info["name"]} not associated with tag {tag_info["name"]}',

+             str(ex.exception))

+ 

+         self.assertEqual(len(self.updates), 0)

+         self.get_tag.assert_called_once_with(tag_info['name'], strict=True)

+         self.get_external_repo.assert_called_once_with(repo_info['name'], strict=True)

+         self.get_tag_external_repos.assert_called_once_with(

+             tag_info=tag_info['id'], repo_info=repo_info['id'])

+ 

+     def test_valid(self):

+         tag_info = {'id': 22, 'name': 'test-tag'}

+         repo_info = {'id': 1, 'name': 'test-repo'}

+         self.context_db.event_id = 42

+         self.context_db.session.user_id = 23

+         self.get_tag.return_value = tag_info

+         self.get_external_repo.return_value = repo_info

+         self.get_tag_external_repos.return_value = [{'id': repo_info['id'],

+                                                      'tag_id': tag_info['id']}]

+         kojihub.remove_external_repo_from_tag(tag_info['name'], repo_info['name'])

+ 

+         self.assertEqual(len(self.updates), 1)

+         update = self.updates[0]

+         self.assertEqual(update.table, 'tag_external_repos')

+         self.assertEqual(update.clauses, ["tag_id = %(tag_id)i", "external_repo_id = %(repo_id)i",

+                                           'active = TRUE'])

+         self.assertEqual(update.data, {'revoke_event': 42, 'revoker_id': 23})

+         self.assertEqual(update.rawdata, {'active': 'NULL'})

+ 

+         self.get_tag.assert_called_once_with(tag_info['name'], strict=True)

+         self.get_external_repo.assert_called_once_with(repo_info['name'], strict=True)

+         self.get_tag_external_repos.assert_called_once_with(

+             tag_info=tag_info['id'], repo_info=repo_info['id'])

file modified
+9 -2
@@ -12,8 +12,11 @@ 

          mock.patch.stopall()

          self.qp_execute_return_value = []

          self.qp_execute_side_effect = None

+         self.qp_execute_one_return_value = []

+         self.qp_execute_one_side_effect = None

+         self.qp_single_value_return_value = None

          self.QueryProcessor = mock.patch('kojihub.kojihub.QueryProcessor',

-                 side_effect=self.get_query).start()

+                                          side_effect=self.get_query).start()

          self.queries = []

  

      def tearDown(self):
@@ -28,6 +31,11 @@ 

          query.execute = mock.MagicMock()

          query.execute.return_value = self.qp_execute_return_value

          query.execute.side_effect = self.qp_execute_side_effect

+         query.executeOne = mock.MagicMock()

+         query.executeOne.return_value = self.qp_execute_one_return_value

+         query.executeOne.side_effect = self.qp_execute_one_side_effect

+         query.singleValue = mock.MagicMock()

+         query.singleValue.return_value = self.qp_single_value_return_value

          self.queries.append(query)

          return query

  
@@ -42,4 +50,3 @@ 

      def assertQueriesEqual(self, arglist):

          for i, query in enumerate(self.queries):

              self.assertQueryEqual(query, **arglist[i])

- 

rebased onto bbd841c

a year ago

Commit a481729 fixes this pull-request

Pull-Request has been merged by tkopecek

a year ago
Metadata
Changes Summary 36
+35 -0
file changed
tests/test_hub/test_create_external_repo.py
+33 -2
file changed
tests/test_hub/test_delete_build_target.py
+55
file added
tests/test_hub/test_delete_external_repo.py
+102 -0
file changed
tests/test_hub/test_edit_external_repo.py
+2 -3
file changed
tests/test_hub/test_getPackageID.py
+46 -17
file changed
tests/test_hub/test_getRPM.py
+19 -18
file changed
tests/test_hub/test_getRPMDeps.py
+1 -1
file changed
tests/test_hub/test_get_active_repos.py
+179 -9
file changed
tests/test_hub/test_get_archive_type.py
+113 -3
file changed
tests/test_hub/test_get_build.py
+39 -9
file changed
tests/test_hub/test_get_build_notification.py
+24 -5
file changed
tests/test_hub/test_get_build_notification_block.py
+3 -18
file changed
tests/test_hub/test_get_build_targets.py
+4 -18
file changed
tests/test_hub/test_get_build_type.py
+4 -21
file changed
tests/test_hub/test_get_channel.py
+34
file added
tests/test_hub/test_get_external_repo_list.py
+18 -5
file changed
tests/test_hub/test_get_group_members.py
+41
file added
tests/test_hub/test_get_image_archive.py
+64
file added
tests/test_hub/test_get_image_build.py
+14 -3
file changed
tests/test_hub/test_get_last_event.py
+12 -16
file changed
tests/test_hub/test_get_last_host_update.py
+19
file added
tests/test_hub/test_get_maven_archive.py
+20 -13
file changed
tests/test_hub/test_get_next_release.py
+71 -29
file changed
tests/test_hub/test_get_notification_recipients.py
+64
file added
tests/test_hub/test_get_ready_hosts.py
+5 -19
file changed
tests/test_hub/test_get_session_info.py
+3 -18
file changed
tests/test_hub/test_get_tag_external_repos.py
+6 -22
file changed
tests/test_hub/test_get_task_children.py
+30 -19
file changed
tests/test_hub/test_get_user.py
+2 -2
file changed
tests/test_hub/test_get_verify_class.py
+1 -1
file changed
tests/test_hub/test_get_volume.py
+19
file added
tests/test_hub/test_get_win_archive.py
+64
file added
tests/test_hub/test_get_win_build.py
+203 -79
file changed
tests/test_hub/test_recycle_build.py
+72
file added
tests/test_hub/test_remove_external_repo_from_tag.py
+9 -2
file changed
tests/test_hub/utils.py