Add typed UI metadata (and functionalities)

c2cgeoportal/__init__.py

@@ -97,7 +97,7 @@ def _render(value, system):
 
 def add_interface(
     config, interface_name=None, interface_type=INTERFACE_TYPE_CGXP, **kwargs
-):  # pragma: nocover
+):  # pragma: no cover
     if interface_type == INTERFACE_TYPE_CGXP:
         if interface_name is None:
             add_interface_cgxp(
@@ -128,7 +128,7 @@ def add_interface(
         )
 
 
-def add_interface_cgxp(config, interface_name, route_names, routes, renderers):  # pragma: nocover
+def add_interface_cgxp(config, interface_name, route_names, routes, renderers):  # pragma: no cover
     # Cannot be at the header to don"t load the model too early
     from c2cgeoportal.views.entry import Entry
 
@@ -175,7 +175,7 @@ def new_f(root, request):
 ngeo_static_init = False
 
 
-def add_interface_ngeo(config, interface_name, route_name, route, renderer):  # pragma: nocover
+def add_interface_ngeo(config, interface_name, route_name, route, renderer):  # pragma: no cover
     # Cannot be at the header to don't load the model too early
     from c2cgeoportal.views.entry import Entry
 
@@ -213,10 +213,10 @@ def new_f(root, request):
         ngeo_static_init = True
 
 
-def add_static_view_ngeo(config):  # pragma: nocover
+def add_static_view_ngeo(config):  # pragma: no cover
     """ Add the project static view for ngeo """
     package = config.get_settings()["package"]
-    _add_static_view(config, "proj-ngeo", "%s:static-ngeo" % package)
+    _add_static_view(config, "static-ngeo", "%s:static-ngeo" % package)
     config.override_asset(
         to_override="c2cgeoportal:project/",
         override_with="%s:static-ngeo/" % package
@@ -246,7 +246,7 @@ def add_admin_interface(config):
 def add_static_view(config):
     """ Add the project static view for CGXP """
     package = config.get_settings()["package"]
-    _add_static_view(config, "proj", "%s:static" % package)
+    _add_static_view(config, "static-cgxp", "%s:static" % package)
     config.override_asset(
         to_override="c2cgeoportal:project/",
         override_with="%s:static/" % package
@@ -385,7 +385,7 @@ def __call__(self, context, request):
                 return False
         return True
 
-    def phash(self):  # pragma: nocover
+    def phash(self):  # pragma: no cover
         return ""
 
 
@@ -414,7 +414,7 @@ def add_cors_route(config, pattern, service):
     """
     Add the OPTIONS route and view need for services supporting CORS.
     """
-    def view(request):  # pragma: nocover
+    def view(request):  # pragma: no cover
         from c2cgeoportal.lib.caching import set_common_headers, NO_CACHE
         return set_common_headers(request, service, NO_CACHE)
 
@@ -423,7 +423,7 @@ def view(request):  # pragma: nocover
     config.add_view(view, route_name=name)
 
 
-def error_handler(http_exception, request):  # pragma: nocover
+def error_handler(http_exception, request):  # pragma: no cover
     """
     View callable for handling all the exceptions that are not already handled.
     """
@@ -510,7 +510,7 @@ def includeme(config):
     config.add_directive("set_user_validator", set_user_validator)
     config.set_user_validator(default_user_validator)
 
-    if settings.get("ogcproxy_enable", False):  # pragma: nocover
+    if settings.get("ogcproxy_enable", False):  # pragma: no cover
         # add an OGCProxy view
         config.add_route_predicate("ogc_server", OgcproxyRoutePredicate)
         config.add_route(
@@ -690,6 +690,10 @@ def includeme(config):
         settings,
         ("admin_interface", "available_metadata"),
         formalchemy_available_metadata)
+    formalchemy_available_metadata = [
+        e if isinstance(e, basestring) else e.get("name")
+        for e in formalchemy_available_metadata
+    ]
 
     config.add_route("checker_all", "/checker_all", request_method="GET")
 

c2cgeoportal/forms.py

@@ -468,7 +468,7 @@ def render_tree(self):
 
 
 # LayerWMTS
-class RoListRenderer(FieldRenderer):  # pragma: nocover
+class RoListRenderer(FieldRenderer):  # pragma: no cover
     def render_readonly(self, **kwargs):
         return helpers.content_tag("span", ("," + helpers.tag("br")).join([
             helpers.literal(value) for value in self.raw_value
@@ -491,7 +491,7 @@ class ChildrenAttributeField(AttributeField):
     def __init__(self, *args, **kargs):
         AttributeField.__init__(self, *args, **kargs)
 
-    def sync(self):  # pragma: nocover
+    def sync(self):  # pragma: no cover
         self.model.children = [
             self.parent.session.query(models.TreeItem).get(int(pk))
             for pk in self.renderer.deserialize()

c2cgeoportal/lib/__init__.py

@@ -29,6 +29,9 @@
 
 
 import re
+import urlparse
+import datetime
+import dateutil
 from urlparse import urlsplit, urlunsplit, urljoin
 from urllib import quote
 
@@ -38,6 +41,18 @@
 from pyramid.config.views import StaticURLInfo
 
 
+def get_types_map(types_array):
+    types_map = {}
+    for type_ in types_array:
+        if isinstance(type_, basestring):
+            types_map[type_] = {
+                "name": type_,
+            }
+        else:
+            types_map[type_["name"]] = type_
+    return types_map
+
+
 def get_url(url, request, default=None, errors=None):
     if url is None:
         return default
@@ -57,7 +72,7 @@ def get_url(url, request, default=None, errors=None):
         server = request.registry.settings.get("servers", {}).get(obj.netloc, None)
         if server is None:
             if default is None and errors is not None:
-                errors.append("The server '%s' isn't found in the config" % obj.netloc)
+                errors.add("The server '%s' isn't found in the config" % obj.netloc)
             return default
         else:
             return "%s%s?%s" % (server, obj.path, obj.query)
@@ -66,6 +81,124 @@ def get_url(url, request, default=None, errors=None):
         return url
 
 
+def get_url2(name, value, request, errors):
+    url = urlparse.urlsplit(value)
+    if url.scheme == "":
+        if url.netloc == "" and url.path not in ("", "/"):
+            # Relative URL like: /dummy/static/url or dummy/static/url
+            return urlparse.urlunsplit(url)
+        errors.add(
+            "The attribute '{}'='{}' isn't an URL."
+            .format(name, value)
+        )
+        return None
+    elif url.scheme in ("http", "https"):
+        if url.netloc == "":
+            errors.add(
+                "The attribute '{}'='{}' isn't a valid URL."
+                .format(name, value)
+            )
+            return None
+        return urlparse.urlunsplit(url)
+    elif url.scheme == "static":
+        if url.path in ("", "/"):
+            errors.add(
+                "The attribute '{}'='{}' can't have an empty path."
+                .format(name, value)
+            )
+            return None
+        proj = url.netloc
+        if proj == "":
+            proj = "project:static"
+        elif ":" not in proj:
+            proj = "project:{}".format(proj)
+        return request.static_url(
+            "{}{}".format(proj, url.path)
+        )
+    elif url.scheme == "config":
+        if url.netloc == "":
+            errors.add(
+                "The attribute '{}'='{}' can't have an empty netloc."
+                .format(name, value)
+            )
+            return None
+        server = request.registry.settings.get("servers", {}).get(url.netloc, None)
+        if server is None:
+            errors.add(
+                "The server '{}' isn't found in the config".format(url.netloc)
+            )
+            return None
+        return u"{}{}?{}".format(server, url.path, url.query)
+
+
+def get_typed(name, value, types, request, errors):
+    try:
+        if name not in types:
+            # ignore
+            return None
+        type_ = types[name]
+        if type_.get("type", "string") == "string":
+            return value
+        elif type_["type"] == "list":
+            return [v.strip() for v in value.split(",")]
+        elif type_["type"] == "boolean":
+            value = value.lower()
+            if value in ["yes", "y", "on", "1"]:
+                return True
+            elif value in ["no", "n", "off", "0"]:
+                return False
+            else:
+                errors.add(
+                    "The boolean attribute '{}'='{}' is not in "
+                    "[yes, y, on, 1, no, n, off, 0].".format(
+                        name, value.lower()
+                    )
+                )
+        elif type_["type"] == "integer":
+            return int(value)
+        elif type_["type"] == "float":
+            return float(value)
+        elif type_["type"] == "date":
+            date = dateutil.parser.parse(
+                value, default=datetime.datetime(1, 1, 1, 0, 0, 0)
+            )
+            if date.time() != datetime.time(0, 0, 0):
+                errors.add("The date attribute '{}'='{}' shouldn't have any time")
+                return None
+            return datetime.date.strftime(
+                date.date(), "%Y-%m-%d"
+            )
+        elif type_["type"] == "time":
+            date = dateutil.parser.parse(
+                value, default=datetime.datetime(1, 1, 1, 0, 0, 0)
+            )
+            if date.date() != datetime.date(1, 1, 1):
+                errors.add("The time attribute '{}'='{}' shouldn't have any date")
+                return None
+            return datetime.time.strftime(
+                date.time(), "%H:%M:%S"
+            )
+        elif type_["type"] == "datetime":
+            date = dateutil.parser.parse(
+                value, default=datetime.datetime(1, 1, 1, 0, 0, 0)
+            )
+            return datetime.datetime.strftime(
+                date, "%Y-%m-%dT%H:%M:%S"
+            )
+        elif type_["type"] == "url":
+            return get_url2(name, value, request, errors)
+        else:
+            errors.add("Unknown type '{}'.".format(type_["type"]))
+    except Exception as e:
+        errors.add(
+            "Unable to parse the attribute '{}'='{}' with the type '{}', error:\n{}"
+            .format(
+                name, value, type_.get("type", "string"), str(e)
+            )
+        )
+    return None
+
+
 def add_url_params(url, params):
     return add_spliturl_params(urlsplit(url), params)
 

c2cgeoportal/lib/authentication.py

@@ -52,7 +52,7 @@ def create_authentication(settings):
     return MultiAuthenticationPolicy(policies)
 
 
-def c2cgeoportal_check(username, password, request):  # pragma: nocover
+def c2cgeoportal_check(username, password, request):  # pragma: no cover
     if request.registry.validate_user(request, username, password):
         return []
     return None

c2cgeoportal/lib/bashcolor.py

@@ -38,5 +38,5 @@
 WHITE = 7
 
 
-def colorize(text, color):   # pragma: nocover
+def colorize(text, color):   # pragma: no cover
     return "\x1b[01;3%im%s\x1b[0m" % (color, text)

c2cgeoportal/lib/cacheversion.py

@@ -43,7 +43,7 @@ def get_cache_version():
     return uuid.uuid4().hex
 
 
-def version_cache_buster(request, subpath, kw):  # pragma: nocover
+def version_cache_buster(request, subpath, kw):  # pragma: no cover
     return urljoin(get_cache_version() + "/", subpath), kw
 
 

c2cgeoportal/lib/caching.py

@@ -49,14 +49,14 @@ def keygen_function(namespace, fn):
 
     if namespace is None:
         namespace = "%s:%s" % (fn.__module__, fn.__name__)
-    else:  # pragma: nocover
+    else:  # pragma: no cover
         namespace = "%s:%s|%s" % (fn.__module__, fn.__name__, namespace)
 
     args = inspect.getargspec(fn)
     has_self = args[0] and args[0][0] in ("self", "cls")
 
     def generate_key(*args, **kw):
-        if kw:  # pragma: nocover
+        if kw:  # pragma: no cover
             raise ValueError(
                 "key creation function does not accept keyword arguments.")
         parts = [namespace]
@@ -89,7 +89,7 @@ def get_region(region=None):
     """
     try:
         return _regions[region]
-    except KeyError:  # pragma: nocover
+    except KeyError:  # pragma: no cover
         raise Exception(
             "No such caching region. A region must be"
             "initialized before it can be used")
@@ -174,7 +174,7 @@ def set_common_headers(
             response.cache_control.private = True
         else:
             response.cache_control.public = True
-    else:  # pragma: nocover
+    else:  # pragma: no cover
         raise "Invalid cache type"
 
     if hasattr(request, "registry"):

c2cgeoportal/lib/dbreflection.py

@@ -81,7 +81,7 @@ def __get__(self, obj, type=None):
             # and class levels we could return an SQL expression here.
             # The code of hybrid_property in SQLAlchemy illustrates
             # how to do that.
-            raise AttributeError  # pragma: nocover
+            raise AttributeError  # pragma: no cover
         target = getattr(obj, self.target)
         return getattr(target, self.value_attr) if target else None
 
@@ -123,7 +123,7 @@ def _xsd_sequence_callback(tb, cls):
 
 
 def _column_reflect_listener(inspector, table, column_info, engine):
-    if isinstance(column_info["type"], Geometry):  # pragma: nocover
+    if isinstance(column_info["type"], Geometry):  # pragma: no cover
         query = engine.execute(
             sql.text(SQL_GEOMETRY_COLUMNS),
             table_schema=table.schema,

c2cgeoportal/lib/filter_capabilities.py

@@ -311,10 +311,10 @@ def endElement(self, name):  # noqa
         elif name == "Name":
             self.in_name = False
 
-    def startElementNS(self, name, qname, attrs):  # pragma: nocover  # noqa
+    def startElementNS(self, name, qname, attrs):  # pragma: no cover  # noqa
         self._do(lambda: self._downstream.startElementNS(name, qname, attrs))
 
-    def endElementNS(self, name, qname):  # pragma: nocover  # noqa
+    def endElementNS(self, name, qname):  # pragma: no cover  # noqa
         self._do(lambda: self._downstream.endElementNS(name, qname))
 
     def _keep_layer(self, layer_name):
@@ -339,10 +339,10 @@ def characters(self, text):
 
         self._do(lambda: self._accumulator.append(text.encode("utf-8")))
 
-    def ignorableWhitespace(self, ws):  # pragma: nocover  # noqa
+    def ignorableWhitespace(self, ws):  # pragma: no cover  # noqa
         self._do(lambda: self._accumulator.append(ws))
 
-    def processingInstruction(self, target, body):  # pragma: nocover  # noqa
+    def processingInstruction(self, target, body):  # pragma: no cover  # noqa
         self._do(lambda: self._downstream.processingInstruction(target, body))
 
     def skippedEntity(self, name):  # pragma: no cover  # noqa