def set_distributor_scratchpad(self, repo_group_id, distributor_id, contents):
        """
        Sets the value of the scratchpad for the given group's distributor,
        replacing the existing value if one is present.

        @param contents: value to save in the scratchpad; must be serializable
        @type  contents: object
        """

        distributor = self.get_distributor(repo_group_id, distributor_id)
        distributor['scratchpad'] = contents
        RepoGroupDistributor.get_collection().save(distributor, safe=True)

    def find_with_distributor_type(self, distributor_type_id):
        """
        Returns a list of repository groups, including a list of distributor
        instances, for all groups that have a configured distributor of the
        given type. The distributor list will be stored in the group under
        the key "distributors"

        @return: list of group objects from the database with an extra key
                 added holding the distributor instances
        """

        group_coll = RepoGroup.get_collection()
        group_distributor_coll = RepoGroupDistributor.get_collection()

        groups_by_id = {}

        spec = {'distributor_type_id' : distributor_type_id}
        group_distributors = list(group_distributor_coll.find(spec))
        for gd in group_distributors:
            group = groups_by_id.get(gd['repo_group_id'], None)
            if group is None:
                group = group_coll.find_one({'id' : gd['repo_group_id']})
                groups_by_id[gd['repo_group_id']] = group

            dists = group.setdefault('distributors', [])
            dists.append(gd)

        return groups_by_id.values()

    def _do_publish(self, group, distributor_id, distributor_instance, conduit, call_config):

        distributor_coll = RepoGroupDistributor.get_collection()
        publish_result_coll = RepoGroupPublishResult.get_collection()
        group_id = group.id

        # Perform the publish
        publish_start_timestamp = _now_timestamp()
        try:
            report = distributor_instance.publish_group(group, conduit, call_config)
        except Exception, e:
            publish_end_timestamp = _now_timestamp()

            # Reload the distributor in case the scratchpad is changed by the plugin
            distributor = distributor_coll.find_one({'id' : distributor_id, 'repo_group_id' : group_id})
            distributor['last_publish'] = publish_end_timestamp
            distributor_coll.save(distributor)

            # Add a publish history entry for the run
            result = RepoGroupPublishResult.error_result(group_id, distributor_id,
                     distributor['distributor_type_id'], publish_start_timestamp,
                     publish_end_timestamp, e, sys.exc_info()[2])
            publish_result_coll.save(result, safe=True)

            _LOG.exception('Exception caught from plugin during publish call for group [%s]' % group_id)
            raise PulpExecutionException(e), None, sys.exc_info()[2]

    def get_distributor(self, repo_group_id, distributor_id):
        """
        Returns an individual distributor on the given repo group, raising
        an exception if one does not exist at the given ID.

        @param repo_group_id: identifies the repo group
        @type  repo_group_id: str

        @param distributor_id: identifies the distributor
        @type  distributor_id: str

        @return: SON representation of the distributor
        @rtype:  dict

        @raise MissingResource: if either there is no distributor for the
        given group ID/distributor ID pair or the group itself does not exist
        """

        # Check the group's existence for the exception contract first
        manager_factory.repo_group_query_manager().get_group(repo_group_id)

        # Check for the distributor if we know the group exists
        spec = {
            'repo_group_id' : repo_group_id,
            'id' : distributor_id,
        }
        distributor = RepoGroupDistributor.get_collection().find_one(spec)

        if distributor is None:
            raise MissingResource(repo_group=repo_group_id, distributor=distributor_id)

        return distributor

    def _do_publish(group, distributor_id, distributor_instance, conduit, call_config):

        distributor_coll = RepoGroupDistributor.get_collection()
        publish_result_coll = RepoGroupPublishResult.get_collection()
        group_id = group.id

        # Perform the publish
        publish_start_timestamp = _now_timestamp()
        try:
            report = distributor_instance.publish_group(group, conduit, call_config)
        except Exception, e:
            publish_end_timestamp = _now_timestamp()

            # Reload the distributor in case the scratchpad is changed by the plugin
            distributor = distributor_coll.find_one({"id": distributor_id, "repo_group_id": group_id})
            distributor["last_publish"] = publish_end_timestamp
            distributor_coll.save(distributor)

            # Add a publish history entry for the run
            result = RepoGroupPublishResult.error_result(
                group_id,
                distributor_id,
                distributor["distributor_type_id"],
                publish_start_timestamp,
                publish_end_timestamp,
                e,
                sys.exc_info()[2],
            )
            publish_result_coll.save(result, safe=True)

            raise

    def test_last_publish(self):
        # Test - Unpublished
        unpublished = self.conduit.last_publish()
        self.assertTrue(unpublished is None)

        last_publish = datetime.datetime.now()
        repo_group_dist = self.distributor_manager.get_distributor(self.group_id,
                                                                   self.distributor_id)
        repo_group_dist['last_publish'] = dateutils.format_iso8601_datetime(last_publish)
        RepoGroupDistributor.get_collection().save(repo_group_dist, safe=True)

        # Test
        found = self.conduit.last_publish()
        self.assertTrue(isinstance(found, datetime.datetime))
        # simulate the DB encoding
        last_publish = dateutils.parse_iso8601_datetime(repo_group_dist['last_publish'])
        self.assertEqual(last_publish, found)

    def test_delete_with_distributor(self):
        # Setup
        group_id = 'doomed'
        self.manager.create_repo_group(group_id)

        distributor_id = 'doomed-dist'
        dist_manager = managers_factory.repo_group_distributor_manager()
        dist_manager.add_distributor(group_id, 'mock-group-distributor', {}, distributor_id=distributor_id)

        distributor = RepoGroupDistributor.get_collection().find_one({'id' : distributor_id})
        self.assertTrue(distributor is not None)

        # Test
        self.manager.delete_repo_group(group_id)

        # Verify
        distributor = RepoGroupDistributor.get_collection().find_one({'id' : distributor_id})
        self.assertTrue(distributor is None)

    def test_add_distributor_multiple_distributors(self):
        # Setup
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {})

        # Test
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {})

        # Verify
        all_distributors = list(RepoGroupDistributor.get_collection().find())
        self.assertEqual(2, len(all_distributors))

    def test_remove_distributor(self):
        # Setup
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {}, distributor_id='d1')

        # Test
        self.distributor_manager.remove_distributor(self.group_id, 'd1')

        # Verify
        distributor = RepoGroupDistributor.get_collection().find_one({'id' : 'd1'})
        self.assertTrue(distributor is None)

    def test_delete(self):
        # Setup
        group_id = 'group'
        distributor_id = 'created'
        self.manager.create_repo_group(group_id)
        self.distributor_manager.add_distributor(group_id, 'mock-group-distributor', {}, distributor_id=distributor_id)

        # Test
        status, body = self.delete('/v2/repo_groups/%s/distributors/%s/' % (group_id, distributor_id))

        # Verify
        self.assertEqual(200, status)
        self.assertEqual(body, None)

        found = RepoGroupDistributor.get_collection().find_one({'id' : distributor_id})
        self.assertEqual(found, None)

    def test_delete(self):
        # Setup
        group_id = "group"
        distributor_id = "created"
        self.manager.create_repo_group(group_id)
        self.distributor_manager.add_distributor(group_id, "mock-group-distributor", {}, distributor_id=distributor_id)

        # Test
        status, body = self.delete("/v2/repo_groups/%s/distributors/%s/" % (group_id, distributor_id))

        # Verify
        self.assertEqual(200, status)
        self.assertEqual(body, None)

        found = RepoGroupDistributor.get_collection().find_one({"id": distributor_id})
        self.assertEqual(found, None)

    def test_add_distributor_replace_existing(self):
        # Setup
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {}, distributor_id='d1')

        # Test
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor-2', {}, distributor_id='d1')

        # Verify
        all_distributors = list(RepoGroupDistributor.get_collection().find())
        self.assertEqual(1, len(all_distributors))

        self.assertEqual(all_distributors[0]['distributor_type_id'], 'mock-group-distributor-2')

        # Plugin Calls
        self.assertEqual(1, mock_plugins.MOCK_GROUP_DISTRIBUTOR.distributor_added.call_count)
        self.assertEqual(1, mock_plugins.MOCK_GROUP_DISTRIBUTOR_2.distributor_added.call_count)
        self.assertEqual(1, mock_plugins.MOCK_GROUP_DISTRIBUTOR.distributor_removed.call_count)

    def test_put(self):
        # Setup
        group_id = 'group-1'
        distributor_id = 'dist-1'
        self.manager.create_repo_group(group_id)
        self.distributor_manager.add_distributor(group_id, 'mock-group-distributor', {'a' : 'A'}, distributor_id=distributor_id)

        # Test
        updated_config = {'b' : 'B'}
        status, body = self.put('/v2/repo_groups/%s/distributors/%s/' % (group_id, distributor_id), {'distributor_config' : updated_config})

        # Verify
        self.assertEqual(200, status)
        self.assertEqual(body['id'], distributor_id)

        found = RepoGroupDistributor.get_collection().find_one({'id' : distributor_id})
        self.assertEqual(found['config'], {'a' : 'A', 'b' : 'B'})

    def remove_distributor(repo_group_id, distributor_id, force=False):
        """
        Removes a distributor from a group.

        @param repo_group_id: identifies the group
        @type  repo_group_id: str

        @param distributor_id: identifies the distributor on the group
        @type  distributor_id: str

        @param force: if true, the distributor will be removed from the database
               regardless of whether or not the plugin's clean up method raises
               an exception

        @raise MissingResource: if there is no group or distributor with the
               given ID
        @raise PulpExecutionException: if the distributor raises an error on cleanup
        """
        distributor_coll = RepoGroupDistributor.get_collection()

        # Validation - calls will raise MissingResource
        group = manager_factory.repo_group_query_manager().get_group(repo_group_id)
        distributor = RepoGroupDistributorManager.get_distributor(repo_group_id, distributor_id)

        # Call the distributor's cleanup method
        distributor_type_id = distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_group_distributor_by_id(
            distributor_type_id)

        call_config = PluginCallConfiguration(plugin_config, distributor['config'])
        transfer_group = common_utils.to_transfer_repo_group(group)

        try:
            distributor_instance.distributor_removed(transfer_group, call_config)
        except Exception:
            _logger.exception('Exception cleaning up distributor [%s] on group [%s]' % (
                distributor_id, repo_group_id))

            if not force:
                raise PulpExecutionException(), None, sys.exc_info()[2]

        # Clean up the database
        distributor_coll.remove(distributor, safe=True)

    def test_post(self):
        # Setup
        group_id = "group-1"
        self.manager.create_repo_group(group_id)

        # Test
        data = {
            "distributor_type_id": "mock-group-distributor",
            "distributor_config": {"a": "A"},
            "distributor_id": "dist-1",
        }
        status, body = self.post("/v2/repo_groups/%s/distributors/" % group_id, data)

        # Verify
        self.assertEqual(201, status)
        self.assertEqual(body["id"], data["distributor_id"])

        found = RepoGroupDistributor.get_collection().find_one({"id": data["distributor_id"]})
        self.assertTrue(found is not None)

    def test_post(self):
        # Setup
        group_id = 'group-1'
        self.manager.create_repo_group(group_id)

        # Test
        data = {
            'distributor_type_id' : 'mock-group-distributor',
            'distributor_config' : {'a' : 'A'},
            'distributor_id' : 'dist-1',
        }
        status, body = self.post('/v2/repo_groups/%s/distributors/' % group_id, data)

        # Verify
        self.assertEqual(201, status)
        self.assertEqual(body['id'], data['distributor_id'])

        found = RepoGroupDistributor.get_collection().find_one({'id' : data['distributor_id']})
        self.assertTrue(found is not None)

    def find_distributors(self, repo_group_id):
        """
        Returns all distributors on the given repo group, returning an empty
        list if none exist.

        @param repo_group_id: identifies the repo group
        @type  repo_group_id: str

        @return: list of SON representations of the group's distributors
        @rtype:  list

        @raise MissingResource: if the group does not exist
        """
        group = RepoGroup.get_collection().find_one({'id' : repo_group_id})
        if group is None:
            raise MissingResource(repo_group=repo_group_id)

        spec = {'repo_group_id' : repo_group_id}
        distributors = list(RepoGroupDistributor.get_collection().find(spec))
        return distributors

    def test_put(self):
        # Setup
        group_id = "group-1"
        distributor_id = "dist-1"
        self.manager.create_repo_group(group_id)
        self.distributor_manager.add_distributor(
            group_id, "mock-group-distributor", {"a": "A"}, distributor_id=distributor_id
        )

        # Test
        updated_config = {"b": "B"}
        status, body = self.put(
            "/v2/repo_groups/%s/distributors/%s/" % (group_id, distributor_id), {"distributor_config": updated_config}
        )

        # Verify
        self.assertEqual(200, status)
        self.assertEqual(body["id"], distributor_id)

        found = RepoGroupDistributor.get_collection().find_one({"id": distributor_id})
        self.assertEqual(found["config"], {"a": "A", "b": "B"})

    def test_add_distributor(self):
        # Setup
        config = {'a': 'a', 'b': None}

        # Test
        added = self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor',
                                                         config)

        # Verify
        expected_config = {'a': 'a'}

        # Returned Value
        self.assertTrue(added is not None)
        self.assertEqual(added['config'], expected_config)
        self.assertEqual(added['distributor_type_id'], 'mock-group-distributor')

        # Database
        distributor = RepoGroupDistributor.get_collection().find_one({'id': added['id']})
        self.assertTrue(distributor is not None)
        self.assertEqual(distributor['config'], expected_config)
        self.assertEqual(distributor['distributor_type_id'], 'mock-group-distributor')

        #   Plugin - Validate Config
        self.assertEqual(1, mock_plugins.MOCK_GROUP_DISTRIBUTOR.validate_config.call_count)
        call_group = mock_plugins.MOCK_GROUP_DISTRIBUTOR.validate_config.call_args[0][0]
        call_config = mock_plugins.MOCK_GROUP_DISTRIBUTOR.validate_config.call_args[0][1]

        self.assertTrue(isinstance(call_group, RepositoryGroup))
        self.assertEqual(call_group.id, self.group_id)

        self.assertTrue(isinstance(call_config, PluginCallConfiguration))
        self.assertTrue(call_config.repo_plugin_config, expected_config)

        #   Plugin - Distributor Added
        self.assertEqual(1, mock_plugins.MOCK_GROUP_DISTRIBUTOR.distributor_added.call_count)
        call_group = mock_plugins.MOCK_GROUP_DISTRIBUTOR.validate_config.call_args[0][0]
        call_config = mock_plugins.MOCK_GROUP_DISTRIBUTOR.validate_config.call_args[0][1]

        self.assertTrue(isinstance(call_group, RepositoryGroup))
        self.assertTrue(isinstance(call_config, PluginCallConfiguration))

    def clean(self):
        super(PublishActionTests, self).clean()

        RepoGroup.get_collection().remove()
        RepoGroupDistributor.get_collection().remove()