How to use the cornice.validators.colander_validator function in cornice

        @service.put(validators=(colander_validator, ), schema=PutRequestSchema())
        def view_put(request):
            """Add flavour"""
            return request.validated

def test_with_klass(self):
        class TemperatureCooler(object):
            """Temp class docstring"""

            def put_view(self):
                """Temp view docstring"""
                pass

        service = Service(
            "TemperatureCooler", "/freshair", klass=TemperatureCooler)
        service.add_view(
            "put",
            "put_view",
            validators=(colander_validator, ),
            schema=RequestSchema())
        ret = _generate_swagger([service])
        self.assertEqual(ret["info"], {
            'version': '0.1',
            'contact': {
                'name': 'Joe Smith',
                'email': 'joe.cool@swagger.com'
            },
            'title': 'Joes API'
        })
        self.assertEqual(ret["basePath"], '/jcool')
        self.assertEqual(ret["swagger"], '2.0')
        self.assertEqual(ret["tags"], [{'name': 'freshair'}])
        self.assertEqual(ret["paths"]["/freshair"]["put"]["summary"],
                         'Temp view docstring')
        params = ret["paths"]["/freshair"]["put"]['parameters']

def test_no_schema(self):
        request = DummyRequest()
        request.validated = mock.sentinel.validated
        colander_validator(request)
        self.assertEqual(request.validated, mock.sentinel.validated)
        self.assertEqual(len(request.errors), 0)

def test_ignore_multiple_views_by_ctype(self):

        service = Service("IceCream", "/icecream/{flavour}")

        class IceCream(object):
            def view_put(self, request):
                return "red"

        service.add_view(
            "put",
            IceCream.view_put,
            validators=(colander_validator, ),
            schema=PutRequestSchema(),
            content_type='application/json',
        )
        service.add_view(
            "put",
            IceCream.view_put,
            validators=(colander_validator, ),
            schema=PutRequestSchema(),
            content_type='text/xml',
        )

        swagger = CorniceSwagger([service])
        swagger.ignore_ctypes = ['text/xml']
        spec = swagger.generate()
        self.assertEquals(spec['paths']['/icecream/{flavour}']['put']['consumes'],
                          ['application/json'])

default_arguments=default_arguments,
            default_object_arguments=default_object_arguments,
            object_get_arguments=object_get_arguments,
        )

        viewset.responses = mock.MagicMock()
        arguments = viewset.object_arguments(mock.MagicMock(), "get")

        self.assertDictEqual(
            arguments,
            {
                "schema": mocked().bind(),
                "cors_headers": mock.sentinel.default_cors_headers,
                "error_handler": mock.sentinel.default_object_error_handler,
                "cors_origins": mock.sentinel.object_get_cors_origin,
                "validators": [colander_validator],
                "response_schemas": viewset.responses.get_and_bind(),
            },

@changeset.get(schema=ChangeSetSchema(), permission="read", validators=(colander_validator,))
def get_changeset(request):
    bid = request.matchdict["bid"]
    cid = request.matchdict["cid"]
    bucket_uri = instance_uri(request, "bucket", id=bid)
    collection_uri = instance_uri(request, "collection", bucket_id=bid, id=cid)

    queryparams = request.validated["querystring"]
    filters = []
    if "_since" in queryparams:
        filters = [Filter('last_modified', queryparams["_since"], COMPARISON.GT)]

    storage = request.registry.storage

    # We'll make sure that data isn't changed while we read metadata, changes, etc.
    before = storage.resource_timestamp(resource_name="record", parent_id=collection_uri)
    # Fetch collection metadata.

    validators=(colander_validator,),
    content_type=CONTENT_TYPES,
    permission=NO_PERMISSION_REQUIRED,
    tags=["Batch"],
    operation_id="batch",
    response_schemas=batch_responses,
)
def post_batch(request):
    requests = request.validated["body"]["requests"]

    request.log_context(batch_size=len(requests))

    limit = request.registry.settings["batch_max_requests"]
    if limit and len(requests) > int(limit):
        error_msg = f"Number of requests is limited to {limit}"
        request.errors.add("body", "requests", error_msg)
        return

args.update(**method_args)

        by_method = f"{endpoint_type}_{method.lower()}_arguments"
        endpoint_args = getattr(self, by_method, {})
        args.update(**endpoint_args)

        request_schema = args.get("schema", RequestSchema())
        object_schema = self.get_object_schema(resource_cls, method)
        request_schema = request_schema.bind(body=object_schema)
        response_schemas = self.responses.get_and_bind(endpoint_type, method, object=object_schema)

        args["schema"] = request_schema
        args["response_schemas"] = response_schemas

        validators = args.get("validators", [])
        validators.append(colander_validator)
        args["validators"] = validators

        return args

             validators=(colander_validator,))
def portier_verify(request):
    """Helper to redirect client towards Portier login form."""
    broker_uri = portier_conf(request, 'broker_uri')
    token = request.validated['body']['id_token']
    # Get the data from the config because the request might only
    # have local network information and not the public facing ones.
    audience = '{scheme}://{host}'.format(scheme=request.registry.settings['http_scheme'],
                                          host=request.registry.settings['http_host'])

    try:
        email, stored_redirect = get_verified_email(
            broker_url=broker_uri,
            token=token,
            audience=audience,
            issuer=broker_uri,
            cache=request.registry.cache)

args.update(**method_args)

        by_method = f"{endpoint_type}_{method.lower()}_arguments"
        endpoint_args = getattr(self, by_method, {})
        args.update(**endpoint_args)

        request_schema = args.get("schema", RequestSchema())
        object_schema = self.get_object_schema(resource_cls, method)
        request_schema = request_schema.bind(body=object_schema)
        response_schemas = self.responses.get_and_bind(endpoint_type, method, object=object_schema)

        args["schema"] = request_schema
        args["response_schemas"] = response_schemas

        validators = args.get("validators", [])
        validators.append(colander_validator)
        args["validators"] = validators

        return args