summaryrefslogtreecommitdiff
path: root/src/leap/base
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/base')
-rw-r--r--src/leap/base/config.py96
-rw-r--r--src/leap/base/constants.py2
-rw-r--r--src/leap/base/pluggableconfig.py421
-rw-r--r--src/leap/base/specs.py104
-rw-r--r--src/leap/base/tests/test_config.py4
-rw-r--r--src/leap/base/tests/test_providers.py48
-rw-r--r--src/leap/base/tests/test_validation.py92
7 files changed, 641 insertions, 126 deletions
diff --git a/src/leap/base/config.py b/src/leap/base/config.py
index 76fbee3c..dc047f80 100644
--- a/src/leap/base/config.py
+++ b/src/leap/base/config.py
@@ -9,13 +9,12 @@ import tempfile
import os
logger = logging.getLogger(name=__name__)
-logger.setLevel('DEBUG')
-import configuration
import requests
from leap.base import exceptions
from leap.base import constants
+from leap.base.pluggableconfig import PluggableConfig
from leap.util.fileutil import (mkdir_p)
# move to base!
@@ -38,13 +37,9 @@ class BaseLeapConfig(object):
def get_config(self, *kwargs):
raise NotImplementedError("abstract base class")
- #XXX todo: enable this property after
- #fixing name clash with "config" in use at
- #vpnconnection
-
- #@property
- #def config(self):
- #return self.get_config()
+ @property
+ def config(self):
+ return self.get_config()
def get_value(self, *kwargs):
raise NotImplementedError("abstract base class")
@@ -54,55 +49,51 @@ class MetaConfigWithSpec(type):
"""
metaclass for JSONLeapConfig classes.
It creates a configuration spec out of
- the `spec` dictionary.
+ the `spec` dictionary. The `properties` attribute
+ of the spec dict is turn into the `schema` attribute
+ of the new class (which will be used to validate against).
"""
# XXX in the near future, this is the
# place where we want to enforce
- # singletons, read-only and stuff.
-
- # TODO:
- # - add a error handler for missing options that
- # we can act easily upon (sys.exit is ugly, for $deity's sake)
+ # singletons, read-only and similar stuff.
def __new__(meta, classname, bases, classDict):
- spec_options = classDict.get('spec', None)
+ schema_obj = classDict.get('spec', None)
+
# not quite happy with this workaround.
# I want to raise if missing spec dict, but only
# for grand-children of this metaclass.
# maybe should use abc module for this.
abcderived = ("JSONLeapConfig",)
- if spec_options is None and classname not in abcderived:
+ if schema_obj is None and classname not in abcderived:
raise exceptions.ImproperlyConfigured(
- "missing spec dict on your derived class")
+ "missing spec dict on your derived class (%s)" % classname)
- # we create a configuration spec attribute from the spec dict
+ # we create a configuration spec attribute
+ # from the spec dict
config_class = type(
classname + "Spec",
- (configuration.Configuration, object),
- {'options': spec_options})
+ (PluggableConfig, object),
+ {'options': schema_obj})
classDict['spec'] = config_class
return type.__new__(meta, classname, bases, classDict)
##########################################################
-# hacking in progress:
+# some hacking still in progress:
# Configs have:
+
# - a slug (from where a filename/folder is derived)
# - a spec (for validation and defaults).
-# this spec is basically a dict that will be used
+# this spec is conformant to the json-schema.
+# basically a dict that will be used
# for type casting and validation, and defaults settings.
# all config objects, since they are derived from BaseConfig, implement basic
# useful methods:
# - save
# - load
-# - get_config (returns a optparse.OptionParser object)
-
-# TODO:
-# - have a good type cast repertory (uris, version, hashes...)
-# - raise validation errors
-# - multilingual objects
##########################################################
@@ -125,10 +116,10 @@ class JSONLeapConfig(BaseLeapConfig):
raise exceptions.ImproperlyConfigured(
"missing spec on JSONLeapConfig"
" derived class")
- assert issubclass(self.spec, configuration.Configuration)
+ assert issubclass(self.spec, PluggableConfig)
- self._config = self.spec()
- self._config.parse_args(list(args))
+ self._config = self.spec(format="json")
+ self._config.load()
self.fetcher = kwargs.pop('fetcher', requests)
# mandatory baseconfig interface
@@ -139,13 +130,6 @@ class JSONLeapConfig(BaseLeapConfig):
folder, filename = os.path.split(to)
if folder and not os.path.isdir(folder):
mkdir_p(folder)
- # lazy evaluation until first level of nesting
- # to allow lambdas with context-dependant info
- # like os.path.expanduser
- config = self.get_config()
- for k, v in config.iteritems():
- if callable(v):
- config[k] = v()
self._config.serialize(to)
def load(self, fromfile=None, from_uri=None, fetcher=None, verify=False):
@@ -155,28 +139,36 @@ class JSONLeapConfig(BaseLeapConfig):
return
if fromfile is None:
fromfile = self.filename
- newconfig = self._config.deserialize(fromfile)
- # XXX check for no errors, etc
- self._config.config = newconfig
+ if os.path.isfile(fromfile):
+ self._config.load(fromfile=fromfile)
+ else:
+ logger.error('tried to load config from non-existent path')
+ logger.error('Not Found: %s', fromfile)
def fetch(self, uri, fetcher=None, verify=True):
if not fetcher:
fetcher = self.fetcher
logger.debug('verify: %s', verify)
request = fetcher.get(uri, verify=verify)
+ # XXX should send a if-modified-since header
# XXX get 404, ...
# and raise a UnableToFetch...
request.raise_for_status()
fd, fname = tempfile.mkstemp(suffix=".json")
- if not request.json:
+
+ if request.json:
+ self._config.load(json.dumps(request.json))
+
+ else:
+ # not request.json
+ # might be server did not announce content properly,
+ # let's try deserializing all the same.
try:
- json.loads(request.content)
+ self._config.load(request.content)
except ValueError:
raise eipexceptions.LeapBadConfigFetchedError
- with open(fname, 'w') as tmp:
- tmp.write(json.dumps(request.json))
- self._loadtemp(fname)
+
return True
def get_config(self):
@@ -191,16 +183,16 @@ class JSONLeapConfig(BaseLeapConfig):
def filename(self):
return self.get_filename()
- # private
+ def validate(self, data):
+ logger.debug('validating schema')
+ self._config.validate(data)
+ return True
- def _loadtemp(self, filename):
- self.load(fromfile=filename)
- os.remove(filename)
+ # private
def _slug_to_filename(self):
# is this going to work in winland if slug is "foo/bar" ?
folder, filename = os.path.split(self.slug)
- # XXX fix import
config_file = get_config_file(filename, folder)
return config_file
diff --git a/src/leap/base/constants.py b/src/leap/base/constants.py
index 7a1415fb..48a18dc3 100644
--- a/src/leap/base/constants.py
+++ b/src/leap/base/constants.py
@@ -16,7 +16,7 @@ DEFINITION_EXPECTED_PATH = "provider.json"
DEFAULT_PROVIDER_DEFINITION = {
u'api_uri': u'https://api.%s/' % DEFAULT_PROVIDER,
u'api_version': u'0.1.0',
- u'ca_cert': u'8aab80ae4326fd30721689db813733783fe0bd7e',
+ u'ca_cert_fingerprint': u'8aab80ae4326fd30721689db813733783fe0bd7e',
u'ca_cert_uri': u'https://%s/cacert.pem' % DEFAULT_PROVIDER,
u'description': {u'en': u'This is a test provider'},
u'display_name': {u'en': u'Test Provider'},
diff --git a/src/leap/base/pluggableconfig.py b/src/leap/base/pluggableconfig.py
new file mode 100644
index 00000000..b8615ad8
--- /dev/null
+++ b/src/leap/base/pluggableconfig.py
@@ -0,0 +1,421 @@
+"""
+generic configuration handlers
+"""
+import copy
+import json
+import logging
+import os
+import time
+import urlparse
+
+import jsonschema
+
+logger = logging.getLogger(__name__)
+
+
+__all__ = ['PluggableConfig',
+ 'adaptors',
+ 'types',
+ 'UnknownOptionException',
+ 'MissingValueException',
+ 'ConfigurationProviderException',
+ 'TypeCastException']
+
+# exceptions
+
+
+class UnknownOptionException(Exception):
+ """exception raised when a non-configuration
+ value is present in the configuration"""
+
+
+class MissingValueException(Exception):
+ """exception raised when a required value is missing"""
+
+
+class ConfigurationProviderException(Exception):
+ """exception raised when a configuration provider is missing, etc"""
+
+
+class TypeCastException(Exception):
+ """exception raised when a
+ configuration item cannot be coerced to a type"""
+
+
+class ConfigAdaptor(object):
+ """
+ abstract base class for config adaotors for
+ serialization/deserialization and custom validation
+ and type casting.
+ """
+ def read(self, filename):
+ raise NotImplementedError("abstract base class")
+
+ def write(self, config, filename):
+ with open(filename, 'w') as f:
+ self._write(f, config)
+
+ def _write(self, fp, config):
+ raise NotImplementedError("abstract base class")
+
+ def validate(self, config, schema):
+ raise NotImplementedError("abstract base class")
+
+
+adaptors = {}
+
+
+class JSONSchemaEncoder(json.JSONEncoder):
+ """
+ custom default encoder that
+ casts python objects to json objects for
+ the schema validation
+ """
+ def default(self, obj):
+ if obj is str:
+ return 'string'
+ if obj is unicode:
+ return 'string'
+ if obj is int:
+ return 'integer'
+ if obj is list:
+ return 'array'
+ if obj is dict:
+ return 'object'
+ if obj is bool:
+ return 'boolean'
+
+
+class JSONAdaptor(ConfigAdaptor):
+ indent = 2
+ extensions = ['json']
+
+ def read(self, _from):
+ if isinstance(_from, file):
+ _from_string = _from.read()
+ if isinstance(_from, str):
+ _from_string = _from
+ return json.loads(_from_string)
+
+ def _write(self, fp, config):
+ fp.write(json.dumps(config,
+ indent=self.indent,
+ sort_keys=True))
+
+ def validate(self, config, schema_obj):
+ schema_json = JSONSchemaEncoder().encode(schema_obj)
+ schema = json.loads(schema_json)
+ jsonschema.validate(config, schema)
+
+
+adaptors['json'] = JSONAdaptor()
+
+#
+# Adaptors
+#
+# Allow to apply a predefined set of types to the
+# specs, so it checks the validity of formats and cast it
+# to proper python types.
+
+# TODO:
+# - multilingual object.
+# - HTTPS uri
+
+
+class DateType(object):
+ fmt = '%Y-%m-%d'
+
+ def to_python(self, data):
+ return time.strptime(data, self.fmt)
+
+ def get_prep_value(self, data):
+ return time.strftime(self.fmt, data)
+
+
+class URIType(object):
+
+ def to_python(self, data):
+ parsed = urlparse.urlparse(data)
+ if not parsed.scheme:
+ raise TypeCastException("uri %s has no schema" % data)
+ return parsed
+
+ def get_prep_value(self, data):
+ return data.geturl()
+
+
+class HTTPSURIType(object):
+
+ def to_python(self, data):
+ parsed = urlparse.urlparse(data)
+ if not parsed.scheme:
+ raise TypeCastException("uri %s has no schema" % data)
+ if parsed.scheme != "https":
+ raise TypeCastException(
+ "uri %s does not has "
+ "https schema" % data)
+ return parsed
+
+ def get_prep_value(self, data):
+ return data.geturl()
+
+
+types = {
+ 'date': DateType(),
+ 'uri': URIType(),
+ 'https-uri': HTTPSURIType(),
+}
+
+
+class PluggableConfig(object):
+
+ options = {}
+
+ def __init__(self,
+ adaptors=adaptors,
+ types=types,
+ format=None):
+
+ self.config = {}
+ self.adaptors = adaptors
+ self.types = types
+ self._format = format
+
+ @property
+ def option_dict(self):
+ if hasattr(self, 'options') and isinstance(self.options, dict):
+ return self.options.get('properties', None)
+
+ def items(self):
+ """
+ act like an iterator
+ """
+ if isinstance(self.option_dict, dict):
+ return self.option_dict.items()
+ return self.options
+
+ def validate(self, config, format=None):
+ """
+ validate config
+ """
+ schema = self.options
+ if format is None:
+ format = self._format
+
+ if format:
+ adaptor = self.get_adaptor(self._format)
+ adaptor.validate(config, schema)
+ else:
+ # we really should make format mandatory...
+ logger.error('no format passed to validate')
+
+ # first round of validation is ok.
+ # now we proceed to cast types if any specified.
+ self.to_python(config)
+
+ def to_python(self, config):
+ """
+ cast types following first type and then format indications.
+ """
+ unseen_options = [i for i in config if i not in self.option_dict]
+ if unseen_options:
+ raise UnknownOptionException(
+ "Unknown options: %s" % ', '.join(unseen_options))
+
+ for key, value in config.items():
+ _type = self.option_dict[key].get('type')
+ if _type is None and 'default' in self.option_dict[key]:
+ _type = type(self.option_dict[key]['default'])
+ if _type is not None:
+ tocast = True
+ if not callable(_type) and isinstance(value, _type):
+ tocast = False
+ if tocast:
+ try:
+ config[key] = _type(value)
+ except BaseException, e:
+ raise TypeCastException(
+ "Could not coerce %s, %s, "
+ "to type %s: %s" % (key, value, _type.__name__, e))
+ _format = self.option_dict[key].get('format', None)
+ _ftype = self.types.get(_format, None)
+ if _ftype:
+ try:
+ config[key] = _ftype.to_python(value)
+ except BaseException, e:
+ raise TypeCastException(
+ "Could not coerce %s, %s, "
+ "to format %s: %s" % (key, value,
+ _ftype.__class__.__name__,
+ e))
+
+ return config
+
+ def prep_value(self, config):
+ """
+ the inverse of to_python method,
+ called just before serialization
+ """
+ for key, value in config.items():
+ _format = self.option_dict[key].get('format', None)
+ _ftype = self.types.get(_format, None)
+ if _ftype and hasattr(_ftype, 'get_prep_value'):
+ try:
+ config[key] = _ftype.get_prep_value(value)
+ except BaseException, e:
+ raise TypeCastException(
+ "Could not serialize %s, %s, "
+ "by format %s: %s" % (key, value,
+ _ftype.__class__.__name__,
+ e))
+ else:
+ config[key] = value
+ return config
+
+ # methods for adding configuration
+
+ def get_default_values(self):
+ """
+ return a config options from configuration defaults
+ """
+ defaults = {}
+ for key, value in self.items():
+ if 'default' in value:
+ defaults[key] = value['default']
+ return copy.deepcopy(defaults)
+
+ def get_adaptor(self, format):
+ """
+ get specified format adaptor or
+ guess for a given filename
+ """
+ adaptor = self.adaptors.get(format, None)
+ if adaptor:
+ return adaptor
+
+ # not registered in adaptors dict, let's try all
+ for adaptor in self.adaptors.values():
+ if format in adaptor.extensions:
+ return adaptor
+
+ def filename2format(self, filename):
+ extension = os.path.splitext(filename)[-1]
+ return extension.lstrip('.') or None
+
+ def serialize(self, filename, format=None, full=False):
+ if not format:
+ format = self._format
+ if not format:
+ format = self.filename2format(filename)
+ if not format:
+ raise Exception('Please specify a format')
+ # TODO: more specific exception type
+
+ adaptor = self.get_adaptor(format)
+ if not adaptor:
+ raise Exception("Adaptor not found for format: %s" % format)
+
+ config = copy.deepcopy(self.config)
+ serializable = self.prep_value(config)
+ adaptor.write(serializable, filename)
+
+ def deserialize(self, string=None, fromfile=None, format=None):
+ """
+ load configuration from a file or string
+ """
+
+ def _try_deserialize():
+ if fromfile:
+ with open(fromfile, 'r') as f:
+ content = adaptor.read(f)
+ elif string:
+ content = adaptor.read(string)
+ return content
+
+ # XXX cleanup this!
+
+ if fromfile:
+ assert os.path.exists(fromfile)
+ if not format:
+ format = self.filename2format(fromfile)
+
+ if not format:
+ format = self._format
+ if format:
+ adaptor = self.get_adaptor(format)
+ else:
+ adaptor = None
+
+ if adaptor:
+ content = _try_deserialize()
+ return content
+
+ # no adaptor, let's try rest of adaptors
+
+ adaptors = self.adaptors[:]
+
+ if format:
+ adaptors.sort(
+ key=lambda x: int(
+ format in x.extensions),
+ reverse=True)
+
+ for adaptor in adaptors:
+ content = _try_deserialize()
+ return content
+
+ def load(self, *args, **kwargs):
+ """
+ load from string or file
+ if no string of fromfile option is given,
+ it will attempt to load from defaults
+ defined in the schema.
+ """
+ string = args[0] if args else None
+ fromfile = kwargs.get("fromfile", None)
+ content = None
+
+ # start with defaults, so we can
+ # have partial values applied.
+ content = self.get_default_values()
+ if string and isinstance(string, str):
+ content = self.deserialize(string)
+
+ if not string and fromfile is not None:
+ #import ipdb;ipdb.set_trace()
+ content = self.deserialize(fromfile=fromfile)
+
+ if not content:
+ logger.error('no content could be loaded')
+ # XXX raise!
+ return
+
+ # lazy evaluation until first level of nesting
+ # to allow lambdas with context-dependant info
+ # like os.path.expanduser
+ for k, v in content.iteritems():
+ if callable(v):
+ content[k] = v()
+
+ self.validate(content)
+ self.config = content
+ return True
+
+
+def testmain():
+ from tests import test_validation as t
+ import pprint
+
+ config = PluggableConfig(_format="json")
+ properties = copy.deepcopy(t.sample_spec)
+
+ config.options = properties
+ config.load(fromfile='data.json')
+
+ print 'config'
+ pprint.pprint(config.config)
+
+ config.serialize('/tmp/testserial.json')
+
+if __name__ == "__main__":
+ testmain()
diff --git a/src/leap/base/specs.py b/src/leap/base/specs.py
index d88dc63f..b4bb8dcf 100644
--- a/src/leap/base/specs.py
+++ b/src/leap/base/specs.py
@@ -1,49 +1,59 @@
leap_provider_spec = {
- 'serial': {
- 'type': int,
- 'default': 1,
- 'required': True,
- },
- 'version': {
- 'type': unicode,
- 'default': '0.1.0'
- #'required': True
- },
- 'domain': {
- 'type': unicode, # XXX define uri type
- 'default': 'testprovider.example.org'
- #'required': True,
- },
- 'display_name': {
- 'type': unicode, # XXX multilingual object?
- 'default': 'test provider'
- #'required': True
- },
- 'description': {
- 'default': 'test provider'
- },
- 'enrollment_policy': {
- 'type': unicode, # oneof ??
- 'default': 'open'
- },
- 'services': {
- 'type': list, # oneof ??
- 'default': ['eip']
- },
- 'api_version': {
- 'type': unicode,
- 'default': '0.1.0' # version regexp
- },
- 'api_uri': {
- 'type': unicode # uri
- },
- 'public_key': {
- 'type': unicode # fingerprint
- },
- 'ca_cert': {
- 'type': unicode
- },
- 'ca_cert_uri': {
- 'type': unicode
- },
+ 'description': 'provider definition',
+ 'type': 'object',
+ 'properties': {
+ 'serial': {
+ 'type': int,
+ 'default': 1,
+ 'required': True,
+ },
+ 'version': {
+ 'type': unicode,
+ 'default': '0.1.0'
+ #'required': True
+ },
+ 'domain': {
+ 'type': unicode, # XXX define uri type
+ 'default': 'testprovider.example.org'
+ #'required': True,
+ },
+ 'display_name': {
+ 'type': dict, # XXX multilingual object?
+ 'default': {u'en': u'Test Provider'}
+ #'required': True
+ },
+ 'description': {
+ 'type': dict,
+ 'default': {u'en': u'Test provider'}
+ },
+ 'enrollment_policy': {
+ 'type': unicode, # oneof ??
+ 'default': 'open'
+ },
+ 'services': {
+ 'type': list, # oneof ??
+ 'default': ['eip']
+ },
+ 'api_version': {
+ 'type': unicode,
+ 'default': '0.1.0' # version regexp
+ },
+ 'api_uri': {
+ 'type': unicode # uri
+ },
+ 'public_key': {
+ 'type': unicode # fingerprint
+ },
+ 'ca_cert_fingerprint': {
+ 'type': unicode,
+ },
+ 'ca_cert_uri': {
+ 'type': unicode,
+ 'format': 'https-uri'
+ },
+ 'languages': {
+ 'type': list,
+ 'default': ['en']
+ }
+ }
}
diff --git a/src/leap/base/tests/test_config.py b/src/leap/base/tests/test_config.py
index bede5ea1..d03149b2 100644
--- a/src/leap/base/tests/test_config.py
+++ b/src/leap/base/tests/test_config.py
@@ -38,14 +38,14 @@ class JSONLeapConfigTest(BaseLeapTest):
class DummyTestConfig(config.JSONLeapConfig):
__metaclass__ = config.MetaConfigWithSpec
- spec = {}
+ spec = {'properties': {}}
with self.assertRaises(exceptions.ImproperlyConfigured) as exc:
DummyTestConfig()
exc.startswith("missing slug")
class DummyTestConfig(config.JSONLeapConfig):
__metaclass__ = config.MetaConfigWithSpec
- spec = {}
+ spec = {'properties': {}}
slug = "foo"
DummyTestConfig()
diff --git a/src/leap/base/tests/test_providers.py b/src/leap/base/tests/test_providers.py
index 9e0ff90c..8d3b8847 100644
--- a/src/leap/base/tests/test_providers.py
+++ b/src/leap/base/tests/test_providers.py
@@ -1,35 +1,39 @@
+import copy
import json
try:
import unittest2 as unittest
except ImportError:
import unittest
-
import os
+import jsonschema
+
from leap import __branding as BRANDING
from leap.testing.basetest import BaseLeapTest
from leap.base import providers
EXPECTED_DEFAULT_CONFIG = {
- "api_version": "0.1.0",
- "description": "test provider",
- "display_name": "test provider",
- "domain": "testprovider.example.org",
- "enrollment_policy": "open",
- "serial": 1,
- "services": [
- "eip"
+ u"api_version": u"0.1.0",
+ u"description": {u'en': u"Test provider"},
+ u"display_name": {u'en': u"Test Provider"},
+ u"domain": u"testprovider.example.org",
+ u"enrollment_policy": u"open",
+ u"serial": 1,
+ u"services": [
+ u"eip"
],
- "version": "0.1.0"
+ u"languages": [u"en"],
+ u"version": u"0.1.0"
}
class TestLeapProviderDefinition(BaseLeapTest):
def setUp(self):
self.definition = providers.LeapProviderDefinition()
- #XXX change to self.definition.config when property is fixed
- self.config = self.definition.get_config()
+ self.definition.save()
+ self.definition.load()
+ self.config = self.definition.config
def tearDown(self):
if hasattr(self, 'testfile') and os.path.isfile(self.testfile):
@@ -57,6 +61,7 @@ class TestLeapProviderDefinition(BaseLeapTest):
self.testfile = self.get_tempfile('test.json')
self.definition.save(to=self.testfile)
deserialized = json.load(open(self.testfile, 'rb'))
+ self.maxDiff = None
self.assertEqual(deserialized, EXPECTED_DEFAULT_CONFIG)
def test_provider_dump_to_slug(self):
@@ -78,6 +83,13 @@ class TestLeapProviderDefinition(BaseLeapTest):
self.assertDictEqual(self.config,
EXPECTED_DEFAULT_CONFIG)
+ def test_provider_validation(self):
+ self.definition.validate(self.config)
+ _config = copy.deepcopy(self.config)
+ _config['serial'] = 'aaa'
+ with self.assertRaises(jsonschema.ValidationError):
+ self.definition.validate(_config)
+
@unittest.skip
def test_load_malformed_json_definition(self):
raise NotImplementedError
@@ -89,18 +101,6 @@ class TestLeapProviderDefinition(BaseLeapTest):
raise NotImplementedError
-class TestLeapProvider(BaseLeapTest):
- def setUp(self):
- pass
-
- def tearDown(self):
- pass
-
- ###
-
- # XXX ??
-
-
class TestLeapProviderSet(BaseLeapTest):
def setUp(self):
diff --git a/src/leap/base/tests/test_validation.py b/src/leap/base/tests/test_validation.py
new file mode 100644
index 00000000..87e99648
--- /dev/null
+++ b/src/leap/base/tests/test_validation.py
@@ -0,0 +1,92 @@
+import copy
+import datetime
+#import json
+try:
+ import unittest2 as unittest
+except ImportError:
+ import unittest
+import os
+
+import jsonschema
+
+from leap.base.config import JSONLeapConfig
+from leap.base import pluggableconfig
+from leap.testing.basetest import BaseLeapTest
+
+SAMPLE_CONFIG_DICT = {
+ 'prop_one': 1,
+ 'prop_uri': "http://example.org",
+ 'prop_date': '2012-12-12',
+}
+
+EXPECTED_CONFIG = {
+ 'prop_one': 1,
+ 'prop_uri': "http://example.org",
+ 'prop_date': datetime.datetime(2012, 12, 12)
+}
+
+sample_spec = {
+ 'description': 'sample schema definition',
+ 'type': 'object',
+ 'properties': {
+ 'prop_one': {
+ 'type': int,
+ 'default': 1,
+ 'required': True
+ },
+ 'prop_uri': {
+ 'type': str,
+ 'default': 'http://example.org',
+ 'required': True,
+ 'format': 'uri'
+ },
+ 'prop_date': {
+ 'type': str,
+ 'default': '2012-12-12',
+ 'format': 'date'
+ }
+ }
+}
+
+
+class SampleConfig(JSONLeapConfig):
+ spec = sample_spec
+
+ @property
+ def slug(self):
+ return os.path.expanduser('~/sampleconfig.json')
+
+
+class TestJSONLeapConfigValidation(BaseLeapTest):
+ def setUp(self):
+ self.sampleconfig = SampleConfig()
+ self.sampleconfig.save()
+ self.sampleconfig.load()
+ self.config = self.sampleconfig.config
+
+ def tearDown(self):
+ if hasattr(self, 'testfile') and os.path.isfile(self.testfile):
+ os.remove(self.testfile)
+
+ # tests
+
+ def test_good_validation(self):
+ self.sampleconfig.validate(SAMPLE_CONFIG_DICT)
+
+ def test_broken_int(self):
+ _config = copy.deepcopy(SAMPLE_CONFIG_DICT)
+ _config['prop_one'] = '1'
+ with self.assertRaises(jsonschema.ValidationError):
+ self.sampleconfig.validate(_config)
+
+ def test_format_property(self):
+ # JsonSchema Validator does not check the format property.
+ # We should have to extend the Configuration class
+ blah = copy.deepcopy(SAMPLE_CONFIG_DICT)
+ blah['prop_uri'] = 'xxx'
+ with self.assertRaises(pluggableconfig.TypeCastException):
+ self.sampleconfig.validate(blah)
+
+
+if __name__ == "__main__":
+ unittest.main()
generated by cgit v1.2.3 (git 2.25.1) at 2025-10-31 22:08:12 +0000