Initial support for manifest validation

Now uses pylint and mypy

Author: Alan Christie

.github/workflows/build.yaml

@@ -43,6 +43,8 @@ jobs:
       run: |
         pytest --cov=decoder
         pyroma .
+        pylint decoder
+        mypy decoder --install-types --non-interactive
     - name: Build
       run: |
         python setup.py bdist_wheel

.mypy.ini

@@ -0,0 +1,2 @@
+[mypy]
+ignore_missing_imports = True

MANIFEST.in

@@ -1,2 +1,3 @@
 include LICENSE
-include decoder/schema.yaml
+include decoder/job-definition-schema.yaml
+include decoder/manifest-schema.yaml

README.rst

@@ -18,9 +18,10 @@ The following encoding/decoding formats are supported: -
 
 - jinja2 (3.0)
 
-The package also provides a ``validate_job_schema()`` function,
-which can (should) be used to validate the Job definition against the
-built-in schema.
+The package also provides ``validate_job_schema()`` and
+``validate_manifest_schema()`` functions, which can (should) be used to
+validate the Manifests and Job definitions against the
+built-in schemas.
 
 .. _jinja2: https://jinja.palletsprojects.com/en/3.0.x/
 
@@ -36,7 +37,8 @@ Once installed you can validate the definition (expected to be a dictionary
 formed from the definition YAML file) with:
 
 >>> from decoder import decoder
->>> error = decoder.validate_job_schema(job_defintion)
+>>> error = decoder.validate_manifest_schema(manifest)
+>>> error = decoder.validate_job_schema(job_definition)
 
 And run the decoder with:
 

build-requirements.txt

@@ -1,3 +1,4 @@
-pyroma == 3.2
-pytest == 6.2.5
-pytest-cov == 3.0.0
+pyroma
+pytest
+pytest-cov
+mypy

decoder/decoder.py

@@ -15,17 +15,28 @@
 # The modules are expected to be called 'decode_<TextEncoding.lower()>'
 from . import decode_jinja2_3_0
 
-# The (built-in) Job Definition schema...
+# The (built-in) schemas...
 # from the same directory as us.
-_SCHEMA_FILE: str = os.path.join(os.path.dirname(__file__), 'schema.yaml')
+_JD_SCHEMA_FILE: str = os.path.join(os.path.dirname(__file__),
+                                    'job-definition-schema.yaml')
+_MANIFEST_SCHEMA_FILE: str = os.path.join(os.path.dirname(__file__),
+                                          'manifest-schema.yaml')
 
-# Load the schema YAML file now.
+# Load the JD schema YAML file now.
 # This must work as the file is installed along with this module.
-_JOB_SCHEMA: Dict[str, Any] = {}
-assert os.path.isfile(_SCHEMA_FILE)
-with open(_SCHEMA_FILE, 'r', encoding='utf8') as schema_file:
-    _JOB_SCHEMA = yaml.load(schema_file, Loader=yaml.FullLoader)
-assert _JOB_SCHEMA
+assert os.path.isfile(_JD_SCHEMA_FILE)
+with open(_JD_SCHEMA_FILE, 'r', encoding='utf8') as schema_file:
+    _JOB_DEFINITION_SCHEMA: Dict[str, Any] =\
+        yaml.load(schema_file, Loader=yaml.FullLoader)
+assert _JOB_DEFINITION_SCHEMA
+
+# Load the Manifest schema YAML file now.
+# This must work as the file is installed along with this module.
+assert os.path.isfile(_MANIFEST_SCHEMA_FILE)
+with open(_MANIFEST_SCHEMA_FILE, 'r', encoding='utf8') as schema_file:
+    _MANIFEST_SCHEMA: Dict[str, Any] =\
+        yaml.load(schema_file, Loader=yaml.FullLoader)
+assert _MANIFEST_SCHEMA
 
 
 class TextEncoding(enum.Enum):
@@ -34,16 +45,33 @@ class TextEncoding(enum.Enum):
     JINJA2_3_0 = 1      # Encoding that complies with Jinja2 v3.0.x
 
 
+def validate_manifest_schema(manifest: Dict[str, Any]) -> Optional[str]:
+    """Checks the Job Definition Manifest (a preloaded job-definition dictionary)
+    against the built-in schema. If there's an error the error text is
+    returned, otherwise None.
+    """
+    assert isinstance(manifest, dict)
+
+    # Validate the Manifest against our schema
+    try:
+        jsonschema.validate(manifest, schema=_MANIFEST_SCHEMA)
+    except jsonschema.ValidationError as ex:
+        return str(ex.message)
+
+    # OK if we get here
+    return None
+
+
 def validate_job_schema(job_definition: Dict[str, Any]) -> Optional[str]:
     """Checks the Job Definition (a preloaded job-definition dictionary)
     against the built-in schema. If there's an error the error text is
     returned, otherwise None.
     """
-    assert job_definition
+    assert isinstance(job_definition, dict)
 
     # Validate the Job Definition against our schema
     try:
-        jsonschema.validate(job_definition, schema=_JOB_SCHEMA)
+        jsonschema.validate(job_definition, schema=_JOB_DEFINITION_SCHEMA)
     except jsonschema.ValidationError as ex:
         return str(ex.message)
 

decoder/schema.yaml renamed to decoder/job-definition-schema.yaml

@@ -21,7 +21,7 @@ properties:
     type: string
     maxLength: 80
   description:
-    tyep: string
+    type: string
   collection:
     type: string
     minLength: 1
@@ -107,7 +107,7 @@ definitions:
       name:
         type: string
         minLength: 1
-        maxLenght: 120
+        maxLength: 120
       tag:
         type: string
         minLength: 1

decoder/manifest-schema.yaml

@@ -0,0 +1,31 @@
+---
+# The JSONSchema for JobDefinition 'Manifest' YAML files.
+#
+# See https://json-schema.org/understanding-json-schema/index.html
+
+$schema: http://json-schema.org/draft-07/schema#
+
+title: Data Manager Job Definition Manifest
+description: >-
+  A manifest for Job Definition files
+
+type: object
+properties:
+  kind:
+    const: DataManagerManifest
+  kind-version:
+    enum:
+    - '2021.1'
+  description:
+    type: string
+  job-definition-files:
+    type: array
+    items:
+      type: string
+      pattern: '^.{1,}(.yaml|.yml)$'
+      maxLength: 255
+    minItems: 1
+required:
+- kind
+- kind-version
+- job-definition-files

tests/test_validate_manifest.py

@@ -0,0 +1,26 @@
+# Tests for the schema validator.
+from typing import Any, Dict
+from copy import deepcopy
+
+import pytest
+pytestmark = pytest.mark.unit
+
+from decoder import decoder
+
+# A minimal Job Definition.
+# Tests can use this and adjust accordingly.
+_MINIMAL: Dict[str, Any] = {
+        'kind': 'DataManagerManifest',
+        'kind-version': '2021.1',
+        'job-definition-files': ['blob.yaml',
+                                 'blob.yml']}
+
+
+def test_validate_minimal():
+    # Arrange
+
+    # Act
+    error = decoder.validate_manifest_schema(_MINIMAL)
+
+    # Assert
+    assert error is None