Skip to content
Snippets Groups Projects
Commit 43d7efb3 authored by Jorrit Schaap's avatar Jorrit Schaap
Browse files

TMSS-1546: fixed tests, added test_amibuous_remote_refs_raises test

parent 1d644766
No related branches found
No related tags found
1 merge request!774TMSS-1545 and TMSS-1546
...@@ -22,7 +22,10 @@ import jsonschema ...@@ -22,7 +22,10 @@ import jsonschema
from copy import deepcopy from copy import deepcopy
import requests import requests
from datetime import datetime, timedelta from datetime import datetime, timedelta
from lofar.common.util import dict_with_overrides from .util import single_line_with_single_spaces
class JSONError(Exception):
pass
DEFAULT_MAX_SCHEMA_CACHE_AGE = timedelta(minutes=1) DEFAULT_MAX_SCHEMA_CACHE_AGE = timedelta(minutes=1)
...@@ -229,7 +232,7 @@ def _fetch_url(url: str) -> str: ...@@ -229,7 +232,7 @@ def _fetch_url(url: str) -> str:
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
time.sleep(2) # retry after a little sleep time.sleep(2) # retry after a little sleep
raise Exception("Could not get: %s" % (url,)) raise JSONError("Could not get: %s" % (url,))
def _get_referenced_definition(ref_url, cache: dict=None, max_cache_age: timedelta=DEFAULT_MAX_SCHEMA_CACHE_AGE): def _get_referenced_definition(ref_url, cache: dict=None, max_cache_age: timedelta=DEFAULT_MAX_SCHEMA_CACHE_AGE):
...@@ -320,7 +323,8 @@ def resolved_remote_refs(schema, cache: dict=None, max_cache_age: timedelta=DEFA ...@@ -320,7 +323,8 @@ def resolved_remote_refs(schema, cache: dict=None, max_cache_age: timedelta=DEFA
resolved_definition = get_sub_schema(referenced_schema, local_ref, None) resolved_definition = get_sub_schema(referenced_schema, local_ref, None)
if current_definition is not None and current_definition != resolved_definition: if current_definition is not None and current_definition != resolved_definition:
raise Exception("ambiguity while resolving remote references in schema $id='%s' $ref='%s'" % (schema.get('$id', '<no_id>'), local_ref)) msg = "ambiguity while resolving remote references in schema $id='%s' $ref='%s' definition1='%s' definition2='%s'" % (schema.get('$id', '<no_id>'), local_ref, single_line_with_single_spaces(current_definition), single_line_with_single_spaces(resolved_definition))
raise JSONError(msg)
write_at_path(copy_of_schema, local_ref, resolved_definition) write_at_path(copy_of_schema, local_ref, resolved_definition)
...@@ -444,7 +448,7 @@ def raise_on_self_refs(schema: dict): ...@@ -444,7 +448,7 @@ def raise_on_self_refs(schema: dict):
id = id.rstrip('/').rstrip('#').rstrip('/') id = id.rstrip('/').rstrip('#').rstrip('/')
for ref in get_refs(schema): for ref in get_refs(schema):
if ref.startswith(id): if ref.startswith(id):
raise Exception("schema $id='%s' contains a $ref to itself: '%s'" %(id, ref)) raise JSONError("schema $id='%s' contains a $ref to itself: '%s'" %(id, ref))
def validate_json_object_with_schema(json_object, schema): def validate_json_object_with_schema(json_object, schema):
......
...@@ -241,6 +241,157 @@ class TestJSONUtils(unittest.TestCase): ...@@ -241,6 +241,157 @@ class TestJSONUtils(unittest.TestCase):
expected_resolved_user_schema = r'''{
"$id": "%s/user_schema.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"type": "object",
"default": {},
"definitions": {
"user_schema_local_prop": {
"type": "boolean"
},
"timestamp": {
"description": "A timestamp defined in UTC",
"type": "string",
"pattern": "\\d{4}-[01]\\d-[0-3]\\dT[0-2]\\d:[0-5]\\d:[0-5]\\d(\\.\\d+)?Z?",
"format": "date-time"
},
"email": {
"type": "string",
"format": "email",
"pattern": "@example\\.com$"
},
"account": {
"type": "object",
"properties": {
"email_address": {
"$ref": "#/definitions/email"
},
"creation_at": {
"$ref": "#/definitions/timestamp"
}
}
}
},
"properties": {
"name": {
"type": "string",
"minLength": 2
},
"user_account": {
"$ref": "#/definitions/account",
"extra_prop": "very important"
},
"other_emails": {
"type": "array",
"items": {
"$ref": "#/definitions/email"
}
}
}
}
''' % (base_url,)
self.assertEqual(json.loads(expected_resolved_user_schema), resolved_user_schema)
finally:
httpd.shutdown()
thread.join(timeout=2)
self.assertFalse(thread.is_alive())
def test_resolved_remote_refs(self):
'''test if $refs to URL's are properly resolved'''
import http.server
import socketserver
from lofar.common.util import find_free_port
port = find_free_port(8000, allow_reuse_of_lingering_port=False)
host = "127.0.0.1"
host_port = "%s:%s" % (host, port)
base_url = "http://" + host_port
base_schema = {"$id": base_url + "/base_schema.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"definitions": {
"timestamp": {
"description": "A timestamp defined in UTC",
"type": "string",
"pattern": "\\d{4}-[01]\\d-[0-3]\\dT[0-2]\\d:[0-5]\\d:[0-5]\\d(\\.\\d+)?Z?",
"format": "date-time"
},
"email": {
"type": "string",
"format": "email",
"pattern": "@example\\.com$"},
"account": {
"type": "object",
"properties": {
"email_address": {
"$ref": "#/definitions/email"
},
"creation_at": {
"$ref": "#/definitions/timestamp"
}
}
}
}}
user_schema = {"$id": base_url + "/user_schema.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"type": "object",
"default": {},
"definitions": {
"user_schema_local_prop": {
"type": "boolean"
}
},
"properties": {
"name": {
"type": "string",
"minLength": 2},
"user_account": {
"$ref": base_url + "/base_schema.json" + "#/definitions/account",
"extra_prop": "very important"
},
"other_emails": {
"type": "array",
"items": {
"$ref": base_url + "/base_schema.json" + "#/definitions/email"
}
}}}
class TestRequestHandler(http.server.BaseHTTPRequestHandler):
'''helper class to serve the schemas via http. Needed for resolving the $ref URLs'''
def send_json_response(self, json_object):
self.send_response(http.HTTPStatus.OK)
self.send_header("Content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(json_object, indent=2).encode('utf-8'))
def do_GET(self):
try:
if self.path == "/base_schema.json":
self.send_json_response(base_schema)
elif self.path == "/user_schema.json":
self.send_json_response(user_schema)
else:
self.send_error(http.HTTPStatus.NOT_FOUND, "No such resource")
except Exception as e:
self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
with socketserver.TCPServer((host, port), TestRequestHandler) as httpd:
thread = threading.Thread(target=httpd.serve_forever)
thread.start()
try:
# the method-under-test
resolved_user_schema = resolved_remote_refs(user_schema)
print('base_schema: ', json.dumps(base_schema, indent=2))
print('user_schema: ', json.dumps(user_schema, indent=2))
print('resolved_user_schema: ', json.dumps(resolved_user_schema, indent=2))
expected_resolved_user_schema = r'''{ expected_resolved_user_schema = r'''{
"$id": "http://127.0.0.1:8000/user_schema.json", "$id": "http://127.0.0.1:8000/user_schema.json",
"$schema": "http://json-schema.org/draft-06/schema#", "$schema": "http://json-schema.org/draft-06/schema#",
...@@ -298,6 +449,88 @@ class TestJSONUtils(unittest.TestCase): ...@@ -298,6 +449,88 @@ class TestJSONUtils(unittest.TestCase):
thread.join(timeout=2) thread.join(timeout=2)
self.assertFalse(thread.is_alive()) self.assertFalse(thread.is_alive())
def test_amibuous_remote_refs_raises(self):
'''test if amibugous $refs raise'''
import http.server
import socketserver
from lofar.common.util import find_free_port
port = find_free_port(8000, allow_reuse_of_lingering_port=False)
host = "127.0.0.1"
host_port = "%s:%s" % (host, port)
base_url = "http://"+host_port
# create 2 schemas with the a different definition for "my_prop"
base_schema1 = { "$id": base_url + "/base_schema1.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"definitions": {
"my_prop": {
"type": "string"
}
}
}
base_schema2 = { "$id": base_url + "/base_schema2.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"definitions": {
"my_prop": {
"type": "int"
}
}
}
# use both in one schema...
user_schema = {"$id": base_url + "/user_schema.json",
"$schema": "http://json-schema.org/draft-06/schema#",
"type": "object",
"properties": {
"abc": {
"$ref": base_url + "/base_schema1.json" + "#/definitions/my_prop"
},
"xyz": {
"$ref": base_url + "/base_schema2.json" + "#/definitions/my_prop"
}
}
}
# run an http server to host them...
class TestRequestHandler(http.server.BaseHTTPRequestHandler):
'''helper class to serve the schemas via http. Needed for resolving the $ref URLs'''
def send_json_response(self, json_object):
self.send_response(http.HTTPStatus.OK)
self.send_header("Content-type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(json_object, indent=2).encode('utf-8'))
def do_GET(self):
try:
if self.path == "/base_schema1.json":
self.send_json_response(base_schema1)
elif self.path == "/base_schema2.json":
self.send_json_response(base_schema2)
elif self.path == "/user_schema.json":
self.send_json_response(user_schema)
else:
self.send_error(http.HTTPStatus.NOT_FOUND, "No such resource")
except Exception as e:
self.send_error(http.HTTPStatus.INTERNAL_SERVER_ERROR, str(e))
with socketserver.TCPServer((host, port), TestRequestHandler) as httpd:
thread = threading.Thread(target=httpd.serve_forever)
thread.start()
try:
with self.assertRaises(Exception) as context:
resolved_remote_refs(user_schema)
self.assertTrue('ambiguity while resolving remote references' in str(context.exception))
finally:
httpd.shutdown()
thread.join(timeout=2)
self.assertFalse(thread.is_alive())
def test_replace_host_in_ref_urls(self): def test_replace_host_in_ref_urls(self):
base_host = "http://foo.bar.com" base_host = "http://foo.bar.com"
path = "/my/path" path = "/my/path"
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment