Merge pull request #6 from mission-liao/op

Op

pyswagger/base.py

@@ -217,6 +217,10 @@ def update_field(self, f, obj):
     def resolve(self, ts):
         """ resolve a list of tokens to an child object
         """
+        # TODO: test case
+        if isinstance(ts, six.string_types):
+            ts = [ts]
+
         obj = self
         while len(ts) > 0:
             t = ts.pop(0)
@@ -230,6 +234,21 @@ def resolve(self, ts):
 
         return obj
 
+    def merge(self, other):
+        """ merge properties from other object,
+        only merge from 'not None' to 'None'.
+        """
+        for name, _ in self.__swagger_fields__:
+            v = getattr(other, name)
+            if v != None and getattr(self, name) == None:
+                if isinstance(v, weakref.ProxyTypes):
+                    # TODO: test case
+                    self.update_field(name, v)
+                elif isinstance(v, BaseObj):
+                    self.update_field(name, weakref.proxy(v))
+                else:
+                    self.update_field(name, v)
+
     @property
     def _parent_(self):
         """ get parent object

pyswagger/core.py

@@ -244,6 +244,7 @@ def resolve(self, path):
         :rtype: weakref.ProxyType
         :raises ValueError: if path is not valid
         """
+        # TODO: test case
         if path == None or len(path) == 0:
             raise ValueError('Empty Path is not allowed')
 

pyswagger/scanner/v2_0/resolve.py

@@ -6,31 +6,77 @@
     Response,
     PathItem,
     )
-from ...utils import jp_split, jp_compose
+from ...utils import jp_compose
+
+
+# TODO: test case
+# TODO: cyclic detection
+
+def is_resolved(obj):
+    return getattr(obj, '$ref') == None or obj.ref_obj != None
+
+def _resolve(obj, app, prefix):
+    if is_resolved(obj):
+        return
+
+    r = getattr(obj, '$ref')
+
+    try:
+        ro = app.resolve(r)
+    except Exception:
+        ro = app.resolve(jp_compose(r, base=prefix)) 
+
+    if not ro:
+        raise ReferenceError('Unable to resolve: {0}'.format(r))
+    if ro.__class__ != obj.__class__:
+        raise TypeError('Referenced Type mismatch: {0}'.format(r))
+
+    obj.update_field('ref_obj', ro)
+
+def _merge(obj, app, prefix):
+    """ resolve $ref as ref_obj, and merge ref_obj to self.
+    This operation should be carried in a cascade manner.
+    """
+
+    cur = obj
+    to_resolve = []
+    while not is_resolved(cur):
+        _resolve(cur, app, prefix)
+
+        to_resolve.append(cur)
+        cur = cur.ref_obj if cur.ref_obj else cur
+
+    while (len(to_resolve)):
+        o = to_resolve.pop()
+        o.merge(o.ref_obj)
 
 
 class Resolve(object):
     """ pre-resolve '$ref' """
 
     class Disp(Dispatcher): pass
 
-    @Disp.register([Schema, Parameter, Response, PathItem])
-    def _resolve(self, path, obj, app):
-        r = getattr(obj, '$ref')
-        if r == None:
-            return
-
-        try:
-            ro = app.resolve(r)
-        except Exception:
-            ps = jp_split(path)[:2]
-            ps.append(r)
-            ro = app.resolve(jp_compose(ps))
-
-        if not ro:
-            raise ReferenceError('Unable to resolve: {0}'.format(r))
-        if ro.__class__ != obj.__class__:
-            raise TypeError('Referenced Type mismatch: {0}'.format(r))
-
-        obj.update_field('ref_obj', ro)
+
+    @Disp.register([Schema])
+    def _schema(self, _, obj, app):
+        _resolve(obj, app, '#/definitions')
+
+    @Disp.register([Parameter])
+    def _parameter(self, _, obj, app):
+        _resolve(obj, app, '#/parameters')
+
+    @Disp.register([Response])
+    def _response(self, _, obj, app):
+        _resolve(obj, app, '#/responses')
+
+    @Disp.register([PathItem])
+    def _path_item(self, _, obj, app):
+
+        # $ref in PathItem is 'merge', not 'replace'
+        # we need to merge properties of others if missing
+        # in current object.
+
+        # TODO: test case
+        _merge(obj, app, '#/paths')
+
 

pyswagger/tests/data/v2_0/resolve/path_item/swagger.json

@@ -0,0 +1,50 @@
+{
+   "swagger":"2.0",
+   "host":"http://test.com",
+   "basePath":"/v1",
+   "paths":{
+      "/a":{
+         "get":{
+            "operationId":"a.get",
+            "responses":{
+               "default":{
+                  "description":"void"
+               }
+            }
+         },
+         "$ref":"#/paths/~1b"
+      },
+      "/b":{
+         "get":{
+            "operationId":"b.get",
+            "responses":{
+               "default":{
+                  "description":"void"
+               }
+            }
+         },
+         "$ref":"#/paths/~1c"
+      },
+      "/c":{
+         "put":{
+            "operationId":"c.put",
+            "responses":{
+               "default":{
+                  "description":"void"
+               }
+            }
+         },
+         "$ref":"#/paths/~1d"
+      },
+      "/d":{
+         "post":{
+            "operationId":"d.post",
+            "responses":{
+               "default":{
+                  "description":"void"
+               }
+            }
+         }
+      }
+   }
+}

pyswagger/tests/test_base.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 from pyswagger import base
 import unittest
+import weakref
 import six
 
 
@@ -16,6 +17,7 @@ class TestObj(six.with_metaclass(base.FieldMeta, base.BaseObj)):
         ('b', {}),
         ('c', {}),
         ('d', None),
+        ('f', None)
     ]
 
 class TestContext(base.Context):
@@ -83,3 +85,27 @@ def test_field_default_value(self):
         o2 = TestObj(base.NullContext())
         self.assertTrue(id(o1.a) != id(o2.a))
 
+    def test_merge(self):
+        """ test merge function """
+        tmp = {'t': {}}
+        obj1 = {'a': [{}, {}, {}], 'd': {}, 'f': ''}
+        obj2 = {'a': [{}]}
+
+        with TestContext(tmp, 't') as ctx:
+            ctx.parse(obj1)
+        o1 = tmp['t']
+
+        with TestContext(tmp, 't') as ctx:
+            ctx.parse(obj2)
+        o2 = tmp['t']
+
+        self.assertTrue(len(o2.a), 1)
+        self.assertEqual(o2.d, None)
+        self.assertEqual(o2.f, None)
+
+        o2.merge(o1)
+        self.assertTrue(len(o2.a), 1)
+        self.assertEqual(o2.f, '')
+        self.assertTrue(isinstance(o2.d, ChildObj))
+        self.assertTrue(isinstance(o2.d, weakref.ProxyTypes))
+

pyswagger/tests/v2_0/test_op_access.py

@@ -0,0 +1,33 @@
+from pyswagger import SwaggerApp, utils
+from ..utils import get_test_data_folder
+import unittest
+
+def _check(u, op):
+    u.assertEqual(op.operationId, 'addPet')
+
+class OperationAccessTestCase(unittest.TestCase):
+    """ test for methods to access Operation """
+
+    @classmethod
+    def setUpClass(kls):
+        kls.app = SwaggerApp._create_(get_test_data_folder(version='2.0', which='wordnik'))
+
+    def test_resolve(self):
+        """
+        """
+        _check(self, self.app.resolve(utils.jp_compose(['#', 'paths', '/pet', 'post'])))
+
+    def test_cascade_resolve(self):
+        """
+        """
+        path = self.app.resolve(utils.jp_compose(['#', 'paths', '/pet']))
+
+        _check(self, path.resolve('post'))
+        _check(self, path.post)
+
+    def test_tag_operationId(self):
+        """
+        """
+        _check(self, self.app.op['pet', 'addPet'])
+        _check(self, self.app.op['addPet'])
+

pyswagger/tests/v2_0/test_resolve.py

@@ -0,0 +1,25 @@
+from pyswagger import SwaggerApp, utils                                        
+from pyswagger.spec.v2_0 import objects                                        
+from ..utils import get_test_data_folder                                       
+import unittest                                                                
+import os                                                                      
+
+
+class ResolvePathItemTestCase(unittest.TestCase):                              
+    """ test for PathItem $ref """                                                 
+
+    @classmethod                                                                   
+    def setUpClass(kls):                                                           
+        kls.app = SwaggerApp._create_(get_test_data_folder(                            
+            version='2.0',                                                                 
+            which=os.path.join('resolve', 'path_item')                                     
+        ))                                                                             
+
+    def test_path_item(self):                                                      
+        """ make sure PathItem is correctly merged """
+        a = self.app.resolve(utils.jp_compose('/a', '#/paths'))                        
+
+        self.assertTrue(isinstance(a, objects.PathItem))     
+        self.assertTrue(a.get.operationId, 'a.get')
+        self.assertTrue(a.put.operationId, 'c.put')
+        self.assertTrue(a.post.operationId, 'd.post')