python3-devel \

    python3-librepo \

    python3-pip \

    python3-rpm \

    python3-tox \

    redhat-rpm-config \

    openssl-devel \

    python3-devel \

    python3-pip \

    python3-rpm \

    python3-tox \

    redhat-rpm-config \

    openssl-devel \

    python3-devel \

    python3-pip \

    python3-rpm \

    python3-tox \

    redhat-rpm-config \

    openssl-devel \

    python3-devel \

    python3-pip \

    python3-rpm \

    python3-tox \

    redhat-rpm-config \

    openssl-devel \

    python3-devel \

    python3-pip \

    python3-rpm \

    python3-tox \

    redhat-rpm-config \

   * ``python3-dateutil``

   * ``python3-mock``

   * ``python3-multilib``

   * ``python3-psycopg2``

   * ``python3-qpid-proton``

   * ``python3-requests``

  mock<=2.0.0

  requests-mock

  coverage

  from __future__ import absolute_import

  import inspect

  import mock

  import unittest

  import koji

  import koji.tasks

          # simple case

          ret = koji.tasks.parse_task_params('sleep', [4])

          # bad args

          with self.assertRaises(koji.ParameterError):

          # bad method

          with self.assertRaises(TypeError):

          # new style

          params = {'__method__': 'hello', 'n': 1}

              if not h_class:

                  missing.append(method)

                  continue

              # for the methods we have, at least one of the signatures should

              # match

  import sys

  import unittest

  import json

  import mock

  import six

  try:

      import libcomps

          session.groupListAdd.assert_not_called()

      @mock.patch('sys.stdout', new_callable=six.StringIO)

          if libcomps is None:

          comps_file = os.path.dirname(__file__) + '/data/comps-example.xml'

          stdout_file = os.path.dirname(

              __file__) + '/data/comps-example.libcomps.out'

              stdout)

      @mock.patch('sys.stdout', new_callable=six.StringIO)

          if libcomps is None:

          comps_file = os.path.dirname(__file__) + '/data/comps-sample.xml'

          stdout_file = os.path.dirname(

              __file__) + '/data/comps-sample.libcomps.out'

      @mock.patch('sys.stdout', new_callable=six.StringIO)

      @mock.patch('koji_cli.commands.libcomps', new=None)

      @mock.patch('koji_cli.commands.yumcomps', create=True, new=yumcomps)

          if yumcomps is None:

          comps_file = os.path.dirname(__file__) + '/data/comps-example.xml'

          stdout_file = os.path.dirname(

              __file__) + '/data/comps-example.yumcomps.out'

      @mock.patch('sys.stdout', new_callable=six.StringIO)

      @mock.patch('koji_cli.commands.libcomps', new=None)

      @mock.patch('koji_cli.commands.yumcomps', create=True, new=yumcomps)

          if yumcomps is None:

          comps_file = os.path.dirname(__file__) + '/data/comps-sample.xml'

          stdout_file = os.path.dirname(

              __file__) + '/data/comps-sample.yumcomps.out'

      def __run_parseTask_test(self, method, params, expect):

          self.session.getTaskRequest.return_value = params

          lines = _parseTaskParams(self.session, method, 1, '/mnt/koji')

      def test_error_with_param(self):

          params = []

      def test_wrapperRPM(self):

          target = 'test-target'

          expect = ["Spec File URL: %s" % params[0]]

          expect.append("Build Tag: %s" % params[1]['name'])

          expect.append("Build: %s" % self.build_templ['nvr'])

      def test_chainmaven(self):

          params = [{

          expect = ["Builds:"]

          for pkg, opt in params[0].items():

              expect.append("  %s" % pkg)

          params = [

              [1, 2, 3, 4],        # dependant task ids

              [

              ]

          ]

          expect = ["Dependant Tasks: %s" % ", ".join([str(dep) for dep in params[0]])]

      def test_chainbuild(self):

          params = [[

          expect = ["Build Groups:"]

          for i, grp in enumerate(params[0]):

          expect.append("Build Target: %s" % params[1])

          expect.append("Options:")

          expect.append("  extra: f26-pre-release")

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

              _printTaskInfo(session, task_id, '/')

  class TestTaskInfo(utils.CliTestCase):

  from __future__ import absolute_import

  import unittest

  from six.moves import range

  from koji_cli.lib import unique_path

  class TestUniquePath(unittest.TestCase):

      def test_unique_path(self):

          for i in range(1000):

  if __name__ == '__main__':

      unittest.main()

          args, kwargs = InsertProcessor.call_args

          ip = IP(*args, **kwargs)

          session.assertPerm.assert_called_with('admin')

          for m in mocks:

          args, kwargs = InsertProcessor.call_args

          ip = IP(*args, **kwargs)

          session.assertPerm.assert_called_with('admin')

          for m in mocks:

  import copy

  import unittest

  import kojihub

              {'foo': 2, 'bar': -1},

              {'foo': 0, 'bar': 0},

          ]

      def test_basic(self):

          opts = None

          expected = copy.copy(self.original)

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_order_by_foo(self):

          opts = {'order': 'foo'}

              {'foo': 2, 'bar': -1},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_order_by_bar(self):

          opts = {'order': 'bar'}

              {'foo': 1, 'bar': 1},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_order_in_reverse(self):

          opts = {'order': '-foo'}

              {'foo': 0, 'bar': 0},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_offset(self):

          opts = {'offset': 1}

              {'foo': 0, 'bar': 0},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_limit(self):

          opts = {'limit': 2}

              {'foo': 2, 'bar': -1},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_limit_and_offset(self):

          opts = {'limit': 1, 'offset': 1}

              {'foo': 2, 'bar': -1},

          ]

          actual = kojihub._applyQueryOpts(self.original, opts)

      def test_count_only(self):

          opts = {'countOnly': True}

          expected = 3

          actual = kojihub._applyQueryOpts(self.original, opts)

          opts = {'countOnly': True, 'offset': 2}

          expected = 1

          actual = kojihub._applyQueryOpts(self.original, opts)

  import koji

  import kojihub

  IP = kojihub.InsertProcessor

  class TestDistRepoInit(unittest.TestCase):

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

          insert = IP(*args, **kwargs)

          insert.execute = mock.MagicMock()

          self.inserts.append(insert)

          return insert

      def setUp(self):

          self.tempdir = tempfile.mkdtemp()

          self.pathinfo = koji.PathInfo(self.tempdir)

          mock.patch('koji.pathinfo', new=self.pathinfo).start()

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

          self.inserts = []

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

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

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

          self.get_event.return_value = 12345

          self.nextval.return_value = 99

      def tearDown(self):

          mock.patch.stopall()

      def test_simple_dist_repo_init(self):

          # simple case

          self.InsertProcessor.assert_called_once()

          ip = self.inserts[0]

          data = {'dist': True, 'create_event': 12345, 'tag_id': 42, 'id': 99,

          # no comps option

          self.copyfile.assert_not_called()

      def test_dist_repo_init_with_comps(self):

          # simple case

          self.InsertProcessor.assert_called_once()

          ip = self.inserts[0]

          data = {'dist': True, 'create_event': 12345, 'tag_id': 42, 'id': 99,

          self.copyfile.assert_called_once()

          assert_policy.assert_called_once_with('dist_repo', {'tag': 'tag'})

          dist_repo_init.assert_called_once()

          make_task.assert_called_once()

          exports.getBuildConfig.assert_called_once_with('tag')

          # generate pkglist file

          self.files.append('pkglist')

          plist = os.path.join(uploaddir, 'pkglist')

          self.rpms = {}

          self.key = '4c8da725'

          with open(plist, 'wt', encoding='utf-8') as f_pkglist:

              for nvr in nvrs:

          self.get_rpm.side_effect = self.our_get_rpm

          self.get_build.side_effect = self.our_get_build

      def tearDown(self):

          mock.patch.stopall()

          shutil.rmtree(self.topdir)

      def our_get_rpm(self, rpminfo, strict=False, multi=False):

          return self.rpms[rpminfo]

      def our_get_build(self, buildInfo, strict=False):

          return self.builds[buildInfo]

      def test_distRepoMove(self):

          session = kojihub.context.session = mock.MagicMock()

          session.user_id = 123

                  raise Exception("Missing file: %s" % path)

              data = open(path, 'rt', encoding='utf-8').read()

              data.strip()

                  return None

          get_rpm.side_effect = mock_get_rpm

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

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

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

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

      def test_getRPMDeps_external_rpm(self, get_rpm):

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

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

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

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

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

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

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

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

      @mock.patch('kojihub.get_rpm')

      @mock.patch('kojihub.get_build')

          get_rpm.side_effect = mock_get_rpm

          re = kojihub.RootExports().getRPMFile(1, 'filename')

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

              kojihub.RootExports().getRPMFile(1, 'filename', strict=True)

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

      def test_getRPMFile_external_rpm(self, get_rpm):

          re = kojihub.RootExports().getRPMFile(1, 'filename')

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

              kojihub.RootExports().getRPMFile(1, 'filename', strict=True)

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

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

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

      def test_getRPMFile_no_rpmfile(self, ope, pr, pb, get_build, get_rpm):

          re = kojihub.RootExports().getRPMFile(1, 'filename')

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

              kojihub.RootExports().getRPMFile(1, 'filename', strict=True)

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

      @mock.patch('kojihub.get_build')

          pi.rpm.return_value = 'test-files-1-1.fc27.noarch.rpm'

          getRPMFile = kojihub.RootExports().getRPMFile

          res = getRPMFile(1, '/fileA')

          res = getRPMFile(1, '/fileB')

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

              res = getRPMFile(1, '/fileB', strict=True)

      @mock.patch('kojihub.get_build')

      @mock.patch('kojihub.Task')

      @mock.patch('kojihub.context')

          task = mock.MagicMock()

          task.assertHost = mock.MagicMock()

          Task.return_value = task

          rpm = {

              'id': 6,

              'name': 'foo',

              'version': '3.1',

          # Check that the log symlink made it to where it was supposed to.

          dest = os.readlink(workdir + '/foo.log')

          dest = os.path.abspath(os.path.join(workdir, dest))

          # And.. check all the sql statements

          expression, kwargs = cursor.execute.mock_calls[0][1]

          expression = " ".join(expression.split())

          expected = 'INSERT INTO archive_rpm_components (archive_id, rpm_id) ' + \

              'VALUES (%(archive_id0)s, %(rpm_id0)s)'

          proc = kojihub.InsertProcessor('sometable')

          actual = str(proc)

          expected = '-- incomplete update: no assigns'

      def test_to_string_with_data(self):

          proc = kojihub.InsertProcessor('sometable', data={'foo': 'bar'})

          actual = str(proc)

          expected = 'INSERT INTO sometable (foo) VALUES (%(foo)s)'

      @mock.patch('kojihub.context')

      def test_simple_execution_with_iterate(self, context):

          context.session.assertLogin = mock.MagicMock()

          proc = kojihub.InsertProcessor('sometable', data={'foo': 'bar'})

          proc.make_create(event_id=1, user_id=2)

          proc.make_create(user_id=2)

          proc.make_create(event_id=1)

          proc.make_create()

      @mock.patch('kojihub.context')

      def test_dup_check(self, context):

          args = cursor.execute.call_args

          actual = ' '.join(args[0][0].split())

          expected = 'SELECT foo FROM sometable WHERE (foo = %(foo)s)'

          proc.make_create()

          proc.dup_check()

          actual = ' '.join(args[0][0].split())

          expected = 'SELECT active, foo FROM sometable WHERE ' + \

              '(active = %(active)s) AND (foo = %(foo)s)'

          proc.set(onething='another')

          proc.rawset(something='something else')

          result = proc.dup_check()

      @mock.patch('kojihub.context')

      def test_raw_data(self, context):

          context.cnx.cursor.return_value = cursor

          proc = kojihub.InsertProcessor('sometable', rawdata={'foo': '\'bar\''})

          result = proc.dup_check()

          actual = str(proc)

          expected = "INSERT INTO sometable (foo) VALUES (('bar'))"  # raw data

  class TestBulkInsertProcessor(unittest.TestCase):

          proc = kojihub.BulkInsertProcessor('sometable')

          actual = str(proc)

          expected = '-- incomplete insert: no data'

      def test_to_string_with_single_row(self):

          proc = kojihub.BulkInsertProcessor('sometable', data=[{'foo': 'bar'}])

          actual = str(proc)

          expected = 'INSERT INTO sometable (foo) VALUES (%(foo0)s)'

          proc = kojihub.BulkInsertProcessor('sometable')

          proc.add_record(foo='bar')

          actual = str(proc)

      @mock.patch('kojihub.context')

      def test_simple_execution(self, context):

          proc.add_record(foo2='bar2')

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

              str(proc)

      def test_missing_values_nostrict(self):

          proc = kojihub.BulkInsertProcessor('sometable', strict=False)

          proc.add_record(foo2='bar2')

          actual = str(proc)

          expected = 'INSERT INTO sometable (foo, foo2) VALUES (%(foo0)s, NULL), (NULL, %(foo21)s)'

      def test_missing_values_explicit_columns(self):

          proc = kojihub.BulkInsertProcessor('sometable', strict=True, columns=['foo', 'foo2'])

          proc.add_record(foo='bar')

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

              str(proc)

      @mock.patch('kojihub.context')

      def test_batch_execution(self, context):

          proc.execute()

          calls = cursor.execute.mock_calls

          # list of (name, positional args, keyword args)