openapi-core/openapi_core/validation/response/validators.py at 20f570d4220481a6bef32adec668e3c2a820106f · python-openapi/openapi-core · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
"""OpenAPI core validation response validators module"""
from __future__ import division

from openapi_core.casting.schemas.exceptions import CastError
from openapi_core.deserializing.exceptions import DeserializeError
from openapi_core.exceptions import MissingResponseContent
from openapi_core.templating.media_types.exceptions import MediaTypeFinderError
from openapi_core.templating.paths.exceptions import PathError
from openapi_core.templating.responses.exceptions import ResponseFinderError
from openapi_core.unmarshalling.schemas.enums import UnmarshalContext
from openapi_core.unmarshalling.schemas.exceptions import (
    UnmarshalError, ValidateError,
)
from openapi_core.unmarshalling.schemas.factories import (
    SchemaUnmarshallersFactory,
)
from openapi_core.validation.response.datatypes import ResponseValidationResult
from openapi_core.validation.validators import BaseValidator


class ResponseValidator(BaseValidator):

    @property
    def schema_unmarshallers_factory(self):
        spec_resolver = self.spec.accessor.dereferencer.resolver_manager.\
            resolver
        return SchemaUnmarshallersFactory(
            spec_resolver, self.format_checker,
            self.custom_formatters, context=UnmarshalContext.RESPONSE,
        )

    def validate(self, request, response):
        try:
            _, operation, _, _, _ = self._find_path(request)
        # don't process if operation errors
        except PathError as exc:
            return ResponseValidationResult(errors=[exc, ])

        try:
            operation_response = self._get_operation_response(
                operation, response)
        # don't process if operation errors
        except ResponseFinderError as exc:
            return ResponseValidationResult(errors=[exc, ])

        data, data_errors = self._get_data(response, operation_response)

        headers, headers_errors = self._get_headers(
            response, operation_response)

        errors = data_errors + headers_errors
        return ResponseValidationResult(
            errors=errors,
            data=data,
            headers=headers,
        )

    def _get_operation_response(self, operation, response):
        from openapi_core.templating.responses.finders import ResponseFinder
        finder = ResponseFinder(operation / 'responses')
        return finder.find(str(response.status_code))

    def _validate_data(self, request, response):
        try:
            _, operation, _, _, _ = self._find_path(request)
        # don't process if operation errors
        except PathError as exc:
            return ResponseValidationResult(errors=[exc, ])

        try:
            operation_response = self._get_operation_response(
                operation, response)
        # don't process if operation errors
        except ResponseFinderError as exc:
            return ResponseValidationResult(errors=[exc, ])

        data, data_errors = self._get_data(response, operation_response)
        return ResponseValidationResult(
            errors=data_errors,
            data=data,
        )

    def _get_data(self, response, operation_response):
        if 'content' not in operation_response:
            return None, []

        try:
            media_type, mimetype = self._get_media_type(
                operation_response / 'content', response)
        except MediaTypeFinderError as exc:
            return None, [exc, ]

        try:
            raw_data = self._get_data_value(response)
        except MissingResponseContent as exc:
            return None, [exc, ]

        try:
            deserialised = self._deserialise_data(mimetype, raw_data)
        except DeserializeError as exc:
            return None, [exc, ]

        try:
            casted = self._cast(media_type, deserialised)
        except CastError as exc:
            return None, [exc, ]

        if 'schema' not in media_type:
            return casted, []

        schema = media_type / 'schema'
        try:
            data = self._unmarshal(schema, casted)
        except (ValidateError, UnmarshalError) as exc:
            return None, [exc, ]

        return data, []

    def _get_headers(self, response, operation_response):
        errors = []

        # @todo: implement
        headers = {}

        return headers, errors

    def _get_data_value(self, response):
        if not response.data:
            raise MissingResponseContent(response)

        return response.data