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

utils: introduce helper classes to manage software and instance schemas

parent 9d5ae147
......@@ -74,6 +74,7 @@ setup(name=name,
'six',
'cachecontrol',
'lockfile',
'jsonschema',
'uritemplate', # used by hateoas navigator
'subprocess32; python_version<"3"',
'ipaddress; python_version<"3"', # used by whitelistfirewall
......
......@@ -24,14 +24,26 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import functools
import json
import logging
import os
import slapos.util
from slapos.util import string_to_boolean, unicode2str
import shutil
import tempfile
import textwrap
import unittest
import shutil
from pwd import getpwnam
from six.moves import SimpleHTTPServer
import jsonschema
import slapos.util
from slapos.slap.slap import DEFAULT_SOFTWARE_TYPE
from slapos.testing.utils import ManagedHTTPServer
from slapos.util import (SoftwareReleaseSchema, SoftwareReleaseSerialisation,
string_to_boolean, unicode2str)
class TestUtil(unittest.TestCase):
"""
Tests methods available in the slapos.util module.
......@@ -232,5 +244,221 @@ class TestUtil(unittest.TestCase):
self.assertRaises(Exception, slapos.util.dumps, Nasty())
class SoftwareReleaseSchemaTestXmlSerialisationMixin:
serialisation = SoftwareReleaseSerialisation.Xml
class SoftwareReleaseSchemaTestJsonInXmlSerialisationMixin:
serialisation = SoftwareReleaseSerialisation.JsonInXml
class SoftwareReleaseSchemaTestMixin(object):
"""Mixin with test methods
"""
software_url = None # type: str
serialisation = None # type: SoftwareReleaseSerialisation
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 == SoftwareReleaseSerialisation.JsonInXml:
# 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 == SoftwareReleaseSerialisation.JsonInXml:
# 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 test_instance_request_parameter_schema_alternate_software_type(self):
schema = SoftwareReleaseSchema(self.software_url, 'alternate')
self.assertTrue(schema.getInstanceRequestParameterSchemaURL())
instance_parameter_schema = schema.getInstanceRequestParameterSchema()
self.assertEqual(instance_parameter_schema['description'],
"Simple instance parameters schema for tests")
class SoftwareReleaseSchemaTestFileSoftwareReleaseMixin(SoftwareReleaseSchemaTestMixin):
"""Mixin with tests and software release profiles and schema in a
temporary directory.
"""
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 SoftwareReleaseSchemaTestHTTPSoftwareReleaseMixin(SoftwareReleaseSchemaTestFileSoftwareReleaseMixin):
"""Mixin serving software release files over http.
"""
def setUp(self):
super(SoftwareReleaseSchemaTestHTTPSoftwareReleaseMixin, self).setUp()
class ProfileHTTPServer(ManagedHTTPServer):
hostname = os.environ['SLAPOS_TEST_IPV4']
working_directory = self.tmpdir
RequestHandler = SimpleHTTPServer.SimpleHTTPRequestHandler
self.logger = logging.getLogger(self.id())
self.logger.propagate = False
server = ProfileHTTPServer(self, 'server')
server.open()
self.addCleanup(server.close)
self.software_url = server.url + '/software.cfg'
class TestSoftwareReleaseSchemaFileSoftwareReleaseXmlSerialisation(
SoftwareReleaseSchemaTestXmlSerialisationMixin,
SoftwareReleaseSchemaTestFileSoftwareReleaseMixin,
unittest.TestCase):
pass
class TestSoftwareReleaseSchemaFileSoftwareReleaseJsonInXmlSerialisation(
SoftwareReleaseSchemaTestJsonInXmlSerialisationMixin,
SoftwareReleaseSchemaTestFileSoftwareReleaseMixin,
unittest.TestCase):
pass
class TestSoftwareReleaseSchemaHTTPSoftwareReleaseXmlSerialisation(
SoftwareReleaseSchemaTestXmlSerialisationMixin,
SoftwareReleaseSchemaTestHTTPSoftwareReleaseMixin,
unittest.TestCase):
pass
class TestSoftwareReleaseSchemaHTTPSoftwareReleaseJsonInXmlSerialisation(
SoftwareReleaseSchemaTestJsonInXmlSerialisationMixin,
SoftwareReleaseSchemaTestHTTPSoftwareReleaseMixin,
unittest.TestCase):
pass
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__':
unittest.main()
......@@ -27,19 +27,32 @@
#
##############################################################################
import enum
import errno
import hashlib
import json
import os
import shutil
import socket
import sqlite3
import struct
import subprocess
import sqlite3
from xml_marshaller.xml_marshaller import Marshaller, Unmarshaller
from lxml import etree
import warnings
import jsonschema
import netaddr
import requests
import six
from lxml import etree
from six.moves.urllib import parse
import hashlib
import netaddr
import shutil
from six.moves.urllib_parse import urljoin
from xml_marshaller.xml_marshaller import Marshaller, Unmarshaller
try:
from typing import Dict, Optional, IO
except ImportError:
pass
try:
......@@ -295,3 +308,148 @@ def rmtree(path):
raise e # XXX make pylint happy
shutil.rmtree(path, onerror=chmod_retry)
def _readAsJson(url):
# type: (str) -> Optional[Dict]
"""Reads and parse the json file located at `url`.
`url` can also be the path of a local file.
"""
if url.startswith('file://'):
url = url[len('file://'):]
path = url if os.path.exists(url) else None
if path:
with open(path) as f:
try:
return json.load(f)
except ValueError:
return None
if url.startswith('http://') or url.startswith('https://'):
try:
r = requests.get(url)
r.raise_for_status()
return r.json()
except (requests.exceptions.RequestException, ValueError):
return None
return None
class SoftwareReleaseSerialisation(str, enum.Enum):
Xml = 'xml'
JsonInXml = 'json-in-xml'
class SoftwareReleaseSchema(object):
def __init__(self, software_url, software_type):
# type: (str, Optional[str]) -> None
self.software_url = software_url
self.software_type = software_type
def getSoftwareSchema(self):
# type: () -> Optional[Dict]
"""Returns the schema for this software.
"""
return _readAsJson(self.software_url + '.json')
def getSoftwareTypeSchema(self):
# type: () -> Optional[Dict]
"""Returns schema for this software type.
"""
software_schema = self.getSoftwareSchema()
if software_schema is None:
return None
software_type = self.software_type
from slapos.slap.slap import DEFAULT_SOFTWARE_TYPE # TODO
if software_type is None:
software_type = DEFAULT_SOFTWARE_TYPE
# XXX some software are using "default" for default software type
if software_type == DEFAULT_SOFTWARE_TYPE \
and software_type not in software_schema['software-type'] \
and 'default' in software_schema['software-type']:
warnings.warn(
"Software release {} does not have schema for DEFAULT_SOFTWARE_TYPE but has one for 'default'."
" Using 'default' instead.".format(self.software_url),
UserWarning,
)
software_type = 'default'
return software_schema['software-type'].get(software_type)
def getSerialisation(self):
# type: () -> Optional[SoftwareReleaseSerialisation]
"""Returns the serialisation method used for parameters.
"""
software_schema = self.getSoftwareSchema()
if software_schema is None:
return None
return SoftwareReleaseSerialisation(software_schema['serialisation'])
def getInstanceRequestParameterSchemaURL(self):
# type: () -> Optional[str]
"""Returns the URL of the schema defining instance parameters.
"""
software_type_schema = self.getSoftwareTypeSchema()
if software_type_schema is None:
return None
software_url = self.software_url
if os.path.exists(software_url):
software_url = 'file://' + software_url
return urljoin(software_url, software_type_schema['request'])
def getInstanceRequestParameterSchema(self):
# type: () -> Optional[Dict]
"""Returns the schema defining instance parameters.
"""
instance_parameter_schema_url = self.getInstanceRequestParameterSchemaURL()
if instance_parameter_schema_url is None:
return None
schema = _readAsJson(instance_parameter_schema_url)
if schema:
# so that jsonschema knows how to resolve references
schema.setdefault('$id', instance_parameter_schema_url)
return schema
def getInstanceConnectionParameterSchemaURL(self):
# type: () -> Optional[str]
"""Returns the URL of the schema defining connection parameters published by the instance.
"""
software_type_schema = self.getSoftwareTypeSchema()
if software_type_schema is None:
return None
return urljoin(self.software_url, software_type_schema['response'])
def getInstanceConnectionParameterSchema(self):
# type: () -> Optional[Dict]
"""Returns the schema defining connection parameters published by the instance.
"""
instance_parameter_schema_url = self.getInstanceConnectionParameterSchemaURL()
if instance_parameter_schema_url is None:
return None
schema = _readAsJson(instance_parameter_schema_url)
if schema:
# so that jsonschema knows how to resolve references
schema.setdefault('$id', instance_parameter_schema_url)
return schema
def validateInstanceParameterDict(self, parameter_dict):
# type: (Dict) -> None
"""Validate instance parameters against the software schema.
Raise jsonschema.ValidationError if parameters does not validate.
"""
schema_url = self.getInstanceRequestParameterSchemaURL()
if schema_url:
instance = parameter_dict
if self.getSerialisation() == SoftwareReleaseSerialisation.JsonInXml:
try:
instance = json.loads(parameter_dict['_'])
except KeyError:
instance = parameter_dict
instance.pop('$schema', None)
jsonschema.validate(
instance=instance,
schema=self.getInstanceRequestParameterSchema(),
)
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