Merge pull request #283 from macisamuele/maci-refactor-spec-flattening

Refactor spec flattening

bravado_core/spec_flattening.py

@@ -13,10 +13,12 @@
 from six.moves.urllib.parse import urlunparse
 from swagger_spec_validator.ref_validators import in_scope
 
+from bravado_core.model import model_discovery
 from bravado_core.model import MODEL_MARKER
 from bravado_core.schema import is_dict_like
 from bravado_core.schema import is_list_like
 from bravado_core.schema import is_ref
+from bravado_core.util import cached_property
 from bravado_core.util import determine_object_type
 from bravado_core.util import ObjectType
 
@@ -98,171 +100,197 @@ def _marshal_uri(target_uri, origin_uri):
     return marshalled_target
 
 
-def _warn_if_uri_clash_on_same_marshaled_representation(uri_schema_mappings, marshal_uri):
-    """
-    Verifies that all the uris present on the definitions are represented by a different marshaled uri.
-    If is not the case a warning will filed.
-
-    In case of presence of warning please keep us informed about the issue, in the meantime you can
-    workaround this calling directly ``flattened_spec(spec, marshal_uri_function)`` passing your
-    marshalling function.
-    """
-    # Check that URIs are NOT clashing to same marshaled representation
-    marshaled_uri_mapping = defaultdict(set)
-    for uri in iterkeys(uri_schema_mappings):
-        marshaled_uri_mapping[marshal_uri(uri)].add(uri)
-
-    if len(marshaled_uri_mapping) != len(uri_schema_mappings):
-        # At least two uris clashed to the same marshaled representation
-        for marshaled_uri, uris in iteritems(marshaled_uri_mapping):
-            if len(uris) > 1:
-                warnings.warn(
-                    message='{s_uris} clashed to {marshaled}'.format(
-                        s_uris=', '.join(sorted(urlunparse(uri) for uri in uris)),
-                        marshaled=marshaled_uri,
-                    ),
-                    category=Warning,
-                )
-
-
-def flattened_spec(swagger_spec, marshal_uri_function=_marshal_uri):
-    """
-    Flatten Swagger Specs description into an unique and JSON serializable document.
-    The flattening injects in place the referenced [path item objects](https://swagger.io/specification/#pathItemObject)
-    while it injects in '#/parameters' the [parameter objects](https://swagger.io/specification/#parameterObject),
-    in '#/definitions' the [schema objects](https://swagger.io/specification/#schemaObject) and in
-    '#/responses' the [response objects](https://swagger.io/specification/#responseObject).
-
-    Note: the object names in '#/definitions', '#/parameters' and '#/responses' are evaluated by
-    ``marshal_uri_function``, the default method takes care of creating unique names for all the used references.
-    Since name clashing are still possible take care that a warning could be filed.
-    If it happen please report to us the specific warning text and the specs that generated it.
-    We can work to improve it and in the mean time you can "plug" a custom marshalling function.
-
-    Note: https://swagger.io/specification/ has been update to track the latest version of the Swagger/OpenAPI specs.
-    Please refer to https://github.com/OAI/OpenAPI-Specification/blob/3.0.0/versions/2.0.md#responseObject for the
-    most recent Swagger 2.0 specifications.
-
-    WARNING: In the future releases all the parameters except swagger_spec and marshal_uri_function will be removed.
-    Please make sure to use only those two parameters.
-    Until the deprecation is not effective you can still pass all the parameters but it's strongly discouraged.
+class _SpecFlattener(object):
+    def __init__(self, swagger_spec, marshal_uri_function):
+        self.swagger_spec = swagger_spec
+        self.spec_url = self.swagger_spec.origin_url
 
-    :param swagger_spec: bravado-core Spec object
-    :type swagger_spec: bravado_core.spec.Spec
-    :param marshal_uri_function: function used to marshal uris in string suitable to be keys in Swagger Specs.
-    :type marshal_uri_function: Callable with the same signature of ``_marshal_uri``
+        if self.spec_url is None:
+            warnings.warn(
+                message='It is recommended to set origin_url to your spec before flattering it. '
+                        'Doing so internal paths will be hidden, reducing the amount of exposed information.',
+                category=Warning,
+            )
 
-    :return: Flattened representation of the Swagger Specs
-    :rtype: dict
-    """
-    spec_url = swagger_spec.origin_url
+        self.known_mappings = {
+            object_type.get_root_holder(): {}
+            for object_type in ObjectType
+            if object_type.get_root_holder()
+        }
+        self.marshal_uri_function = marshal_uri_function
 
-    if spec_url is None:
-        warnings.warn(
-            message='It is recommended to set origin_url to your spec before flattering it. '
-                    'Doing so internal paths will be hidden, reducing the amount of exposed information.',
-            category=Warning,
+    @cached_property
+    def marshal_uri(self):
+        return functools.partial(
+            self.marshal_uri_function,
+            origin_uri=urlparse(self.spec_url) if self.spec_url else None,
         )
 
-    known_mappings = {
-        object_type.get_root_holder(): {}
-        for object_type in ObjectType
-        if object_type.get_root_holder()
-    }
-
-    # Define marshal_uri method to be used by descend
-    marshal_uri = functools.partial(
-        marshal_uri_function,
-        origin_uri=urlparse(spec_url) if spec_url else None,
-    )
+    @cached_property
+    def spec_resolver(self):
+        return self.swagger_spec.resolver
 
-    # Avoid object attribute extraction during descend
-    spec_resolver = swagger_spec.resolver
-    resolve = swagger_spec.resolver.resolve
+    @cached_property
+    def resolve(self):
+        return self.swagger_spec.resolver.resolve
 
-    default_type_to_object = swagger_spec.config['default_type_to_object'] if swagger_spec else True
+    @cached_property
+    def default_type_to_object(self):
+        return self.swagger_spec.config['default_type_to_object']
 
-    def descend(value):
+    def descend(self, value):
         if is_ref(value):
             # Update spec_resolver scope to be able to dereference relative specs from a not root file
-            with in_scope(spec_resolver, value):
-                uri, deref_value = resolve(value['$ref'])
+            with in_scope(self.spec_resolver, value):
+                uri, deref_value = self.resolve(value['$ref'])
                 object_type = determine_object_type(
                     object_dict=deref_value,
-                    default_type_to_object=default_type_to_object,
+                    default_type_to_object=self.default_type_to_object,
                 )
 
                 known_mapping_key = object_type.get_root_holder()
                 if known_mapping_key is None:
-                    return descend(value=deref_value)
+                    return self.descend(value=deref_value)
                 else:
                     uri = urlparse(uri)
-                    if uri not in known_mappings.get(known_mapping_key, {}):
+                    if uri not in self.known_mappings.get(known_mapping_key, {}):
                         # The placeholder is present to interrupt the recursion
                         # during the recursive traverse of the data model (``descend``)
-                        known_mappings[known_mapping_key][uri] = None
+                        self.known_mappings[known_mapping_key][uri] = None
 
-                        known_mappings[known_mapping_key][uri] = descend(value=deref_value)
+                        self.known_mappings[known_mapping_key][uri] = self.descend(value=deref_value)
 
-                    return {'$ref': '#/{}/{}'.format(known_mapping_key, marshal_uri(uri))}
+                    return {'$ref': '#/{}/{}'.format(known_mapping_key, self.marshal_uri(uri))}
 
         elif is_dict_like(value):
             return {
-                key: descend(value=subval)
+                key: self.descend(value=subval)
                 for key, subval in iteritems(value)
             }
 
         elif is_list_like(value):
             return [
-                descend(value=subval)
+                self.descend(value=subval)
                 for index, subval in enumerate(value)
             ]
 
         else:
             return value
 
-    # Create internal copy of spec_dict to avoid external dict pollution
-    resolved_spec = descend(value=copy.deepcopy(swagger_spec.spec_dict))
+    def remove_duplicate_models(self):
+        # Run model-discovery in order to tag the models available in self.known_mappings['definitions']
+        # This is a required step that removes duplications of models due to the presence of models
+        # in swagger.json#/definitions and the equivalent models generated by flattening
+        if not self.swagger_spec.definitions:
+            return
 
-    # If definitions were identified by the swagger_spec object
-    if swagger_spec.definitions is not None:
         # local imports due to circular dependency
         from bravado_core.spec import Spec
-        from bravado_core.model import model_discovery
 
         # Run model-discovery in order to tag the models available in known_mappings['definitions']
         # This is a required step that removes duplications of models due to the presence of models
         # in swagger.json#/definitions and the equivalent models generated by flattening
         model_discovery(Spec(
             spec_dict={
                 'definitions': {
-                    marshal_uri(uri): value
-                    for uri, value in iteritems(known_mappings['definitions'])
+                    self.marshal_uri(uri): value
+                    for uri, value in iteritems(self.known_mappings['definitions'])
                 },
             },
         ))
 
         flatten_models = {
             # schema objects might not have a "type" set so they won't be tagged as models
             definition.get(MODEL_MARKER)
-            for definition in itervalues(known_mappings['definitions'])
+            for definition in itervalues(self.known_mappings['definitions'])
         }
 
-        for model_name, model_type in iteritems(swagger_spec.definitions):
+        for model_name, model_type in iteritems(self.swagger_spec.definitions):
             if model_name in flatten_models:
                 continue
             model_url = urlparse(model_type._json_reference)
-            known_mappings['definitions'][model_url] = descend(value=model_type._model_spec)
+            self.known_mappings['definitions'][model_url] = self.descend(
+                value=model_type._model_spec,
+            )
+
+    def warn_if_uri_clash_on_same_marshaled_representation(self, uri_schema_mappings):
+        """
+        Verifies that all the uris present on the definitions are represented by a different marshaled uri.
+        If is not the case a warning will filed.
+
+        In case of presence of warning please keep us informed about the issue, in the meantime you can
+        workaround this calling directly ``flattened_spec(spec, marshal_uri_function)`` passing your
+        marshalling function.
+        """
+        # Check that URIs are NOT clashing to same marshaled representation
+        marshaled_uri_mapping = defaultdict(set)
+        for uri in iterkeys(uri_schema_mappings):
+            marshaled_uri_mapping[self.marshal_uri(uri)].add(uri)
+
+        if len(marshaled_uri_mapping) != len(uri_schema_mappings):
+            # At least two uris clashed to the same marshaled representation
+            for marshaled_uri, uris in iteritems(marshaled_uri_mapping):
+                if len(uris) > 1:
+                    warnings.warn(
+                        message='{s_uris} clashed to {marshaled}'.format(
+                            s_uris=', '.join(sorted(urlunparse(uri) for uri in uris)),
+                            marshaled=marshaled_uri,
+                        ),
+                        category=Warning,
+                    )
+
+    @cached_property
+    def resolved_specs(self):
+        # Create internal copy of spec_dict to avoid external dict pollution
+        resolved_spec = self.descend(value=copy.deepcopy(self.swagger_spec.spec_dict))
+
+        # Remove models' duplication due to the presence of models in
+        # swagger.json#/definitions and equivalent models generated by flattening
+        self.remove_duplicate_models()
+
+        for mapping_key, mappings in iteritems(self.known_mappings):
+            self.warn_if_uri_clash_on_same_marshaled_representation(mappings)
+            if len(mappings) > 0:
+                resolved_spec.update(
+                    {
+                        mapping_key: {
+                            self.marshal_uri(uri): value
+                            for uri, value in iteritems(mappings)
+                        },
+                    },
+                )
 
-    for mapping_key, mappings in iteritems(known_mappings):
-        _warn_if_uri_clash_on_same_marshaled_representation(
-            uri_schema_mappings=mappings,
-            marshal_uri=marshal_uri,
-        )
-        if len(mappings) > 0:
-            resolved_spec.update({mapping_key: {
-                marshal_uri(uri): value
-                for uri, value in iteritems(mappings)
-            }})
+        return resolved_spec
+
+
+def flattened_spec(swagger_spec, marshal_uri_function=_marshal_uri):
+    """
+    Flatten Swagger Specs description into an unique and JSON serializable document.
+    The flattening injects in place the referenced [path item objects](https://swagger.io/specification/#pathItemObject)
+    while it injects in '#/parameters' the [parameter objects](https://swagger.io/specification/#parameterObject),
+    in '#/definitions' the [schema objects](https://swagger.io/specification/#schemaObject) and in
+    '#/responses' the [response objects](https://swagger.io/specification/#responseObject).
 
-    return resolved_spec
+    Note: the object names in '#/definitions', '#/parameters' and '#/responses' are evaluated by
+    ``marshal_uri_function``, the default method takes care of creating unique names for all the used references.
+    Since name clashing are still possible take care that a warning could be filed.
+    If it happen please report to us the specific warning text and the specs that generated it.
+    We can work to improve it and in the mean time you can "plug" a custom marshalling function.
+
+    Note: https://swagger.io/specification/ has been update to track the latest version of the Swagger/OpenAPI specs.
+    Please refer to https://github.com/OAI/OpenAPI-Specification/blob/3.0.0/versions/2.0.md#responseObject for the
+    most recent Swagger 2.0 specifications.
+
+    WARNING: In the future releases all the parameters except swagger_spec and marshal_uri_function will be removed.
+    Please make sure to use only those two parameters.
+    Until the deprecation is not effective you can still pass all the parameters but it's strongly discouraged.
+
+    :param swagger_spec: bravado-core Spec object
+    :type swagger_spec: bravado_core.spec.Spec
+    :param marshal_uri_function: function used to marshal uris in string suitable to be keys in Swagger Specs.
+    :type marshal_uri_function: Callable with the same signature of ``_marshal_uri``
+
+    :return: Flattened representation of the Swagger Specs
+    :rtype: dict
+    """
+    return _SpecFlattener(swagger_spec, marshal_uri_function).resolved_specs

tests/conftest.py

@@ -47,8 +47,8 @@ def minimal_swagger_dict(minimal_swagger_abspath):
 
 
 @pytest.fixture
-def minimal_swagger_spec(minimal_swagger_dict):
-    return Spec.from_dict(minimal_swagger_dict)
+def minimal_swagger_spec(minimal_swagger_dict, minimal_swagger_abspath):
+    return Spec.from_dict(minimal_swagger_dict, origin_url=get_url(minimal_swagger_abspath))
 
 
 @pytest.fixture

tests/model/model_discovery_test.py

@@ -1,24 +1,33 @@
 # -*- coding: utf-8 -*-
 import mock
+import pytest
 
+from bravado_core.model import _run_post_processing
 from bravado_core.model import model_discovery
 from bravado_core.spec import Spec
 
 
-@mock.patch('bravado_core.model._run_post_processing', autospec=True)
-def test_model_discovery_flow_no_ref_dereference(mock__run_post_processing, minimal_swagger_dict):
+@pytest.fixture
+def wrap__run_post_processing():
+    with mock.patch(
+        'bravado_core.model._run_post_processing',
+        wraps=_run_post_processing,
+    ) as _wrap__run_post_processing:
+        yield _wrap__run_post_processing
+
+
+def test_model_discovery_flow_no_ref_dereference(wrap__run_post_processing, minimal_swagger_dict):
     spec = Spec(
         spec_dict=minimal_swagger_dict,
         config={
             'internally_dereference_refs': False,
         },
     )
     model_discovery(swagger_spec=spec)
-    mock__run_post_processing.assert_called_once_with(spec)
+    wrap__run_post_processing.assert_called_once_with(spec)
 
 
-@mock.patch('bravado_core.model._run_post_processing', autospec=True)
-def test_model_discovery_flow_with_ref_dereference(mock__run_post_processing, minimal_swagger_dict):
+def test_model_discovery_flow_with_ref_dereference_with_no_definitions(wrap__run_post_processing, minimal_swagger_dict):
     spec = Spec(
         spec_dict=minimal_swagger_dict,
         config={
@@ -28,8 +37,24 @@ def test_model_discovery_flow_with_ref_dereference(mock__run_post_processing, mi
     )
     model_discovery(swagger_spec=spec)
 
+    # _run_post_processing is called 3 times
+    # 1. post processing on initial specs
+    # 2. post processing on on bravado_core.spec_flattening.flattened_spec
+    assert wrap__run_post_processing.call_count == 2
+
+
+def test_model_discovery_flow_with_ref_dereference_with_definitions(wrap__run_post_processing, minimal_swagger_dict):
+    spec = Spec(
+        spec_dict=dict(minimal_swagger_dict, definitions={'model': {'type': 'object'}}),
+        config={
+            'internally_dereference_refs': True,
+        },
+        origin_url='',
+    )
+    model_discovery(swagger_spec=spec)
+
     # _run_post_processing is called 3 times
     # 1. post processing on initial specs
     # 2. post processing on on bravado_core.spec_flattening.flattened_spec
     # 3. post processing to rebuild definitions to remove possible references in the model specs
-    assert mock__run_post_processing.call_count == 3
+    assert wrap__run_post_processing.call_count == 3