def POST(self, repo_id):
        # Params
        params = self.params()
        query = params.get('criteria', {})
        options = params.get('options', {})
        timeout = params.get('timeout', 60)

        try:
            criteria = UnitAssociationCriteria.from_client_input(query)
        except:
            _LOG.exception('Error parsing association criteria [%s]' % query)
            raise exceptions.PulpDataException(), None, sys.exc_info()[2]

        try:
            timeout = int(timeout)
        except ValueError:
            raise exceptions.InvalidValue(['timeout']), None, sys.exc_info()[2]

        # Coordinator configuration
        resources = {dispatch_constants.RESOURCE_REPOSITORY_TYPE: {repo_id: dispatch_constants.RESOURCE_READ_OPERATION}}
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                action_tag('resolve_dependencies')]

        dependency_manager = manager_factory.dependency_manager()
        call_request = CallRequest(dependency_manager.resolve_dependencies_by_criteria,
                                   [repo_id, criteria, options],
                                   resources=resources, tags=tags, archive=True)

        return execution.execute_sync_ok(self, call_request, timeout=timedelta(seconds=timeout))

    def _generate_response(cls, query, options, *args, **kwargs):
        """
        Perform the database query using the given search data, and return the resuls as a JSON
        serialized HttpReponse object.

        This overrides the base class so we can validate repo existance and to choose the search
        method depending on how many unit types we are dealing with.

        :param query: The criteria that should be used to search for objects
        :type  query: dict
        :param options: additional options for including extra data
        :type  options: dict

        :return:      The serialized search results in an HttpReponse
        :rtype:       django.http.HttpResponse
        """
        repo_id = kwargs.get('repo_id')
        model.Repository.objects.get_repo_or_missing_resource(repo_id)
        criteria = UnitAssociationCriteria.from_client_input(query)
        manager = manager_factory.repo_unit_association_query_manager()
        if criteria.type_ids is not None and len(criteria.type_ids) == 1:
            type_id = criteria.type_ids[0]
            units = manager.get_units_by_type(repo_id, type_id, criteria=criteria)
        else:
            units = manager.get_units(repo_id, criteria=criteria)
        for unit in units:
            content.remap_fields_with_serializer(unit['metadata'])
        return generate_json_response_with_pulp_encoder(units)

    def POST(self, repo_id):

        params = self.params()
        criteria = params.get("criteria", None)

        if criteria is not None:
            try:
                criteria = UnitAssociationCriteria.from_client_input(criteria)
            except:
                logger.error("Error parsing unassociation criteria [%s]" % criteria)
                raise exceptions.PulpDataException(), None, sys.exc_info()[2]

        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id), action_tag("unassociate")]
        async_result = unassociate_by_criteria.apply_async_with_reservation(
            dispatch_constants.RESOURCE_REPOSITORY_TYPE,
            repo_id,
            [
                repo_id,
                criteria,
                RepoContentUnit.OWNER_TYPE_USER,
                manager_factory.principal_manager().get_principal()["login"],
            ],
            tags=tags,
        )
        raise exceptions.OperationPostponed(async_result)

    def post(self, request, dest_repo_id):
        """
        Associate units matching the criteria into the given repository

        :param request: WSGI request object
        :type  request: django.core.handlers.wsgi.WSGIRequest
        :param dest_repo_id: id of the repository content will be copied to
        :type  dest_repo_id: str

        :raises exceptions.MissingValue: if required param source_repo_id is not passed
        :raises exceptions.InvalidValue: if source_repo_id is not found or if criteria
                                              params cannot be parsed
        :raises exceptions.OperationPostponed: dispatch a publish repo task
        """
        model.Repository.objects.get_repo_or_missing_resource(dest_repo_id)
        criteria_body = request.body_as_json.get('criteria', {})
        overrides = request.body_as_json.get('override_config', None)
        source_repo_id = request.body_as_json.get('source_repo_id', None)
        if source_repo_id is None:
            raise exceptions.MissingValue(['source_repo_id'])

        # Catch MissingResource because this is body data, raise 400 rather than 404
        try:
            model.Repository.objects.get_repo_or_missing_resource(source_repo_id)
        except exceptions.MissingResource:
            raise exceptions.InvalidValue(['source_repo_id'])

        try:
            criteria = UnitAssociationCriteria.from_client_input(criteria_body)
        except exceptions.InvalidValue, e:
            invalid_criteria = exceptions.InvalidValue('criteria')
            invalid_criteria.add_child_exception(e)
            raise invalid_criteria

    def unassociate_by_criteria(cls, repo_id, criteria, notify_plugins=True):
        """
        Unassociate units that are matched by the given criteria.

        :param repo_id:        identifies the repo
        :type  repo_id:        str
        :param criteria:
        :param notify_plugins: if true, relevant plugins will be informed of the removal
        :type  notify_plugins: bool
        """
        criteria = UnitAssociationCriteria.from_dict(criteria)
        repo = model.Repository.objects.get_repo_or_missing_resource(repo_id)

        unassociate_units = load_associated_units(repo_id, criteria)

        if len(unassociate_units) == 0:
            return {}

        # Convert the units into transfer units. This happens regardless of whether or not
        # the plugin will be notified as it's used to generate the return result.
        # If all source types have been converted to mongo, search via new style.
        repo_unit_types = set(repo.content_unit_counts.keys())
        if repo_unit_types.issubset(set(plugin_api.list_unit_models())):
            transfer_units = list(cls._units_from_criteria(repo, criteria))
        else:
            transfer_units = None
            if unassociate_units is not None:
                transfer_units = list(create_transfer_units(unassociate_units))

        if notify_plugins:
            remove_from_importer(repo_id, transfer_units)

        unit_map = {}  # maps unit_type_id to a list of unit_ids

        for unit in unassociate_units:
            id_list = unit_map.setdefault(unit["unit_type_id"], [])
            id_list.append(unit["unit_id"])

        collection = RepoContentUnit.get_collection()

        for unit_type_id, unit_ids in unit_map.items():
            spec = {"repo_id": repo_id, "unit_type_id": unit_type_id, "unit_id": {"$in": unit_ids}}
            collection.remove(spec)

            unique_count = sum(
                1
                for unit_id in unit_ids
                if not RepoUnitAssociationManager.association_exists(repo_id, unit_id, unit_type_id)
            )
            if not unique_count:
                continue

            repo_controller.update_unit_count(repo_id, unit_type_id, -unique_count)

        repo_controller.update_last_unit_removed(repo_id)

        # Match the return type/format as copy
        serializable_units = [u.to_id_dict() for u in transfer_units]

        return {"units_successful": serializable_units}

    def test_unassociate_via_criteria_no_matches(self):
        self.manager.associate_unit_by_id(self.repo_id, "type-1", "unit-1", OWNER_TYPE_USER, "admin")
        self.manager.associate_unit_by_id(self.repo_id, "type-1", "unit-2", OWNER_TYPE_USER, "admin")

        criteria_doc = {"type_ids": ["type-2"]}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        self.manager.unassociate_by_criteria(self.repo_id, criteria, OWNER_TYPE_USER, "admin")

        self.assertTrue(self.manager.association_exists(self.repo_id, "unit-1", "type-1"))
        self.assertTrue(self.manager.association_exists(self.repo_id, "unit-2", "type-1"))

    def test_unassociate_via_criteria(self):
        self.manager.associate_unit_by_id(self.repo_id, self.unit_type_id, self.unit_id, OWNER_TYPE_USER, "admin")
        self.manager.associate_unit_by_id(self.repo_id, self.unit_type_id, self.unit_id_2, OWNER_TYPE_USER, "admin")

        criteria_doc = {"filters": {"association": {"unit_id": {"$in": [self.unit_id, "unit-X"]}}}}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        self.manager.unassociate_by_criteria(self.repo_id, criteria, OWNER_TYPE_USER, "admin")

        self.assertFalse(self.manager.association_exists(self.repo_id, self.unit_id, self.unit_type_id))
        self.assertTrue(self.manager.association_exists(self.repo_id, self.unit_id_2, self.unit_type_id))

    def test_unassociate_via_criteria_no_matches(self):
        self.manager.associate_unit_by_id('repo-1', 'type-1', 'unit-1', OWNER_TYPE_USER, 'admin')
        self.manager.associate_unit_by_id('repo-1', 'type-1', 'unit-2', OWNER_TYPE_USER, 'admin')

        criteria_doc = {'type_ids': ['type-2']}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        self.manager.unassociate_by_criteria('repo-1', criteria, OWNER_TYPE_USER, 'admin')

        self.assertTrue(self.manager.association_exists('repo-1', 'unit-1', 'type-1'))
        self.assertTrue(self.manager.association_exists('repo-1', 'unit-2', 'type-1'))

    def unassociate_by_criteria(repo_id, criteria, notify_plugins=True):
        """
        Unassociate units that are matched by the given criteria.

        :param repo_id:        identifies the repo
        :type  repo_id:        str
        :param criteria:
        :param notify_plugins: if true, relevant plugins will be informed of the removal
        :type  notify_plugins: bool
        """
        criteria = UnitAssociationCriteria.from_dict(criteria)
        association_query_manager = manager_factory.repo_unit_association_query_manager()
        unassociate_units = association_query_manager.get_units(repo_id, criteria=criteria)

        if len(unassociate_units) == 0:
            return {}

        unit_map = {}  # maps unit_type_id to a list of unit_ids

        for unit in unassociate_units:
            id_list = unit_map.setdefault(unit['unit_type_id'], [])
            id_list.append(unit['unit_id'])

        collection = RepoContentUnit.get_collection()

        for unit_type_id, unit_ids in unit_map.items():
            spec = {'repo_id': repo_id,
                    'unit_type_id': unit_type_id,
                    'unit_id': {'$in': unit_ids}
                    }
            collection.remove(spec)

            unique_count = sum(
                1 for unit_id in unit_ids if not RepoUnitAssociationManager.association_exists(
                    repo_id, unit_id, unit_type_id))
            if not unique_count:
                continue

            repo_controller.update_unit_count(repo_id, unit_type_id, -unique_count)

        repo_controller.update_last_unit_removed(repo_id)

        # Convert the units into transfer units. This happens regardless of whether or not
        # the plugin will be notified as it's used to generate the return result,
        transfer_units = create_transfer_units(unassociate_units)

        if notify_plugins:
            remove_from_importer(repo_id, transfer_units)

        # Match the return type/format as copy
        serializable_units = [u.to_id_dict() for u in transfer_units]

        return {'units_successful': serializable_units}

    def test_unassociate_via_criteria_no_matches(self, mock_ctrl):
        self.manager.associate_unit_by_id(self.repo_id, 'type-1', 'unit-1')
        self.manager.associate_unit_by_id(self.repo_id, 'type-1', 'unit-2')

        criteria_doc = {'type_ids': ['type-2']}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        result = self.manager.unassociate_by_criteria(self.repo_id, criteria)
        self.assertEquals(result, {})

        self.assertTrue(self.manager.association_exists(self.repo_id, 'unit-1', 'type-1'))
        self.assertTrue(self.manager.association_exists(self.repo_id, 'unit-2', 'type-1'))

    def test_unassociate_via_criteria(self, mock_repo_qs, mock_ctrl):
        self.manager.associate_unit_by_id(self.repo_id, self.unit_type_id, self.unit_id)
        self.manager.associate_unit_by_id(self.repo_id, self.unit_type_id, self.unit_id_2)

        criteria_doc = {'filters': {'association': {'unit_id': {'$in': [self.unit_id, 'unit-X']}}}}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        self.manager.unassociate_by_criteria(self.repo_id, criteria)

        self.assertFalse(self.manager.association_exists(self.repo_id, self.unit_id,
                                                         self.unit_type_id))
        self.assertTrue(self.manager.association_exists(self.repo_id, self.unit_id_2,
                                                        self.unit_type_id))
        mock_repo_qs.get_repo_or_missing_resource.assert_called_once_with(self.repo_id)

    def test_unassociate_via_criteria(self):
        self.manager.associate_unit_by_id('repo-1', 'type-1', 'unit-1', OWNER_TYPE_USER, 'admin')
        self.manager.associate_unit_by_id('repo-1', 'type-1', 'unit-2', OWNER_TYPE_USER, 'admin')
        self.manager.associate_unit_by_id('repo-1', 'type-1', 'unit-3', OWNER_TYPE_USER, 'admin')
        self.manager.associate_unit_by_id('repo-1', 'type-2', 'unit-1', OWNER_TYPE_IMPORTER, 'yum')
        self.manager.associate_unit_by_id('repo-1', 'type-2', 'unit-2', OWNER_TYPE_IMPORTER, 'yum')

        criteria_doc = {'filters': {'association': {'unit_id': {'$in': ['unit-1', 'unit-3']}}}}

        criteria = UnitAssociationCriteria.from_client_input(criteria_doc)

        self.manager.unassociate_by_criteria('repo-1', criteria, OWNER_TYPE_USER, 'admin')

        self.assertFalse(self.manager.association_exists('repo-1', 'unit-1', 'type-1'))
        self.assertTrue(self.manager.association_exists('repo-1', 'unit-2', 'type-1'))
        self.assertFalse(self.manager.association_exists('repo-1', 'unit-3', 'type-1'))
        self.assertTrue(self.manager.association_exists('repo-1', 'unit-1', 'type-2'))
        self.assertTrue(self.manager.association_exists('repo-1', 'unit-2', 'type-2'))

    def POST(self, repo_id):

        params = self.params()
        criteria = params.get('criteria', None)

        if criteria is not None:
            try:
                criteria = UnitAssociationCriteria.from_client_input(criteria)
            except:
                _logger.error('Error parsing unassociation criteria [%s]' % criteria)
                raise exceptions.PulpDataException(), None, sys.exc_info()[2]

        task_tags = [tags.resource_tag(tags.RESOURCE_REPOSITORY_TYPE, repo_id),
                     tags.action_tag('unassociate')]
        async_result = unassociate_by_criteria.apply_async_with_reservation(
            tags.RESOURCE_REPOSITORY_TYPE, repo_id,
            [repo_id, criteria, RepoContentUnit.OWNER_TYPE_USER,
             manager_factory.principal_manager().get_principal()['login']], tags=task_tags)
        raise exceptions.OperationPostponed(async_result)

    def test_parse_criteria(self):
        # Setup
        query = {
            'type_ids': ['rpm'],
            'filters': {
                'unit': {'$and': [
                    {'$regex': '^p.*'},
                    {'$not': 'ython$'},
                ]},
                'association': {'created': {'$gt': 'now'}},
            },

            'limit': 100,
            'skip': 200,
            'fields': {
                'unit': ['name', 'version'],
                'association': ['created'],
            },
            'remove_duplicates': True,
        }

        # Test
        criteria = UnitAssociationCriteria.from_client_input(query)

        # Verify
        self.assertEqual(criteria.type_ids, ['rpm'])
        self.assertEqual(criteria.association_filters, {'created': {'$gt': 'now'}})
        self.assertEqual(criteria.limit, 100)
        self.assertEqual(criteria.skip, 200)
        self.assertEqual(criteria.unit_fields, ['name', 'version'])
        self.assertEqual(criteria.association_fields, ['created', 'unit_id', 'unit_type_id'])
        self.assertEqual(criteria.remove_duplicates, True)

        #   Check the special $not handling in the unit filter
        self.assertTrue('$and' in criteria.unit_filters)
        and_list = criteria.unit_filters['$and']

        self.assertTrue('$regex' in and_list[0])
        self.assertEqual(and_list[0]['$regex'], '^p.*')

        self.assertTrue('$not' in and_list[1])
        self.assertEqual(and_list[1]['$not'], re.compile('ython$'))

    def POST(self, dest_repo_id):

        # Params
        params = self.params()
        source_repo_id = params.get('source_repo_id', None)
        overrides = params.get('override_config', None)

        if source_repo_id is None:
            raise exceptions.MissingValue(['source_repo_id'])

        # A 404 only applies to things in the URL, so the destination repo
        # check allows the MissingResource to bubble up, but if the source
        # repo doesn't exist, it's considered bad data.
        repo_query_manager = manager_factory.repo_query_manager()
        repo_query_manager.get_repository(dest_repo_id)

        try:
            repo_query_manager.get_repository(source_repo_id)
        except exceptions.MissingResource:
            raise exceptions.InvalidValue(['source_repo_id'])

        criteria = params.get('criteria', None)
        if criteria is not None:
            try:
                criteria = UnitAssociationCriteria.from_client_input(criteria)
            except:
                _LOG.error('Error parsing association criteria [%s]' % criteria)
                raise exceptions.PulpDataException(), None, sys.exc_info()[2]

        association_manager = manager_factory.repo_unit_association_manager()
        resources = {dispatch_constants.RESOURCE_REPOSITORY_TYPE: {source_repo_id: dispatch_constants.RESOURCE_READ_OPERATION,
                                                                   dest_repo_id: dispatch_constants.RESOURCE_UPDATE_OPERATION}}
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, dest_repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, source_repo_id),
                action_tag('associate')]
        call_request = CallRequest(association_manager.associate_from_repo,
                                   [source_repo_id, dest_repo_id],
                                   {'criteria': criteria, 'import_config_override': overrides},
                                   resources=resources,
                                   tags=tags,
                                   archive=True)
        return execution.execute_async(self, call_request)

    def POST(self, repo_id):
        # Params
        params = self.params()
        query = params.get('criteria', {})
        options = params.get('options', {})
        timeout = params.get('timeout', 60)

        try:
            criteria = UnitAssociationCriteria.from_client_input(query)
        except:
            logger.error('Error parsing association criteria [%s]' % query)
            raise exceptions.PulpDataException(), None, sys.exc_info()[2]

        try:
            timeout = int(timeout)
        except ValueError:
            raise exceptions.InvalidValue(['timeout']), None, sys.exc_info()[2]

        dependency_manager = manager_factory.dependency_manager()
        result = dependency_manager.resolve_dependencies_by_criteria(repo_id, criteria, options)
        return self.ok(result)

    def test_get_units_by_type_not_query(self):
        """
        Mongo really doesn't like $not queries when regular expressions are
        involved. This test is to make sure that across mongo and pymongo
        versions a not expression against a regular expression continues to
        work.

        There is an important step in the parsing of the criteria from the
        REST call into the Criteria object. This call will use that method
        to more closely test the end to end experience.
        """

        # Setup

        # I got bit by the fact that incoming REST requests are in unicode;
        # the criteria parsing didn't account for this. This example specifically
        # replicates that by having the not and its value in unicode.

        query_string = {
            'filters' : {
                'unit' : {
                    'key_1' : {
                        u'$not' : u'.*aa.*'
                    }
                }
            }
        }
        #criteria = unit_association_criteria(query_string)
        criteria = UnitAssociationCriteria.from_client_input(query_string)

        # Test
        units = self.manager.get_units_by_type('repo-1', 'alpha', criteria)

        # Verify
        self.assertEqual(len(self.units['alpha']) - 1, len(units))
        for u in units:
            self.assertTrue(u['metadata']['key_1'] != 'aardvark')

    def associate_from_repo(cls, source_repo_id, dest_repo_id, criteria,
                            import_config_override=None):
        """
        Creates associations in a repository based on the contents of a source
        repository. Units from the source repository can be filtered by
        specifying a criteria object.

        The destination repository must have an importer that can support
        the types of units being associated. This is done by analyzing the
        unit list and the importer metadata and takes place before the
        destination repository is called.

        Pulp does not actually perform the associations as part of this call.
        The unit list is determined and passed to the destination repository's
        importer. It is the job of the importer to make the associate calls
        back into Pulp where applicable.

        If criteria is None, the effect of this call is to copy the source
        repository's associations into the destination repository.

        :param source_repo_id:         identifies the source repository
        :type  source_repo_id:         str
        :param dest_repo_id:           identifies the destination repository
        :type  dest_repo_id:           str
        :param criteria:               optional; if specified, will filter the units retrieved from
                                       the source repository
        :type  criteria:               pulp.server.db.model.criteria.UnitAssociationCriteria
        :param import_config_override: optional config containing values to use for this import only
        :type  import_config_override: dict
        :return:                       dict with key 'units_successful' whose
                                       value is a list of unit keys that were copied.
                                       units that were associated by this operation
        :rtype:                        dict
        :raise MissingResource:        if either of the specified repositories don't exist
        """
        criteria = UnitAssociationCriteria.from_dict(criteria)
        source_repo = model.Repository.objects.get_repo_or_missing_resource(source_repo_id)
        dest_repo = model.Repository.objects.get_repo_or_missing_resource(dest_repo_id)

        dest_repo_importer = model.Importer.objects.get_or_404(repo_id=dest_repo_id)
        source_repo_importer = model.Importer.objects.get_or_404(repo_id=source_repo_id)

        # The docs are incorrect on the list_importer_types call; it actually
        # returns a dict with the types under key "types" for some reason.
        supported_type_ids = set(plugin_api.list_importer_types(
            dest_repo_importer.importer_type_id)['types'])

        # Get the unit types from the repo source repo
        source_repo_unit_types = set(source_repo.content_unit_counts.keys())

        # Now we can make sure the destination repository's importer is capable
        # of importing either the selected units or all of the units
        if not source_repo_unit_types.issubset(supported_type_ids):
            raise exceptions.PulpCodedException(error_code=error_codes.PLP0044)
        transfer_units = None
        # if all source types have been converted to mongo - search via new style
        if source_repo_unit_types.issubset(set(plugin_api.list_unit_models())):
            transfer_units = cls._units_from_criteria(source_repo, criteria)
        else:
            # else, search via old style
            associate_us = load_associated_units(source_repo_id, criteria)
            # If units were supposed to be filtered but none matched, we're done
            if len(associate_us) == 0:
                # Return an empty list to indicate nothing was copied
                return {'units_successful': []}
            # Convert all of the units into the plugin standard representation if
            # a filter was specified
            transfer_units = None
            if associate_us is not None:
                transfer_units = create_transfer_units(associate_us)

        # Convert the two repos into the plugin API model
        transfer_dest_repo = dest_repo.to_transfer_repo()
        transfer_source_repo = source_repo.to_transfer_repo()

        # Invoke the importer
        importer_instance, plugin_config = plugin_api.get_importer_by_id(
            dest_repo_importer.importer_type_id)

        call_config = PluginCallConfiguration(plugin_config, dest_repo_importer.config,
                                              import_config_override)
        conduit = ImportUnitConduit(
            source_repo_id, dest_repo_id, source_repo_importer.importer_type_id,
            dest_repo_importer.importer_type_id)

        try:
            copied_units = importer_instance.import_units(
                transfer_source_repo, transfer_dest_repo, conduit, call_config,
                units=transfer_units)
            if isinstance(copied_units, tuple):
                suc_units_ids = [u.to_id_dict() for u in copied_units[0] if u is not None]
                unsuc_units_ids = [u.to_id_dict() for u in copied_units[1]]
                repo_controller.rebuild_content_unit_counts(dest_repo)
                return {'units_successful': suc_units_ids,
                        'units_failed_signature_filter': unsuc_units_ids}
            unit_ids = [u.to_id_dict() for u in copied_units if u is not None]
            repo_controller.rebuild_content_unit_counts(dest_repo)
            return {'units_successful': unit_ids}
        except Exception:
            msg = _('Exception from importer [%(i)s] while importing units into repository [%(r)s]')
#.........这里部分代码省略.........