omit =

      /usr/*

      tests/*

  [report]

  exclude_lines =

  test3:

  	coverage3 erase

  	    --rcfile .coveragerc3 --source . \

  	@echo Full coverage report at file://${PWD}/htmlcov/index.html

  test-tarball:

      return rv

  def watch_logs(session, tasklist, opts, poll_interval):

      print("Watching logs (this may be safely interrupted)...")

                                  sys.stdout.write("\n")

                              sys.stdout.write("==> %s <==\n" % currlog)

                              lastlog = currlog

              if opts.follow:

  import koji

  from koji.plugin import export_cli

  from koji_cli.lib import _, activate_session, OptionParser, watch_tasks, \

  import six

                  log = session.downloadTaskOutput(task_id, 'runroot.log', volume=volume)

                  # runroot output, while normally text, can be *anything*, so

                  # treat it as binary

      info = session.getTaskInfo(task_id)

      if info is None:

          sys.exit(1)

  KOJID_FILENAME = os.path.dirname(__file__) + "/../../builder/kojid"

  if sys.version_info[0] >= 3:

      import importlib.util

      kojid = importlib.util.module_from_spec(spec)

      spec.loader.exec_module(kojid)

  else:

  import json

  import mock

  import os

  import tempfile

  try:

      import unittest2 as unittest

          fn = os.path.join(os.path.dirname(__file__), 'data/calls', name,'calls.json')

          with open(fn) as fp:

              data = json.load(fp)

          for call in data:

              key = self._munge([call['method'], call['args'], call['kwargs']])

              self._testcalls[key] = call

          self.assertEqual(from_addr, "koji@example.com")

          self.assertEqual(recipients, ["user@example.com"])

          fn = os.path.join(os.path.dirname(__file__), 'data/calls', 'build_notif_1', 'message.txt')

              msg_expect = fp.read()

          self.assertEqual(message, msg_expect)

      def test_volume_id_substitutions(self):

          """Check that volume ID is shorten corect by shortenVolID method."""

              name = values['name']

              expected_vol_id = values['expected-id']

              result_vol_id = self.handler._shortenVolID(name, self.version, self.release)

  from __future__ import absolute_import

  import os

  import subprocess

  try:

      import unittest2 as unittest

          output = popen.stdout.read()

          # rpm outputs a line for each subpackage

          version = output.splitlines()[0]

          return version

      def test_docs_version(self):

          import importlib.machinery

          loader = importlib.machinery.SourceFileLoader(mod_name, CLI_FILENAME)

          spec = importlib.util.spec_from_loader(loader.name, loader)

      else:

          import imp

          plugin = imp.load_source(mod_name, CLI_FILENAME)

      def test_send_queued_msgs_fail(self, getLogger, Container):

          context.protonmsg_msgs = [('test.topic', {'testheader': 1}, 'test body')]

          conf = tempfile.NamedTemporaryFile()

  urls = amqps://broker1.example.com:5671 amqps://broker2.example.com:5671

  cert = /etc/koji-hub/plugins/client.pem

  cacert = /etc/koji-hub/plugins/ca.pem

  topic_prefix = koji

  connect_timeout = 10

  send_timeout = 60

          conf.flush()

          protonmsg.CONFIG_FILE = conf.name

          protonmsg.CONFIG = None

      def test_send_queued_msgs_success(self, getLogger, Container):

          context.protonmsg_msgs = [('test.topic', {'testheader': 1}, 'test body')]

          conf = tempfile.NamedTemporaryFile()

  urls = amqps://broker1.example.com:5671 amqps://broker2.example.com:5671

  cert = /etc/koji-hub/plugins/client.pem

  cacert = /etc/koji-hub/plugins/ca.pem

  topic_prefix = koji

  connect_timeout = 10

  send_timeout = 60

          conf.flush()

          protonmsg.CONFIG_FILE = conf.name

          protonmsg.CONFIG = None

      def test_send_queued_msgs_test_mode(self, getLogger, Container):

          context.protonmsg_msgs = [('test.topic', {'testheader': 1}, 'test body')]

          conf = tempfile.NamedTemporaryFile()

  urls = amqps://broker1.example.com:5671 amqps://broker2.example.com:5671

  cert = /etc/koji-hub/plugins/client.pem

  cacert = /etc/koji-hub/plugins/ca.pem

  connect_timeout = 10

  send_timeout = 60

  test_mode = on

          conf.flush()

          protonmsg.CONFIG_FILE = conf.name

          protonmsg.CONFIG = None

  class TestRunrootConfig(unittest.TestCase):

      def test_bad_config_paths0(self, safe_config_parser):

          cp = FakeConfigParser()

          del cp.CONFIG['path0']['mountpoint']

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

              "bad config: missing options in path0 section")

      def test_bad_config_absolute_path(self, safe_config_parser):

          cp = FakeConfigParser()

          cp.CONFIG['paths']['default_mounts'] = ''

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

              "bad config: all paths (default_mounts, safe_roots, path_subs) needs to be absolute: ")

      def test_valid_config(self, safe_config_parser):

          safe_config_parser.return_value = FakeConfigParser()

          session = mock.MagicMock()

          options.workdir = '/tmp/nonexistentdirectory'

          runroot.RunRootTask(123, 'runroot', {}, session, options)

      def test_valid_config_alt(self, safe_config_parser):

          safe_config_parser.return_value = FakeConfigParser(CONFIG2)

          session = mock.MagicMock()

          options.workdir = '/tmp/nonexistentdirectory'

          runroot.RunRootTask(123, 'runroot', {}, session, options)

      def test_pathnum_gaps(self, safe_config_parser):

          session = mock.MagicMock()

          options = mock.MagicMock()

          paths = list([CONFIG2[k] for k in ('path0', 'path1', 'path2')])

          self.assertEqual(task2.config['paths'], paths)

      def test_bad_path_sub(self, safe_config_parser):

          session = mock.MagicMock()

          options = mock.MagicMock()

  class TestMounts(unittest.TestCase):

      def setUp(self, safe_config_parser):

          safe_config_parser.return_value = FakeConfigParser()

          self.session = mock.MagicMock()

          os_unlink.assert_not_called()

  class TestHandler(unittest.TestCase):

      def setUp(self, safe_config_parser):

          self.session = mock.MagicMock()

          self.br = mock.MagicMock()

          self.tempdir = tempfile.mkdtemp()

          self.session = mock.MagicMock()

          self.uploadpath = mock.MagicMock()

          self.config = '''

              default:*

              nocommon:*:no

              excClass, exc = sys.exc_info()[:2]

              values['result'] = exc

              values['excClass'] = excClass

      else:

          values['result'] = None

          values['excClass'] = None