TMSS-235: refactored schema-url-modification routines. added tests

LCS/PyCommon/json_utils.py

@@ -88,44 +88,72 @@ def add_defaults_to_json_object_for_schema(json_object: dict, schema: str, ref_r
         _DefaultValidatingDraft6Validator(schema).validate(copy_of_json_object)
     return copy_of_json_object
 
+def replace_host_in_urls(schema, old_base_url: str, new_base_url: str, keys=['$id', '$ref', '$schema']):
+    '''return the given schema with all fields in the given keys which start with the given old_base_url updated so they point to the given new_base_url'''
+    if isinstance(schema, dict):
+        updated_schema = {}
+        for key, value in schema.items():
+            if key in keys:
+                if value.startswith(old_base_url):
+                    try:
+                        # deduct referred schema name and version from ref-value
+                        head, anchor, tail = value.partition('#')
+                        host, slash, path = head.lstrip('http://').lstrip('https://').partition('/')
 
-def transform_nodes(schema, transform_function):
-    '''return the given schema with all $ref fields updated so they point to the given base_url'''
-    if isinstance(schema, list):
-        return [transform_function(transform_nodes(item, transform_function)) for item in schema]
+                        # and construct the proper ref url
+                        updated_schema[key] = new_base_url.rstrip('/') + '/' + path + anchor + tail
+                    except:
+                        # apparently the reference is not conform the expected lofar common json schema path...
+                        # so, just accept the original value and assume that the user uploaded a proper schema
+                        updated_schema[key] = value
+                    continue
 
-    if isinstance(schema, dict):
-        # Loop over all key/value pairs and recursively transform each value node.
-        return {key: transform_function(transform_nodes(value, transform_function)) for key, value in schema.items()}
+            updated_schema[key] = replace_host_in_urls(value, new_base_url, old_base_url, keys)
+        return updated_schema
 
-    # schema is not a list, nor a dict, so it is a normal value node. Let's transform it and return it.
-    return transform_function(schema)
+    if isinstance(schema, list):
+        return [replace_host_in_urls(item, new_base_url, old_base_url, keys) for item in schema]
 
+    return schema
 
-def resolve_ref(ref_url):
+def get_referenced_subschema(ref_url):
     '''fetch the schema given by the ref_url, and get the sub-schema given by the #/ path in the ref_url'''
     # deduct referred schema name and version from ref-value
-    head, hash, tail = ref_url.partition('#')
+    head, anchor, tail = ref_url.partition('#')
     # TODO: maybe use cache for requested urls?
-    reffered_schema = json.loads(requests.get(ref_url).text)
+    referenced_schema = json.loads(requests.get(ref_url).text)
 
     # extract sub-schema
     parts = tail.strip('/').split('/')
     for part in parts:
-        reffered_schema = reffered_schema[part]
-    return reffered_schema
+        referenced_schema = referenced_schema[part]
 
-def replace_ref(node):
-    '''replace the '$ref':<URL> pair by the resolved (sub)schema, or just return this same node if no $ref:<URL> pair present.'''
-    if isinstance(node, dict):
-        if '$ref' in node and 'http' in node['$ref']:
-            return resolve_ref(node['$ref'])
-    return node
+    return referenced_schema
 
-def resolved_refs(schema):
-    '''return the given schema with all $ref:<URL> fields replaced by the resolved $refs'''
-    return transform_nodes(schema, transform_function=replace_ref)
 
+def resolved_refs(schema):
+    '''return the given schema with all $ref fields replaced by the referred json (sub)schema that they point to.'''
+    if isinstance(schema, dict):
+        updated_schema = {}
+        for key, value in schema.items():
+            if key == "$ref" and isinstance(value, str):
+                if value.startswith('#'):
+                    # reference to local document, no need for http injection
+                    updated_schema[key] = value
+                else:
+                    try:
+                        # by returning the referenced (sub)schema, the $ref-key and url-value are replaced from the caller's perspective.
+                        return get_referenced_subschema(value)
+                    except:
+                        # can't get the referenced schema
+                        # so, just accept the original value and assume that the user uploaded a proper schema
+                        updated_schema[key] = value
+            else:
+                updated_schema[key] = resolved_refs(value)
+        return updated_schema
 
+    if isinstance(schema, list):
+        return [resolved_refs(item) for item in schema]
 
+    return schema
 

LCS/PyCommon/test/t_json_utils.py

@@ -133,7 +133,32 @@ class TestJSONUtils(unittest.TestCase):
             thread.join(timeout=2)
             self.assertFalse(thread.is_alive())
 
+    def test_replace_host_in_ref_urls(self):
+        base_host = "http://foo.bar.com"
+        path = "/my/path"
 
+        schema = {"$id": base_host + path + "/user_schema.json",
+                  "$schema": "http://json-schema.org/draft-06/schema#",
+                  "type": "object",
+                  "default": {},
+                  "properties": {
+                      "name": {
+                          "type": "string",
+                          "minLength": 2 },
+                      "email": {
+                          "$ref": base_host + path + "/base_schema.json" + "/#definitions/email"  }
+                  } }
+
+        new_base_host = 'http://127.0.0.1'
+        url_fixed_schema = replace_host_in_urls(schema, 'http://foo.bar.com', new_base_host)
+
+        print('schema: ', json.dumps(schema, indent=2))
+        print('url_fixed_schema: ', json.dumps(url_fixed_schema, indent=2))
+
+        self.assertEqual(new_base_host+path+"/user_schema.json", url_fixed_schema['$id'])
+        self.assertEqual(new_base_host+path+"/base_schema.json" + "/#definitions/email", url_fixed_schema['properties']['email']['$ref'])
+        self.assertEqual("http://json-schema.org/draft-06/schema#", url_fixed_schema['$schema'])
+        self.assertEqual(json.dumps(schema, indent=2).replace(base_host, new_base_host), json.dumps(url_fixed_schema, indent=2))
 
 if __name__ == '__main__':
     unittest.main()