schemas: add url map converter for version

Signed-off-by: Parth Shandilya <[email protected]>

Author: ParthS007

cap/converter.py

@@ -0,0 +1,24 @@
+"""URL Map Converter module."""
+
+import re
+
+from flask import abort
+from werkzeug.routing import BaseConverter
+
+SCHEMA_VERSION_REGEXP = r'^\d+\.\d+\.\d+$'
+
+
+class SchemaVersionConverter(BaseConverter):
+    """Converter class for validating schema version number."""
+
+    def __init__(self, url_map):
+        """Initialise converter."""
+        super().__init__(url_map)
+
+    def to_python(self, value):
+        if not bool(re.match(SCHEMA_VERSION_REGEXP, value)):
+            abort(404)
+        return value
+
+    def to_url(self, value):
+        return value

cap/factory.py

@@ -18,6 +18,13 @@
 from invenio_cache import BytecodeCache
 from jinja2 import ChoiceLoader, FileSystemLoader
 
+from .converter import SchemaVersionConverter
+
+CAP_URL_MAP_CONVERTER = {
+    # Key should match the variable name in URL
+    'schema_version': SchemaVersionConverter,
+}
+
 
 def config_loader(app, **kwargs_config):
     """Add loading templates."""
@@ -55,6 +62,7 @@ def config_loader(app, **kwargs_config):
     blueprint_entry_points=['invenio_base.api_blueprints'],
     extension_entry_points=['invenio_base.api_apps'],
     converter_entry_points=['invenio_base.api_converters'],
+    converters=CAP_URL_MAP_CONVERTER,
     wsgi_factory=wsgi_proxyfix(),
     instance_path=instance_path,
     app_class=app_class(),

cap/modules/schemas/utils.py

@@ -25,12 +25,14 @@
 
 import re
 from functools import wraps
+
 from flask import abort
-from .permissions import AdminSchemaPermission
-from .models import Schema
 from invenio_jsonschemas.errors import JSONSchemaNotFound
 from jsonpatch import JsonPatchConflict
 
+from .models import Schema
+from .permissions import AdminSchemaPermission
+
 
 def is_later_version(version1, version2):
     matched1 = re.match(r"(\d+)\.(\d+)\.(\d+)", version1)
@@ -143,6 +145,7 @@ def get_default_mapping(name, version):
 
 def get_schema(f):
     """Decorator to check if schema exists by name and/or version."""
+
     @wraps(f)
     def wrapper(name=None, version=None, schema=None, *args, **kwargs):
         if name:
@@ -154,14 +157,17 @@ def wrapper(name=None, version=None, schema=None, *args, **kwargs):
             except JSONSchemaNotFound:
                 abort(404)
         return f(name=name, version=version, schema=schema, *args, **kwargs)
+
     return wrapper
 
 
 def schema_admin_permission(f):
     """Decorator to check if user has admin permission."""
+
     @wraps(f)
     def wrapper(name=None, version=None, schema=None, *args, **kwargs):
         if not AdminSchemaPermission(schema).can():
             abort(403)
         return f(name=name, version=version, schema=schema, *args, **kwargs)
+
     return wrapper

cap/modules/schemas/views.py

@@ -87,7 +87,7 @@ def get_all_versions(name=None):
     '/<string:name>/permissions', methods=['GET', 'POST', 'DELETE']
 )
 @blueprint.route(
-    '/<string:name>/<string:version>/permissions',
+    '/<string:name>/<schema_version:version>/permissions',
     methods=['GET', 'POST', 'DELETE'],
 )
 # @login_required
@@ -117,11 +117,9 @@ def permissions(name=None, version=None):
         return jsonify({}), 204
 
 
+@blueprint.route('/<string:name>/notifications', methods=['GET', 'PATCH'])
 @blueprint.route(
-    '/<string:name>/notifications', methods=['GET', 'PATCH']
-)
-@blueprint.route(
-    '/<string:name>/<string:version>/notifications',
+    '/<string:name>/<schema_version:version>/notifications',
     methods=['GET', 'PATCH'],
 )
 @get_schema
@@ -130,30 +128,36 @@ def permissions(name=None, version=None):
 def notifications_config(name=None, version=None, schema=None, *args, **kwargs):
     """CRUD operations for schema configuration."""
     serialized_config = schema.config_serialize()
-    notifications_config = serialized_config.get("config", {}).get("notifications", {})
+    notifications_config = serialized_config.get("config", {}).get(
+        "notifications", {}
+    )
     if request.method == "PATCH":
         try:
             data = request.get_json()
-            patched_notifications_config = apply_patch(notifications_config, data)
-            patched_object = {'config': {'notifications': patched_notifications_config}}
+            patched_notifications_config = apply_patch(
+                notifications_config, data
+            )
+            patched_object = {
+                'config': {'notifications': patched_notifications_config}
+            }
             schema.update(**patched_object)
             db.session.commit()
             return jsonify(schema.config_serialize()), 201
         except (
-                JsonPatchException,
-                JsonPatchConflict,
-                JsonPointerException,
-                TypeError,
-            ) as err:
-                return (
-                    jsonify(
-                        {
-                            'message': 'Could not apply '
-                            'json-patch to object: {}'.format(err)
-                        }
-                    ),
-                    400,
-                )
+            JsonPatchException,
+            JsonPatchConflict,
+            JsonPointerException,
+            TypeError,
+        ) as err:
+            return (
+                jsonify(
+                    {
+                        'message': 'Could not apply '
+                        'json-patch to object: {}'.format(err)
+                    }
+                ),
+                400,
+            )
         except IntegrityError:
             return (
                 jsonify(
@@ -401,7 +405,7 @@ def patch(self, name, version):
     ],
 )
 blueprint.add_url_rule(
-    '/<string:name>/<string:version>',
+    '/<string:name>/<schema_version:version>',
     view_func=schema_view_func,
     methods=['GET', 'PUT', 'DELETE', 'PATCH'],
 )

tests/integration/schemas/test_schemas_views.py

@@ -93,6 +93,53 @@ def test_get_when_user_outside_of_experiment_returns_403(
     assert resp.status_code == 403
 
 
+def test_get_schema_version_converter_with_wrong_version(client, db, users, auth_headers_for_user, superuser):
+    cms_user = users['cms_user']
+    schema = Schema(
+        name='cms-schema',
+        version='1.2.3',
+        fullname='CMS Schema 1.2.3',
+        experiment='CMS',
+        deposit_schema={'title': 'deposit_schema'},
+        deposit_options={'title': 'deposit_options'},
+        record_schema={'title': 'record_schema'},
+        record_options={'title': 'record_options'},
+        record_mapping={
+            'mappings': {
+                'properties': {
+                    'title': {
+                        'type': 'text'
+                    }
+                }
+            }
+        },
+        deposit_mapping={
+            'mappings':
+                {
+                    'properties': {
+                        'keyword': {
+                            'type': 'keyword'
+                        }
+                    }
+                }
+        },
+        is_indexed=True,
+    )
+
+    db.session.add(schema)
+    db.session.commit()
+
+    resp = client.get('/jsonschemas/cms-schema/1.0.0', headers=auth_headers_for_user(cms_user))
+    assert resp.status_code == 404
+
+    resp = client.get('/jsonschemas/cms-schema/noti', headers=auth_headers_for_user(cms_user))
+    assert resp.status_code == 404
+
+    resp = client.get('/jsonschemas/cms-schema/notifications', headers=auth_headers_for_user(superuser))
+    assert resp.status_code == 200
+    assert resp.json == {}
+
+
 def test_get(client, db, users, auth_headers_for_user):
     cms_user = users['cms_user']
     schema = Schema(