Merge pull request #131 from MCapitani/links

add support for "links" in Response

Author: p1c2u

openapi_core/schema/links/__init__.py

@@ -0,0 +1,44 @@
+"""OpenAPI core links generators module"""
+from six import iteritems
+
+from openapi_core.compat import lru_cache
+from openapi_core.schema.links.models import Link
+from openapi_core.schema.parameters.generators import ParametersGenerator
+from openapi_core.schema.servers.generators import ServersGenerator
+
+
+class LinksGenerator(object):
+
+    def __init__(self, dereferencer, schemas_registry):
+        self.dereferencer = dereferencer
+        self.schemas_registry = schemas_registry
+
+    def generate(self, links):
+        for link_name, link in iteritems(links):
+            link_deref = self.dereferencer.dereference(link)
+            operation_id = link_deref.get('operationId')
+            parameters = link_deref.get('parameters', {})
+            request_body = link_deref.get('requestBody')  # string or dict
+            description = link_deref.get('description')
+            server_spec = link_deref.get('server')
+            server = self.servers_generator.generate(server_spec) \
+                if server_spec is not None \
+                else None
+
+            yield link_name, Link(
+                operation_id,
+                parameters,
+                request_body,
+                description,
+                server
+            )
+
+    @property
+    @lru_cache()
+    def parameters_generator(self):
+        return ParametersGenerator(self.dereferencer, self.schemas_registry)
+
+    @property
+    @lru_cache()
+    def servers_generator(self):
+        return ServersGenerator(self.dereferencer)

openapi_core/schema/links/generators.py

@@ -0,0 +1,26 @@
+"""OpenAPI core links models module"""
+
+
+class Link(object):
+    """Represents an OpenAPI Link."""
+
+    def __init__(
+            self,
+            operation_id,
+            parameters,
+            request_body,
+            description,
+            server
+    ):
+        """
+        request_body is assumed to be either a string (JSON, YAML or
+        runtime expression) or an object (deserialized JSON or YAML)
+        """
+        self.operationId = operation_id
+        self.description = description
+        self.server = server
+        self.parameters = dict(parameters) if parameters else {}
+        self.request_body = request_body
+
+    def __getitem__(self, item):
+        return self.parameters[item]

openapi_core/schema/links/models.py

@@ -2,6 +2,7 @@
 from six import iteritems
 
 from openapi_core.compat import lru_cache
+from openapi_core.schema.links.generators import LinksGenerator
 from openapi_core.schema.media_types.generators import MediaTypeGenerator
 from openapi_core.schema.parameters.generators import ParametersGenerator
 from openapi_core.schema.responses.models import Response
@@ -19,6 +20,8 @@ def generate(self, responses):
             description = response_deref['description']
             headers = response_deref.get('headers')
             content = response_deref.get('content')
+            links_dict = response_deref.get('links', {})
+            links = self.links_generator.generate(links_dict)
 
             media_types = None
             if content:
@@ -30,7 +33,7 @@ def generate(self, responses):
 
             yield http_status, Response(
                 http_status, description,
-                content=media_types, headers=parameters)
+                content=media_types, headers=parameters, links=links)
 
     @property
     @lru_cache()
@@ -41,3 +44,8 @@ def media_types_generator(self):
     @lru_cache()
     def parameters_generator(self):
         return ParametersGenerator(self.dereferencer, self.schemas_registry)
+
+    @property
+    @lru_cache()
+    def links_generator(self):
+        return LinksGenerator(self.dereferencer, self.schemas_registry)

openapi_core/schema/responses/generators.py

@@ -0,0 +1,48 @@
+openapi: "3.0.0"
+info:
+  title: Minimal valid OpenAPI specification
+  version: "0.1"
+paths:
+  /linked/noParam:
+    get:
+      operationId: noParOp
+      responses:
+        default:
+          description: the linked result
+  /linked/withParam:
+    get:
+      operationId: paramOp
+      parameters:
+        - name: opParam
+          in: query
+          description: test
+          schema:
+            type: string
+      responses:
+        default:
+          description: the linked result
+  /status:
+    get:
+      responses:
+        default:
+          description: Return something
+          links:
+            noParamLink:
+              operationId: noParOp
+  /status/{resourceId}:
+    get:
+      parameters:
+        - name: resourceId
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        default:
+          description: Return something else
+          links:
+            paramLink:
+              operationId: paramOp
+              parameters:
+                opParam: $request.path.resourceId
+              requestBody: test

tests/integration/data/v3.0/links.yaml

@@ -0,0 +1,36 @@
+from openapi_core.shortcuts import create_spec
+
+
+class TestLinkSpec(object):
+
+    def test_no_param(self, factory):
+        spec_dict = factory.spec_from_file("data/v3.0/links.yaml")
+        spec = create_spec(spec_dict)
+        resp = spec['/status']['get'].get_response()
+
+        assert len(resp.links) == 1
+
+        link = resp.links['noParamLink']
+
+        assert link.operationId == 'noParOp'
+        assert link.server is None
+        assert link.request_body is None
+        assert len(link.parameters) == 0
+
+    def test_param(self, factory):
+        spec_dict = factory.spec_from_file("data/v3.0/links.yaml")
+        spec = create_spec(spec_dict)
+        resp = spec['/status/{resourceId}']['get'].get_response()
+
+        assert len(resp.links) == 1
+
+        link = resp.links['paramLink']
+
+        assert link.operationId == 'paramOp'
+        assert link.server is None
+        assert link.request_body == 'test'
+        assert len(link.parameters) == 1
+
+        param = link.parameters['opParam']
+
+        assert param == '$request.path.resourceId'

tests/integration/test_link_spec.py

@@ -0,0 +1,44 @@
+import mock
+import pytest
+
+from openapi_core.schema.links.models import Link
+from openapi_core.schema.servers.models import Server
+
+
+class TestLinks(object):
+
+    @pytest.fixture
+    def link_factory(self):
+        def link_factory(request_body, server):
+            parameters = {
+                'par1': mock.sentinel.par1,
+                'par2': mock.sentinel.par2,
+            }
+            return Link(
+                'op_id',
+                parameters,
+                request_body,
+                'Test link',
+                server
+            )
+        return link_factory
+
+    servers = [
+        None,
+        Server("https://bad.remote.domain.net/"),
+        Server("http://localhost")
+    ]
+
+    request_body_list = [
+        None,
+        "request",
+        '{"request": "value", "opt": 2}',
+        {"request": "value", "opt": 2}
+    ]
+
+    @pytest.mark.parametrize("server", servers)
+    @pytest.mark.parametrize("request_body", request_body_list)
+    def test_iteritems(self, link_factory, request_body, server):
+        link = link_factory(request_body, server)
+        for par_name in link.parameters.keys():
+            assert link[par_name] == link.parameters[par_name]