BuildRequires:  sed

  %if 0%{?fedora} <= 27

  BuildRequires:  python-progressbar >= 2.3

  BuildRequires:  PyYAML >= 3.11

  %else

  BuildRequires:  python2-progressbar >= 2.3

  BuildRequires:  python2-pyyaml >= 3.11

  %endif

  testcloud >= 0.1.15

  # Test suite requirements

  mock >= 2.0.0

  pytest >= 2.7.3

  pytest-cov >= 2.2.1

  '''Functional tests for libtaskotron/koji_utils.py'''

  import os

  from libtaskotron.ext.fedora import koji_utils

          is missing an rpm """

          rpmdir = tmpdir.mkdir("rpmdownload")

          test_koji = koji_utils.KojiClient(stub_koji)

          test_koji.get_nvr_rpms(self.ref_nvr, str(rpmdir), arches=[self.ref_arch])

      def test_get_nvr_rpms_dest_exists(self, tmpdir, monkeypatch):

          '''Destination dir must be created even if no files are downloaded'''

          test_koji = koji_utils.KojiClient(stub_koji)

          monkeypatch.setattr(test_koji, 'nvr_to_urls', lambda *args, **kwargs: [])

      def test_get_tagged_rpms_dest_exists(self, tmpdir, monkeypatch):

          '''Destination dir must be created even if no files are downloaded'''

          test_koji = koji_utils.KojiClient(stub_koji)

          monkeypatch.setattr(test_koji, 'get_nvr_rpms', stub_get_nvr_rpms)

          test_koji.get_tagged_rpms('some tag', dest=rpmdir)

          assert os.path.isdir(rpmdir)

  # See the LICENSE file for more details on Licensing

  import pytest

  import configparser

  from libtaskotron.directives import resultsdb_directive

                              u'start_time': None,

                              u'status': u'SCHEDULED'}

          self.test_rdb = resultsdb_directive.ResultsdbDirective(self.stub_rdb)

          # while it appears useless, this actually sets config in several tests

          # we expect rdb to be called 2 times:

          # check for testcase, and report result

          # the log url depends on the arg_data, so construct it here

          ref_log_url = '%s/%s' % (self.conf.artifacts_baseurl, self.ref_arg_data['uuid'])

          with pytest.raises(exc.TaskotronDirectiveError):

              self.test_rdb.process(self.ref_input, self.ref_arg_data)

  import pytest

  import mock

  import StringIO

  from libtaskotron.directives import yumrepoinfo_directive

  from libtaskotron.exceptions import TaskotronDirectiveError

          repoinfo = yumrepoinfo.YumRepoInfo(filelist=[], arch='x86_64')

          repoinfo.parser.readfp(StringIO.StringIO(TEST_CONF))

          monkeypatch.setattr(yumrepoinfo, 'get_yumrepoinfo', stub_getrepoinfo)

          # don't set the repoinfo object, we've stubbed out the code that would

          directive.process(ref_input, None)

          # check the first arg of the first call to the stub object

      def test_only_meta_arches(self):

          directive = yumrepoinfo_directive.YumrepoinfoDirective(filelist=[self.temp_conf.strpath])

  '''Unit tests for libtaskotron/directives/bodhi_directive.py'''

  import pytest

  from libtaskotron.directives import bodhi_directive

  from libtaskotron.exceptions import TaskotronDirectiveError

          self.ref_rpmfile = '%s/%s.rpm' % (self.ref_targetdir, self.ref_nvr1)

          self.helper = bodhi_directive.BodhiDirective()

      def test_action_download_existing_update(self):

          '''Test download of existing update'''

          output_data = test_bodhi.process(self.ref_input, None)

          ref_output_data = {'downloaded_rpms': [self.ref_rpmfile,

                                                 self.ref_rpmfile]}

          # checks whether get_nvr_rpms was called for every update

          # with correct nvrs

          '''Test download when all arches are demanded'''

          self.ref_input['arch'] = 'all'

          test_bodhi.process(self.ref_input, None)

          ref_arches = set(config.get_config().supported_arches + ['noarch'])

          # checks whether all get_nvr_rpms calls demanded all arches

      def test_action_download_source(self):

          '''Test download of source packages'''

          self.ref_input['arch'] = []

          self.ref_input['src'] = True

          test_bodhi.process(self.ref_input, None)

          # checks whether all get_nvr_rpms calls demanded only source pkgs

      def test_action_download_multiple_arch(self):

          '''Test download of multiple arches packages'''

          self.ref_arch = ['x86_64', 'noarch']

          self.ref_input['arch'] = self.ref_arch

          test_bodhi.process(self.ref_input, None)

      def test_action_download_added_noarch(self):

          '''Test whether noarch is automaticaly added'''

          self.ref_input['arch'] = 'i386'

          test_bodhi.process(self.ref_input, None)

          # checks whether noarch is demanded in all get_nvr_rpms calls

      def test_action_download_nonexisting_update(self):

          '''Test whether exception is raised when no update is found'''

          test_bodhi = bodhi_directive.BodhiDirective(stub_bodhi, stub_koji)

          with pytest.raises(TaskotronDirectiveError):

              test_bodhi.process(self.ref_input, None)

      def test_invalid_action(self):

          '''Test response on non-existing action'''

          self.ref_input['action'] = 'foo'

          test_bodhi = bodhi_directive.BodhiDirective(stub_bodhi, stub_koji)

      def test_action_download_insufficient_params(self):

          '''Test response on unsufficient input for download action'''

          test_bodhi = bodhi_directive.BodhiDirective(stub_bodhi, stub_koji)

          for missing_arg in ['action', 'arch', 'update_id', 'target_dir']:

  '''Unit tests for libtaskotron/bodhi_utils.py'''

  import pytest

  from munch import Munch

  import fedora.client.bodhi

      def test_query_existing_update(self):

          '''Test query for existing update'''

          ref_query_answer = {'updates': [self.ref_update]}

          bodhi = bodhi_utils.BodhiUtils(client=stub_bodhi)

          update = bodhi.get_update(self.ref_update_id)

      def test_query_non_existing_update(self, monkeypatch):

          '''Test query for non-existing update'''

          ref_query_answer = {'updates': []}

          bodhi = bodhi_utils.BodhiUtils(client=stub_bodhi)

          update = bodhi.get_update(self.ref_update_id)

      def mock_query(self, builds):

          '''Mock bodhi.client.query()'''

          builds = builds.split()

          updates = []

          for build in builds:

      @pytest.fixture

      def setup(self, monkeypatch):

          '''Run this before every test invocation'''

          self.bodhi = bodhi_utils.BodhiUtils(self.client)

      def test_basic(self):

          '''One update per build, mixed results'''

                                                       'multi2-a-1.0-1.fc20',

                                                       'foo-fail-1.2-3.fc20'])

          # 3 calls, before one of multi1 packages should be skipped

          assert len(calls) == 3

  from libtaskotron import buildbot_utils

  import os.path

  import pytest

  import mock

  from libtaskotron import file_utils

                            'path': self.ref_path,

                            'target_dir': '/var/tmp/foo'}

          monkeypatch.setattr(file_utils, 'download', self.mock_download)

      def _get_url(self, path):

      def test_download(self):

          self.helper.process(self.ref_input, None)

          assert len(download_calls) == 1

                                                         self.ref_path[0])

      def test_localpath(self):

          self.helper.process(self.ref_input, None)

          assert len(download_calls) == 1

                                                         self.ref_localpath[0])

      def test_multiple_localpath(self):

          self.helper.process(self.ref_input, None)

          assert len(download_calls) == 2

                                                         ref_localpath[0])

                                                         ref_localpath[1])

      def test_incorrect_localpath(self):

          self.helper.process(self.ref_input, None)

          assert len(download_calls) == 1

      def test_missing_path(self):

          self.ref_input.pop('path', None)

          self.helper.process(self.ref_input, None)

          assert len(download_calls) == 1

          # self.ref_nvr was not touched, so if nvr parsing gets priority, the URL will not match

      def test_raises_on_404(self, monkeypatch):

          '''Directive must raise when file is missing in distgit (404)'''

          self.helper.process(self.ref_input, None)

  # License: GPL-2.0+ <http://spdx.org/licenses/GPL-2.0+>

  # See the LICENSE file for more details on Licensing

  import pytest

  import mock

  import os

                            'koji_build': self.ref_nvr,

                            'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(self.ref_input, None)

          assert len(getrpm_calls) == 1

          assert requested_nvr == self.ref_nvr

          assert not requested_debuginfo

      def test_parse_download_tag_command(self):

          self.ref_input = {'action': 'download_tag',

                            'arch': self.ref_arch,

                            'koji_tag': self.ref_tag,

                            'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(self.ref_input, None)

          assert len(getrpm_calls) == 1

          assert requested_tag == self.ref_tag

      def test_parse_download_latest_stable_command(self):

          self.ref_input = {'action': 'download_latest_stable',

                            'arch': self.ref_arch,

                            'koji_build': self.ref_nvr,

                            'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(self.ref_input, None)

          assert len(koji_calls) == 2

          assert requested_nvr == self.ref_previous_nvr

          assert not requested_src

      def test_download_latest_stable_no_builds(self):

          '''Don't do anything if there are no latest stable builds'''

          self.ref_input = {'action': 'download_latest_stable',

                            'arch': self.ref_arch,

                            'koji_build': self.ref_nvr,

                            'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          retval = test_helper.process(self.ref_input, None)

          assert len(koji_calls) == 1

          assert retval == {}

      @pytest.mark.parametrize('action', ACTIONS)

          ref_input = {'action': action, 'arch': self.ref_arch,

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          ref_input = {'action': action, 'arch': self.ref_arch,

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          ref_input = {'action': action, 'arch': [self.ref_arch],

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          ref_input = {'action': action, 'arch': 'x86_64',

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

      @pytest.mark.parametrize('action', ACTIONS)

      def test_exclude_arch(self, action):

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          ref_input = {'action': action, 'arch': [], 'src': True,

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          ref_input = {'action': action, 'arch': [], 'debuginfo': True,

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          test_helper.process(ref_input, None)

          assert len(getrpm_calls) >= 1

          for call in getrpm_calls:

              requested_arches = call[2]['arches']

          test_helper.process(ref_input, None)

          if action in BUILD_LOG_ACTIONS:

              assert len(build_log_calls) == 1

          else:

          ref_input = {'action': action, 'arch': [], 'build_log': True,

                       'koji_build': self.ref_nvr, 'koji_tag': self.ref_tag,

                       'target_dir': self.ref_targetdir}

          stub_koji = mock.MagicMock()

          stub_koji.get_build_log.return_value = ref_get_bl_return

          test_helper = koji_directive.KojiDirective(stub_koji)

                            'arch': self.ref_arch,

                            'koji_build': self.ref_nvr,

                            'target_dir': rpmdir}

          test_helper = koji_directive.KojiDirective(stub_koji)

          assert not os.path.exists(rpmdir)

  '''Unit tests for libtaskotron/koji_utils.py'''

  import pytest

  import os

  import itertools

  import mock

      # =====================

      def test_latest_by_tag_first_tag_miss(self):

          test_koji = koji_utils.KojiClient(stub_koji)

          outcome = test_koji.latest_by_tag(self.ref_tags, self.ref_name)

          assert outcome == self.ref_latest_stable

      def test_latest_by_tag_second_tag_miss(self):

          test_koji = koji_utils.KojiClient(stub_koji)

          outcome = test_koji.latest_by_tag(self.ref_tags, self.ref_name)

          assert outcome == self.ref_latest_stable

      def test_latest_by_tag_build_not_found(self):

          test_koji = koji_utils.KojiClient(stub_koji)

          assert test_koji.latest_by_tag(self.ref_tags, self.ref_name) is None

      # =====================

      def test_rpms_to_build(self):

          test_koji = koji_utils.KojiClient(stub_koji)

          outcome = test_koji.rpms_to_build([self.ref_filename,self.ref_filename])

          assert outcome == [self.ref_build, self.ref_build]

          # because two rpms come from same build, it gets called twice for each

          # rpm, once for build

      def test_rpms_to_build_exceptions(self):

          test_koji = koji_utils.KojiClient(stub_koji)

          with pytest.raises(exc.TaskotronRemoteError):

              test_koji.rpms_to_build([self.ref_filename])

          test_koji = koji_utils.KojiClient(stub_koji)