def setUpModule():  # pylint:disable=invalid-name
    """Conditionally skip tests."""
    if selectors.bug_is_untestable(1991, config.get_config().version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/1991')
    if selectors.bug_is_untestable(2242, config.get_config().version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/2242')
    set_up_module()

    def test_update_is_correct(self):
        """Assert the uploaded erratum is in ``updateinfo.xml`` and correct.

        Specificially, this method does the following:

        1. Select the "update" element in ``updateinfo.xml`` corresponding to
           the uploaded erratum.
        2. Build our own erratum from the selected "update" element.
        3. Compare the uploaded erratum to the generated erratum.

        .. WARNING:: This test may be erroneous. See
            :func:`parse_updateinfo_update`.
        """
        # Find our erratum in ``updateinfo.xml``.
        updates = [
            update for update in self.updateinfo.findall('update')
            if update.find('id').text == self.erratum['id']
        ]
        self.assertEqual(len(updates), 1)

        # Parse and verify the erratum. Simply asserting that the original
        # erratum and constructed erratum are equal produces awful failure
        # messages. Iterating like this lets us narrow things down a bit.
        old_erratum = self.erratum.copy()
        if selectors.bug_is_untestable(2021, self.cfg.version):
            del old_erratum['release']
        if selectors.bug_is_untestable(2042, self.cfg.version):
            for package_list in old_erratum['pkglist']:
                for package in package_list['packages']:
                    del package['sum']
        new_erratum = parse_updateinfo_update(updates[0])
        for key, value in old_erratum.items():
            with self.subTest(key=key):
                self.assertEqual(value, new_erratum[key])

    def setUpClass(cls):
        """Create and sync a docker repository with a v2 registry.

        After doing the above, read the repository's JSON app file, tags, and
        the manifest of the first tag.

        This method requires Pulp 2.8 and above, and will raise a ``SkipTest``
        exception if run against an earlier version of Pulp.
        """
        super(SyncPublishV2TestCase, cls).setUpClass()
        if cls.cfg.version < Version('2.8'):
            raise unittest.SkipTest('These tests require Pulp 2.8 or above.')
        if (cls.cfg.version >= Version('2.9') and
                selectors.bug_is_untestable(1909, cls.cfg.version)):
            raise unittest.SkipTest('https://pulp.plan.io/issues/1909')
        if (cls.cfg.version >= Version('2.10') and
                selectors.bug_is_untestable(2287, cls.cfg.version)):
            raise unittest.SkipTest('https://pulp.plan.io/issues/2287')
        docker_utils.repo_create(
            cls.cfg,
            feed=DOCKER_V2_FEED_URL,
            repo_id=cls.repo_id,
            upstream_name=DOCKER_UPSTREAM_NAME,
        )
        cls.completed_proc = docker_utils.repo_sync(cls.cfg, cls.repo_id)
        cls.app_file = _get_app_file(cls.cfg, cls.repo_id)
        cls.tags = _get_tags(cls.cfg, cls.repo_id)
        cls.manifest = _get_manifest(cls.cfg, cls.repo_id, cls.tags['tags'][0])

 def test_connection_error(self):
     """Make the dependent function raise a connection error."""
     ver = Version('0')
     with mock.patch.object(selectors, '_get_bug') as get_bug:
         get_bug.side_effect = requests.exceptions.ConnectionError
         with self.assertWarns(RuntimeWarning):
             selectors.bug_is_testable(None, ver)
         with self.assertWarns(RuntimeWarning):
             selectors.bug_is_untestable(None, ver)

def setUpModule():  # pylint:disable=invalid-name
    """Skip tests if the RPM plugin is not installed."""
    set_up_module()
    cfg = config.get_config()
    if cfg.version < Version('2.8'):
        raise unittest.SkipTest('This module requires Pulp 2.8 or greater.')
    if selectors.bug_is_untestable(2272, cfg.version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/2272')
    if selectors.bug_is_untestable(2144, cfg.version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/2144')

 def test_failures(self):
     """Verify Pulp doesn't create distributors when given bad rel paths."""
     if selectors.bug_is_untestable(1106, self.cfg.version):
         self.skipTest('https://pulp.plan.io/issues/1106')
     for i, response in enumerate(self.responses[3:]):
         with self.subTest(i=i):
             self.assertEqual(response.status_code, 400)

 def test_unsigned_drpm(self):
     """Import an unsigned DRPM into Pulp. Verify it has no signature."""
     if selectors.bug_is_untestable(1806, self.cfg.version):
         self.skipTest('https://pulp.plan.io/issues/1806')
     repo_href = self._create_repo_import_unit(DRPM_UNSIGNED_URL)
     unit = self._find_unit(repo_href, DRPM_UNSIGNED_URL)
     self.assertNotIn('signing_key', unit['metadata'])

 def test_signed_drpm(self):
     """Import a signed DRPM into Pulp. Verify its signature."""
     if selectors.bug_is_untestable(1806, self.cfg.version):
         self.skipTest('https://pulp.plan.io/issues/1806')
     repo_href = self._create_repo_import_unit(DRPM_URL)
     unit = self._find_unit(repo_href, DRPM_URL)
     self._verify_pkg_key(unit, PULP_FIXTURES_KEY_ID)

    def setUpClass(cls):
        """Create and sync a docker repository with a v2 registry."""
        super(CopyAllTagsTestCase, cls).setUpClass()
        if (cls.cfg.version >= Version('2.9') and
                selectors.bug_is_untestable(1909, cls.cfg.version)):
            raise unittest.SkipTest('https://pulp.plan.io/issues/1909')

        # Create a pair of repositories.
        docker_utils.repo_create(
            cls.cfg,
            enable_v1='false',
            enable_v2='true',
            feed=DOCKER_V2_FEED_URL,
            repo_id=cls.repo_ids[0],
            upstream_name=_UPSTREAM_NAME,
        )
        docker_utils.repo_create(cls.cfg, repo_id=cls.repo_ids[1])

        # Sync the first and copy some content units to the second.
        docker_utils.repo_sync(cls.cfg, cls.repo_ids[0])
        cls.copy = docker_utils.repo_copy(
            cls.cfg,
            unit_type='tag',
            from_repo_id=cls.repo_ids[0],
            to_repo_id=cls.repo_ids[1],
        )

def setUpModule():  # pylint:disable=invalid-name
    """Conditionally skip tests. Cache packages to be uploaded to repos.

    Skip the tests in this module if:

    * The RPM plugin is unsupported.
    * `Pulp #1991 <https://pulp.plan.io/issues/1991>`_ is untestable for the
      version of Pulp under test.
    """
    if selectors.bug_is_untestable(1991, config.get_config().version):
        raise unittest2.SkipTest('https://pulp.plan.io/issues/1991')
    set_up_module()
    global _PACKAGES  # pylint:disable=global-statement
    try:
        _PACKAGES = {
            'signed drpm': utils.http_get(DRPM_URL),
            'signed rpm': utils.http_get(RPM_URL),
            'signed srpm': utils.http_get(SRPM_URL),
            'unsigned drpm': utils.http_get(DRPM_UNSIGNED_URL),
            'unsigned rpm': utils.http_get(RPM_UNSIGNED_URL),
            'unsigned srpm': utils.http_get(SRPM_UNSIGNED_URL),
        }
    except:
        _PACKAGES = None
        raise

    def test_blob_digests(self):
        """Assert that the checksum embedded in each blob's URL is correct.

        For each of the "fsLayers" in the repository manifest, download and
        checksum its blob, and compare this checksum to the one embedded in the
        blob's URL.
        """
        # Issue 1781 only affects RHEL 6.
        if (selectors.bug_is_untestable(1781, self.cfg.version) and
                cli.Client(self.cfg, cli.echo_handler).run((
                    'grep',
                    '-i',
                    'red hat enterprise linux server release 6',
                    '/etc/redhat-release',
                )).returncode == 0):
            self.skipTest('https://pulp.plan.io/issues/1781')

        for fs_layer in self.manifest['fsLayers']:
            with self.subTest(fs_layer=fs_layer):
                blob_sum = fs_layer['blobSum']
                blob = api.Client(self.cfg).get(
                    '/pulp/docker/v2/{}/blobs/{}'
                    .format(self.repo_id, blob_sum)
                ).content
                algo, expected_digest = blob_sum.split(':')
                hasher = getattr(hashlib, algo)()
                hasher.update(blob)
                self.assertEqual(expected_digest, hasher.hexdigest())

 def test_display_order_occurences(self):
     """Assert ``display_order`` occurs once if omitted from the unit."""
     if selectors.bug_is_untestable(1787, self.cfg.version):
         self.skipTest('https://pulp.plan.io/issues/1787')
     input_id = self.package_groups['minimal']['id']
     output = _get_groups_by_id(self.root_element)[input_id]
     self.assertEqual(len(output.findall('display_order')), 1)

    def test_force_sync(self):
        """Test whether one can force Pulp to perform a full sync."""
        cfg = config.get_config()
        if selectors.bug_is_untestable(1982, cfg.version):
            self.skipTest("https://pulp.plan.io/issues/1982")

        # Create and sync a repository.
        client = cli.Client(cfg)
        repo_id = utils.uuid4()
        client.run("pulp-admin rpm repo create --repo-id {} --feed {}".format(repo_id, RPM_SIGNED_FEED_URL).split())
        self.addCleanup(client.run, "pulp-admin rpm repo delete --repo-id {}".format(repo_id).split())
        sync_repo(cfg, repo_id)

        # Delete a random RPM
        rpms = self._list_rpms(cfg)
        client.run("{} rm -rf {}".format("sudo" if not is_root(cfg) else "", random.choice(rpms)).split())
        with self.subTest(comment="Verify the RPM was removed."):
            self.assertEqual(len(self._list_rpms(cfg)), len(rpms) - 1)

        # Sync the repository *without* force_sync.
        sync_repo(cfg, repo_id)
        with self.subTest(comment="Verify the RPM has not been restored."):
            self.assertEqual(len(self._list_rpms(cfg)), len(rpms) - 1)

        # Sync the repository again
        sync_repo(cfg, repo_id, force_sync=True)
        with self.subTest(comment="Verify the RPM has been restored."):
            self.assertEqual(len(self._list_rpms(cfg)), len(rpms))

    def test_publish_override_config(self):
        """Use the ``packages_directory`` publish override option.

        Create a distributor with default options, and use it to publish the
        repository. Specify the ``packages_directory`` option during the
        publish as an override option. Verify packages end up in the specified
        directory, relative to the published repository's root.
        """
        if selectors.bug_is_untestable(1976, self.cfg.version):
            self.skipTest('https://pulp.plan.io/issues/1976')
        client = api.Client(self.cfg, api.json_handler)
        distributor = client.post(
            urljoin(self.repo_href, 'distributors/'),
            gen_distributor(),
        )
        packages_dir = utils.uuid4()
        client.post(urljoin(self.repo_href, 'actions/publish/'), {
            'id': distributor['id'],
            'override_config': {'packages_directory': packages_dir},
        })
        primary_xml = get_parse_repodata_primary_xml(self.cfg, distributor)
        package_hrefs = get_package_hrefs(primary_xml)
        self.assertGreater(len(package_hrefs), 0)
        for package_href in package_hrefs:
            with self.subTest(package_href=package_href):
                self.assertEqual(os.path.dirname(package_href), packages_dir)

    def setUpClass(cls):
        """Create an RPM repository with a valid feed and sync it.

        Do the following:

        1. Reset Pulp, including the Squid cache.
        2. Create a repository with the "background" download policy.
        3. Sync and publish the repository.
        4. Download an RPM from the repository.
        """
        super(BackgroundTestCase, cls).setUpClass()
        if (selectors.bug_is_untestable(1905, cls.cfg.version) and
                _os_is_rhel6(cls.cfg)):
            raise unittest.SkipTest('https://pulp.plan.io/issues/1905')

        # Required to ensure content is actually downloaded.
        utils.reset_squid(cls.cfg)
        utils.reset_pulp(cls.cfg)

        # Create, sync and publish a repository.
        repo = _create_repo(cls.cfg, 'background')
        cls.resources.add(repo['_href'])
        report = utils.sync_repo(cls.cfg, repo['_href']).json()

        # Record the tasks spawned when syncing the repository, and the state
        # of the repository itself after the sync.
        client = api.Client(cls.cfg)
        cls.repo = client.get(repo['_href'], params={'details': True}).json()
        cls.tasks = tuple(api.poll_spawned_tasks(cls.cfg, report))

        # Download an RPM.
        path = urljoin('/pulp/repos/', repo['id'] + '/')
        path = urljoin(path, RPM)
        cls.rpm = client.get(path)

    def setUpClass(cls):
        """Create a repository and add an importer and distributor to it.

        Do the following:

        1. Create a repository.
        2. Read the repository's importers and distributors.
        3. Add an importer and distributor to the repo.
        4. Re-read the repository's importers and distributors.
        """
        super(AddImporterDistributorTestCase, cls).setUpClass()
        if cls.cfg.version >= Version("2.10") and selectors.bug_is_untestable(2082, cls.cfg.version):
            raise SkipTest("https://pulp.plan.io/issues/2082")

        # Steps 1 and 2.
        client = api.Client(cls.cfg, api.json_handler)
        href = client.post(REPOSITORY_PATH, {"id": utils.uuid4()})["_href"]
        cls.resources.add(href)
        cls.pre_imp = client.get(urljoin(href, "importers/"))
        cls.pre_dist = client.get(urljoin(href, "distributors/"))

        # Steps 3 and 4.
        client.response_handler = api.safe_handler
        cls.add_imp = client.post(urljoin(href, "importers/"), {"importer_type_id": "iso_importer"})
        cls.add_dist = client.post(
            urljoin(href, "distributors/"),
            {"distributor_config": {}, "distributor_id": utils.uuid4(), "distributor_type_id": "iso_distributor"},
        )
        client.response_handler = api.json_handler
        cls.post_imp = client.get(urljoin(href, "importers/"))
        cls.post_dist = client.get(urljoin(href, "distributors/"))

    def test_sync_downloaded_content(self):
        """Create two repositories with the same feed, and sync them serially.

        More specifically, this test creates two puppet repositories with
        identical feeds, syncs them serially, and verifies that both have equal
        non-zero content unit counts.
        """
        cfg = config.get_config()
        if selectors.bug_is_untestable(1937, cfg.version):
            self.skipTest('https://pulp.plan.io/issues/1937')
        utils.pulp_admin_login(cfg)

        # Create two repos, schedule them for deletion, and sync them.
        client = cli.Client(cfg)
        repo_ids = [utils.uuid4() for _ in range(2)]
        for repo_id in repo_ids:
            client.run((
                'pulp-admin puppet repo create '
                '--repo-id {} --feed {} --queries {}'
            ).format(repo_id, PUPPET_FEED, PUPPET_QUERY).split())
            self.addCleanup(client.run, (
                'pulp-admin puppet repo delete --repo-id {}'
            ).format(repo_id).split())
            client.run((
                'pulp-admin puppet repo sync run --repo-id {}'
            ).format(repo_id).split())

        # Verify the number of puppet modules in each repository.
        unit_counts = [
            get_num_units_in_repo(cfg, repo_id) for repo_id in repo_ids
        ]
        for i, unit_count in enumerate(unit_counts):
            with self.subTest(i=i):
                self.assertGreater(unit_count, 0)
        self.assertEqual(unit_counts[0], unit_counts[1])

    def test_all(self):
        """Publish the rpm rsync distributor before the yum distributor."""
        cfg = config.get_config()
        if selectors.bug_is_untestable(2187, cfg.version):
            self.skipTest('https://pulp.plan.io/issues/2187')

        # Create a user and a repository.
        ssh_user, priv_key = self.make_user(cfg)
        ssh_identity_file = self.write_private_key(cfg, priv_key)
        repo = self.make_repo(cfg, {'remote': {
            'host': urlparse(cfg.base_url).netloc,
            'root': '/home/' + ssh_user,
            'ssh_identity_file': ssh_identity_file,
            'ssh_user': ssh_user,
        }})

        # Publish with the rsync distributor.
        distribs = _get_dists_by_type_id(cfg, repo['_href'])
        self.verify_publish_is_skip(cfg, utils.publish_repo(
            cfg,
            repo,
            {'id': distribs['rpm_rsync_distributor']['id']}
        ).json())

        # Verify that the rsync distributor hasn't placed files
        sudo = '' if utils.is_root(cfg) else 'sudo '
        cmd = (sudo + 'ls -1 /home/{}'.format(ssh_user)).split()
        dirs = set(cli.Client(cfg).run(cmd).stdout.strip().split('\n'))
        self.assertNotIn('content', dirs)

def setUpModule():  # pylint:disable=invalid-name
    """Conditionally skip tests. Cache packages to be uploaded to repos.

    Skip the tests in this module if:

    * The RPM plugin is unsupported.
    * `Pulp #1991 <https://pulp.plan.io/issues/1991>`_ is untestable for the
      version of Pulp under test.
    """
    cfg = config.get_config()
    if selectors.bug_is_untestable(1991, cfg.version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/1991')
    set_up_module()
    try:
        _SIGNED_PACKAGES['rpm'] = utils.http_get(RPM_URL)
        _SIGNED_PACKAGES['srpm'] = utils.http_get(SRPM_URL)
        _UNSIGNED_PACKAGES['rpm'] = utils.http_get(RPM_UNSIGNED_URL)
        _UNSIGNED_PACKAGES['srpm'] = utils.http_get(SRPM_UNSIGNED_URL)
        if selectors.bug_is_testable(1806, cfg.version):
            _SIGNED_PACKAGES['drpm'] = utils.http_get(DRPM_URL)
            _UNSIGNED_PACKAGES['drpm'] = utils.http_get(DRPM_UNSIGNED_URL)
    except:
        _SIGNED_PACKAGES.clear()
        _UNSIGNED_PACKAGES.clear()
        raise

def setUpModule():  # pylint:disable=invalid-name
    """Conditionally skip tests. Create repositories with fixture data."""
    cfg = config.get_config()
    if selectors.bug_is_untestable(1991, cfg.version):
        raise unittest.SkipTest('https://pulp.plan.io/issues/1991')
    set_up_module()

    # Fetch RPMs.
    _SIGNED_PACKAGES['rpm'] = utils.http_get(RPM_URL)
    _SIGNED_PACKAGES['srpm'] = utils.http_get(SRPM_URL)
    _UNSIGNED_PACKAGES['rpm'] = utils.http_get(RPM_UNSIGNED_URL)
    _UNSIGNED_PACKAGES['srpm'] = utils.http_get(SRPM_UNSIGNED_URL)
    if selectors.bug_is_testable(1806, cfg.version):
        _SIGNED_PACKAGES['drpm'] = utils.http_get(DRPM_URL)
        _UNSIGNED_PACKAGES['drpm'] = utils.http_get(DRPM_UNSIGNED_URL)

    # Create repos, and upload RPMs to them.
    client = api.Client(cfg, api.json_handler)
    try:
        repo = client.post(REPOSITORY_PATH, gen_repo())
        _REPOS['signed'] = repo
        for type_id, pkg in _SIGNED_PACKAGES.items():
            utils.upload_import_unit(cfg, pkg, type_id, repo['_href'])

        repo = client.post(REPOSITORY_PATH, gen_repo())
        _REPOS['unsigned'] = repo
        for type_id, pkg in _UNSIGNED_PACKAGES.items():
            utils.upload_import_unit(cfg, pkg, type_id, repo['_href'])
    except:
        _SIGNED_PACKAGES.clear()
        _UNSIGNED_PACKAGES.clear()
        for _ in range(len(_REPOS)):
            client.delete(_REPOS.popitem()[1]['_href'])
        raise