Commit d6883553 authored by Jérome Perrin's avatar Jérome Perrin

wip all

parent 63a443e3
...@@ -24,14 +24,23 @@ ...@@ -24,14 +24,23 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
# #
############################################################################## ##############################################################################
import functools
import json
import os import os
import slapos.util import shutil
from slapos.util import string_to_boolean, unicode2str
import tempfile import tempfile
import textwrap
import typing
import unittest import unittest
import shutil
from pwd import getpwnam from pwd import getpwnam
import jsonschema
import slapos.util
from slapos.util import SoftwareReleaseSchema, string_to_boolean, unicode2str
from slapos.slap.slap import DEFAULT_SOFTWARE_TYPE
class TestUtil(unittest.TestCase): class TestUtil(unittest.TestCase):
""" """
Tests methods available in the slapos.util module. Tests methods available in the slapos.util module.
...@@ -232,5 +241,165 @@ class TestUtil(unittest.TestCase): ...@@ -232,5 +241,165 @@ class TestUtil(unittest.TestCase):
self.assertRaises(Exception, slapos.util.dumps, Nasty()) self.assertRaises(Exception, slapos.util.dumps, Nasty())
TestMixin = object
if typing.TYPE_CHECKING:
TestMixin = unittest.TestCase
class SoftwareReleaseSchemaTestCase(TestMixin):
software_url = None # type: str
serialisation = "json-in-xml"
def test_software_schema(self):
schema = SoftwareReleaseSchema(self.software_url, None)
software_schema = schema.getSoftwareSchema()
self.assertEqual(software_schema['name'], 'Test Software')
self.assertEqual(len(software_schema['software-type']), 2)
def test_serialisation(self):
schema = SoftwareReleaseSchema(self.software_url, None)
self.assertEqual(schema.getSerialisation(), self.serialisation)
def test_instance_request_parameter_schema_default_software_type(self):
schema = SoftwareReleaseSchema(self.software_url, None)
self.assertTrue(schema.getInstanceRequestParameterSchemaURL())
instance_parameter_schema = schema.getInstanceRequestParameterSchema()
self.assertEqual(instance_parameter_schema['description'],
"Simple instance parameters schema for tests")
def test_connection_parameter_schema(self):
schema = SoftwareReleaseSchema(self.software_url, None)
self.assertTrue(schema.getInstanceConnectionParameterSchemaURL())
instance_parameter_schema = schema.getInstanceConnectionParameterSchema()
self.assertEqual(instance_parameter_schema['description'],
"Simple connection parameters schema for tests")
def test_instance_request_parameter_validate_default_software_type(self):
schema = SoftwareReleaseSchema(self.software_url, None)
self.assertTrue(schema.getInstanceRequestParameterSchemaURL())
instance_ok = {'key': 'value', 'type': 'default'}
schema.validateInstanceParameterDict(instance_ok)
if self.serialisation == "json-in-xml":
# already serialized values are also tolerated
schema.validateInstanceParameterDict({'_': json.dumps(instance_ok)})
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict({"wrong": True})
instance_ok['key'] = False # wrong type
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict(instance_ok)
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict({'_': json.dumps(instance_ok)})
def test_instance_request_parameter_validate_alternate_software_type(self):
schema = SoftwareReleaseSchema(self.software_url, 'alternate')
self.assertTrue(schema.getInstanceRequestParameterSchemaURL())
instance_ok = {'key': 'value', 'type': 'alternate'}
schema.validateInstanceParameterDict(instance_ok)
if self.serialisation == "json-in-xml":
# already serialized values are also tolerated
schema.validateInstanceParameterDict({'_': json.dumps(instance_ok)})
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict({"wrong": True})
instance_ok['type'] = 'wrong'
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict(instance_ok)
with self.assertRaises(jsonschema.ValidationError):
schema.validateInstanceParameterDict({'_': json.dumps(instance_ok)})
def XXtest_instance_request_parameter_schema_alternate(self):
schema = SoftwareReleaseSchema(self.software_url, 'alternate')
self.assertTrue(schema.getInstanceRequestParameterSchemaURL())
instance_parameter_schema = schema.getInstanceRequestParameterSchema()
self.assertEqual(instance_parameter_schema['description'], "Simple instance schema for tests")
class TestSoftwareReleaseSchemaFile(SoftwareReleaseSchemaTestCase, unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tmpdir)
tmpfile = functools.partial(os.path.join, self.tmpdir)
with open(tmpfile('software.cfg'), 'w') as f:
f.write(
textwrap.dedent("""\
[buildout]
"""))
with open(tmpfile('software.cfg.json'), 'w') as f:
json.dump(
{
"name": "Test Software",
"description": "Dummy software for Test",
"serialisation": self.serialisation,
"software-type": {
DEFAULT_SOFTWARE_TYPE: {
"title": "Default",
"description": "Default type",
"request": "instance-default-input-schema.json",
"response": "instance-default-output-schema.json",
"index": 0
},
"alternate": {
"title": "Alternate",
"description": "Alternate type",
"request": "instance-alternate-input-schema.json",
"response": "instance-alternate-output-schema.json",
"index": 0
},
}
}, f)
for software_type in ('default', 'alternate'):
with open(
tmpfile('instance-{software_type}-input-schema.json'.format(
software_type=software_type)), 'w') as f:
json.dump(
{
"$schema": "http://json-schema.org/draft-07/schema",
"description": "Simple instance parameters schema for tests",
"required": ["key", "type"],
"properties": {
"key": {
"$ref": "./schemas-definitions.json#/key"
},
"type": {
"type": "string",
"const": software_type
}
},
"type": "object"
}, f)
with open(
tmpfile('instance-{software_type}-output-schema.json'.format(
software_type=software_type)), 'w') as f:
json.dump(
{
"$schema": "http://json-schema.org/draft-07/schema",
"description": "Simple connection parameters schema for tests",
}, f)
with open(tmpfile('schemas-definitions.json'), 'w') as f:
json.dump({"key": {"type": "string"}}, f)
self.software_url = tmpfile('software.cfg')
class TestSoftwareReleaseSchemaEdgeCases(unittest.TestCase):
def test_software_schema_file_not_exist(self):
schema = SoftwareReleaseSchema('/file/not/exist', None)
self.assertIsNone(schema.getSoftwareSchema())
def test_software_schema_wrong_URL(self):
schema = SoftwareReleaseSchema('http://slapos.invalid/software.cfg', None)
self.assertIsNone(schema.getSoftwareSchema())
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()
...@@ -332,7 +332,7 @@ def _readAsJson(url): ...@@ -332,7 +332,7 @@ def _readAsJson(url):
r = requests.get(url) r = requests.get(url)
r.raise_for_status() r.raise_for_status()
return r.json() return r.json()
except (requests.exceptions.HTTPError, ValueError): except (requests.exceptions.RequestException, ValueError):
return None return None
return None return None
...@@ -442,7 +442,12 @@ class SoftwareReleaseSchema(object): ...@@ -442,7 +442,12 @@ class SoftwareReleaseSchema(object):
""" """
schema_url = self.getInstanceRequestParameterSchemaURL() schema_url = self.getInstanceRequestParameterSchemaURL()
if schema_url: if schema_url:
instance = parameter_dict if self.getSerialisation() == SoftwareReleaseSerialisation.Xml else json.loads(parameter_dict['_']) instance = parameter_dict
if self.getSerialisation() == SoftwareReleaseSerialisation.JsonInXml:
try:
instance = json.loads(parameter_dict['_'])
except KeyError:
instance = parameter_dict
instance.pop('$schema', None) instance.pop('$schema', None)
jsonschema.validate( jsonschema.validate(
instance=instance, instance=instance,
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment