def publish(group_id, distributor_id, publish_config_override=None):
        """
        Requests the given distributor publish the repository group.

        :param group_id:                identifies the repo group
        :type  group_id:                str
        :param distributor_id:          identifies the group's distributor
        :type  distributor_id:          str
        :param publish_config_override: values to pass the plugin for this publish call alone
        :type  publish_config_override: dict
        """
        distributor_manager = manager_factory.repo_group_distributor_manager()
        distributor = distributor_manager.get_distributor(group_id, distributor_id)
        distributor_type_id = distributor["distributor_type_id"]
        distributor_instance, plugin_config = plugin_api.get_group_distributor_by_id(distributor_type_id)

        group_query_manager = manager_factory.repo_group_query_manager()

        # Validation
        group = group_query_manager.get_group(group_id)
        distributor_type_id = distributor["distributor_type_id"]

        # Assemble the data needed for publish
        conduit = RepoGroupPublishConduit(group_id, distributor)

        call_config = PluginCallConfiguration(plugin_config, distributor["config"], publish_config_override)
        transfer_group = common_utils.to_transfer_repo_group(group)
        transfer_group.working_dir = common_utils.group_distributor_working_dir(distributor_type_id, group_id)

        # TODO: Add events for group publish start/complete
        RepoGroupPublishManager._do_publish(transfer_group, distributor_id, distributor_instance, conduit, call_config)

    def put(self, request, repo_group_id, distributor_id):
        """
        Change information about a single repo group distributor association.

        :param request: WSGI request object
        :type  request: django.core.handlers.wsgi.WSGIRequest
        :param repo_group_id: repo group the distributor is associated with
        :type  repo_group_id: str
        :param distributor_id: distributor to update
        :type  distributor_id: str

        :return: response containing a serialized dict of the modified distributor association
        :rtype: django.http.HttpResponse
        :raises pulp_exceptions.MissingValue: if param 'distributor_config' is not in the body
        """
        params = request.body_as_json
        distributor_config = params.get('distributor_config', None)
        if distributor_config is None:
            raise pulp_exceptions.MissingValue(['distributor_config'])
        distributor_manager = managers_factory.repo_group_distributor_manager()
        result = distributor_manager.update_distributor_config(repo_group_id, distributor_id,
                                                               distributor_config)
        result['_href'] = reverse(
            'repo_group_distributor_resource',
            kwargs={'repo_group_id': repo_group_id, 'distributor_id': distributor_id}
        )
        return generate_json_response_with_pulp_encoder(result)

    def PUT(self, repo_group_id, distributor_id):
        params = self.params()

        distributor_config = params.get('distributor_config', None)

        if distributor_config is None:
            raise pulp_exceptions.MissingValue(['distributor_config'])

        distributor_manager = managers_factory.repo_group_distributor_manager()

        resources = {
            dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE :
                    {repo_group_id : dispatch_constants.RESOURCE_UPDATE_OPERATION},
            dispatch_constants.RESOURCE_REPOSITORY_GROUP_DISTRIBUTOR_TYPE :
                    {distributor_id : dispatch_constants.RESOURCE_UPDATE_OPERATION},
            }
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE, repo_group_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_DISTRIBUTOR_TYPE, distributor_id),
                action_tag('update_distributor')
        ]

        call_request = CallRequest(distributor_manager.update_distributor_config,
                                   args=[repo_group_id, distributor_id, distributor_config],
                                   resources=resources,
                                   tags=tags,
                                   archive=True)

        result = execution.execute(call_request)

        href = serialization.link.current_link_obj()
        result.update(href)

        return self.ok(result)

    def DELETE(self, repo_group_id, distributor_id):
        params = self.params()
        force = params.get('force', False)

        distributor_manager = managers_factory.repo_group_distributor_manager()

        resources = {
            dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE :
                    {repo_group_id : dispatch_constants.RESOURCE_UPDATE_OPERATION},
            dispatch_constants.RESOURCE_REPOSITORY_GROUP_DISTRIBUTOR_TYPE :
                    {distributor_id : dispatch_constants.RESOURCE_DELETE_OPERATION},
                     }
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE, repo_group_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_DISTRIBUTOR_TYPE, distributor_id),
                action_tag('remove_distributor')
               ]
        call_request = CallRequest(distributor_manager.remove_distributor,
                                   args=[repo_group_id, distributor_id],
                                   kwargs={'force' : force},
                                   resources=resources,
                                   tags=tags,
                                   archive=True)

        execution.execute(call_request)
        return self.ok(None)

    def delete_repo_group(self, group_id):
        """
        Delete a repo group.
        @param group_id: unique id of the repo group to delete
        @type group_id: str
        """
        collection = validate_existing_repo_group(group_id)

        # Delete all distributors on the group
        distributor_manager = manager_factory.repo_group_distributor_manager()
        distributors = distributor_manager.find_distributors(group_id)
        for distributor in distributors:
            distributor_manager.remove_distributor(group_id, distributor['id'])

        # Delete the working directory for the group
        working_dir = common_utils.repo_group_working_dir(group_id)
        if os.path.exists(working_dir):
            try:
                shutil.rmtree(working_dir)
            except Exception:
                _LOG.exception('Error while deleting working dir [%s] for repo group [%s]' % (working_dir, group_id))
                raise

        # Delete from the database
        collection.remove({'id': group_id}, safe=True)

    def post(self, request, repo_group_id):
        """
        Asssociate a distributor with a repo group.

        :param request: WSGI request object
        :type  request: django.core.handlers.wsgi.WSGIRequest
        :param repo_group_id: matching distributors will be associated with this repo group
        :type  repo_group_id: str

        :return: response containing a serialized dict of the distributor association
        :rtype: django.http.HttpResponse
        """
        # Params (validation will occur in the manager)
        params = request.body_as_json
        distributor_type_id = params.get(distributor_constants.DISTRIBUTOR_TYPE_ID_KEY, None)
        distributor_config = params.get(distributor_constants.DISTRIBUTOR_CONFIG_KEY, None)
        distributor_id = params.get(distributor_constants.DISTRIBUTOR_ID_KEY, None)
        distributor_manager = managers_factory.repo_group_distributor_manager()
        created = distributor_manager.add_distributor(repo_group_id, distributor_type_id,
                                                      distributor_config, distributor_id)
        created['_href'] = reverse('repo_group_distributor_resource',
                                   kwargs={'repo_group_id': repo_group_id,
                                           'distributor_id': created['id']})
        response = generate_json_response_with_pulp_encoder(created)
        return generate_redirect_response(response, created['_href'])

    def DELETE(self, repo_group_id, distributor_id):
        params = self.params()
        force = params.get('force', False)

        distributor_manager = managers_factory.repo_group_distributor_manager()
        distributor_manager.remove_distributor(repo_group_id, distributor_id, force=force)
        return self.ok(None)

    def setUp(self):
        super(RepoGroupDistributorTests, self).setUp()

        mock_plugins.install()

        self.manager = manager_factory.repo_group_manager()
        self.distributor_manager = manager_factory.repo_group_distributor_manager()

 def _get_distributor_instance_and_config(self, group_id, distributor_id):
     # separated out convenience method for use in testing
     distributor_manager = manager_factory.repo_group_distributor_manager()
     distributor = distributor_manager.get_distributor(group_id, distributor_id)
     distributor_type_id = distributor['distributor_type_id']
     distributor_instance, plugin_config = plugin_api.get_group_distributor_by_id(distributor_type_id)
     return distributor, distributor_instance, plugin_config

    def POST(self, repo_group_id):
        # Params (validation will occur in the manager)
        params = self.params()
        distributor_type_id = params.get('distributor_type_id', None)
        distributor_config = params.get('distributor_config', None)
        distributor_id = params.get('distributor_id', None)

        distributor_manager = managers_factory.repo_group_distributor_manager()

        resources = {dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE : {
            repo_group_id : dispatch_constants.RESOURCE_UPDATE_OPERATION
        }}
        weight = pulp_config.config.getint('tasks', 'create_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_TYPE, repo_group_id),
                action_tag('add_distributor')]
        if distributor_id is not None:
            tags.append(resource_tag(dispatch_constants.RESOURCE_REPOSITORY_GROUP_DISTRIBUTOR_TYPE, distributor_id))

        call_request = CallRequest(distributor_manager.add_distributor,
                                   [repo_group_id, distributor_type_id, distributor_config, distributor_id],
                                   resources=resources,
                                   weight=weight,
                                   tags=tags)
        created = execution.execute(call_request)

        href = serialization.link.child_link_obj(created['id'])
        created.update(href)

        return self.created(href['_href'], created)

    def GET(self, repo_group_id, distributor_id):
        distributor_manager = managers_factory.repo_group_distributor_manager()
        dist = distributor_manager.get_distributor(repo_group_id, distributor_id)

        href = serialization.link.current_link_obj()
        dist.update(href)

        return self.ok(dist)

    def setUp(self):
        super(RepoGroupDistributorManagerTests, self).setUp()
        mock_plugins.install()

        self.group_manager = manager_factory.repo_group_manager()
        self.distributor_manager = manager_factory.repo_group_distributor_manager()

        self.group_id = 'test-group'
        self.group_manager.create_repo_group(self.group_id)

    def setUp(self):
        super(RepoGroupPublishManagerTests, self).setUp()
        mock_plugins.install()

        self.group_manager = manager_factory.repo_group_manager()
        self.distributor_manager = manager_factory.repo_group_distributor_manager()
        self.publish_manager = manager_factory.repo_group_publish_manager()

        self.group_id = 'publish-group'
        self.group_manager.create_repo_group(self.group_id)

        self.distributor_id = 'publish-dist'
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {}, distributor_id=self.distributor_id)

    def create_and_configure_repo_group(self, group_id, display_name=None, description=None,
                                        repo_ids=None, notes=None, distributor_list=None):
        """
        Create a new repository group and add distributors in a single call. This is equivalent to
        calling RepoGroupManager.create_repo_group and then RepoGroupDistributorManager.add_distributor
        for each distributor in the distributor list.

        :param group_id: unique id of the repository group
        :type group_id: str
        :param display_name: user-friendly name of the repository id
        :type display_name: str or None
        :param description: description of the repository group
        :type description: str or None
        :param repo_ids: the list of repository ids in this repository group
        :type repo_ids: list of str or None
        :param notes: A collection of key=value pairs
        :type notes: dict or None
        :param distributor_list: A list of dictionaries used to add distributors. The following keys are
            expected: from pulp.common.constants: DISTRIBUTOR_TYPE_ID_KEY, DISTRIBUTOR_CONFIG_KEY, and
            DISTRIBUTOR_ID_KEY, which should hold values str, dict, and str or None
        :type distributor_list: list of dict
        :return: SON representation of the repo group
        :rtype: bson.SON
        """
        if distributor_list is None:
            distributor_list = ()

        # Validate the distributor list before creating a repo group
        if not isinstance(distributor_list, (list, tuple)) or not \
                all(isinstance(dist, dict) for dist in distributor_list):
            raise pulp_exceptions.InvalidValue(['distributor_list'])

        # Create the repo group using the vanilla group create method
        repo_group = self.create_repo_group(group_id, display_name, description, repo_ids, notes)

        distributor_manager = manager_factory.repo_group_distributor_manager()

        for distributor in distributor_list:
            try:
                # Attempt to add the distributor to the group.
                type_id = distributor.get(distributor_constants.DISTRIBUTOR_TYPE_ID_KEY)
                plugin_config = distributor.get(distributor_constants.DISTRIBUTOR_CONFIG_KEY)
                distributor_id = distributor.get(distributor_constants.DISTRIBUTOR_ID_KEY)
                distributor_manager.add_distributor(group_id, type_id, plugin_config, distributor_id)
            except Exception, e:
                # If an exception occurs, pass it on after cleaning up the repository group
                _LOG.exception('Exception adding distributor to repo group [%s]; the group will'
                               ' be deleted' % group_id)
                self.delete_repo_group(group_id)
                raise e, None, sys.exc_info()[2]

    def setUp(self):
        super(RepoGroupPublishConduitTests, self).setUp()
        mock_plugins.install()
        manager_factory.initialize()

        self.group_manager = manager_factory.repo_group_manager()
        self.distributor_manager = manager_factory.repo_group_distributor_manager()

        self.group_id = 'conduit-group'
        self.distributor_id = 'conduit-distributor'

        self.group_manager.create_repo_group(self.group_id)
        self.distributor_manager.add_distributor(self.group_id, 'mock-group-distributor', {}, distributor_id=self.distributor_id)

        self.conduit = RepoGroupPublishConduit(self.group_id, self.distributor_id)

    def POST(self, repo_group_id):
        # Params (validation will occur in the manager)
        params = self.params()
        distributor_type_id = params.get(distributor_constants.DISTRIBUTOR_TYPE_ID_KEY, None)
        distributor_config = params.get(distributor_constants.DISTRIBUTOR_CONFIG_KEY, None)
        distributor_id = params.get(distributor_constants.DISTRIBUTOR_ID_KEY, None)

        distributor_manager = managers_factory.repo_group_distributor_manager()

        created = distributor_manager.add_distributor(repo_group_id, distributor_type_id,
                                                      distributor_config, distributor_id)

        href = serialization.link.child_link_obj(created['id'])
        created.update(href)

        return self.created(href['_href'], created)

    def PUT(self, repo_group_id, distributor_id):
        params = self.params()

        distributor_config = params.get('distributor_config', None)

        if distributor_config is None:
            raise pulp_exceptions.MissingValue(['distributor_config'])

        distributor_manager = managers_factory.repo_group_distributor_manager()

        result = distributor_manager.update_distributor_config(repo_group_id, distributor_id,
                                                               distributor_config)

        href = serialization.link.current_link_obj()
        result.update(href)

        return self.ok(result)

    def get_scratchpad(self):
        """
        Returns the value set in the scratchpad for this repository group. If no
        value has been set, None is returned.

        @return: value saved for the repository group and this distributor
        @rtype:  object

        @raises DistributorConduitException: wraps any exception that may occur
                in the Pulp server
        """
        try:
            distributor_manager = manager_factory.repo_group_distributor_manager()
            value = distributor_manager.get_distributor_scratchpad(self.group_id, self.distributor_id)
            return value
        except Exception, e:
            _LOG.exception('Error getting scratchpad for repository [%s]' % self.group_id)
            raise DistributorConduitException(e), None, sys.exc_info()[2]

    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 set_scratchpad(self, value):
        """
        Saves the given value to the scratchpad for this repository group. It
        can later be retrieved in subsequent syncs through get_scratchpad. The
        type for the given value is anything that can be stored in the database
        (string, list, dict, etc.).

        @param value: will overwrite the existing scratchpad
        @type  value: object

        @raises DistributorConduitException: wraps any exception that may occur
                in the Pulp server
        """
        try:
            distributor_manager = manager_factory.repo_group_distributor_manager()
            distributor_manager.set_distributor_scratchpad(self.group_id, self.distributor_id, value)
        except Exception, e:
            _LOG.exception('Error setting scratchpad for repository [%s]' % self.group_id)
            raise DistributorConduitException(e), None, sys.exc_info()[2]