Merge pull request #13 from lyubick/issues-7-11

ISSUE #11: Ignore empty file option added

.github/workflows/self-test.yaml

@@ -24,4 +24,10 @@ jobs:
         uses: ./
         with:
           json-schema-file: test/schema/json_schema.json
-          yaml-json-file-dir: test/emptyJSONs,test/emptyYAMLSs,
+          yaml-json-file-dir: test/emptyJSONs,test/emptyYAMLSs
+      - name: Run Action (check mapping)
+        uses: ./
+        with:
+          json-schema-file: test/schema/json_schema.json
+          yaml-json-file-dir: test/emptyJSONs,test/emptyYAMLSs
+          schema-mapping: "test/emptyJSONs:test/schema/json_schema.json,test/emptyYAMLSs:test/schema/json_schema.json"

README.md

@@ -21,13 +21,17 @@ jobs:
           json-schema-file: path/to/my/cool/schema.json
           yaml-json-file-dir: path/to/my/cool/yaml/file.yaml
           recursive: false
+          ignore-empty: false
+          schema-mapping: 'path/to/my/cool/yaml/file.yaml:path/to/my/cool/schema.json'
 ```
 One should provide two parameters:
-- `json-schema-file`, points to legit JSON Schema file
-- `yaml-json-file-dir`, is a comma separated list that contains
+- `json-schema-file`, required, points to legit JSON Schema files. In case of mapping this schema will be used as default (fallback) schema.
+- `yaml-json-file-dir`, is a comma separated list that contains:
   - Single YAML or JSON files
   - Directories that will be parsed for `.yaml` or `.yml` or `.json` files
-- `recursive`, True/False depending on if recursive scan for YAML or JSON files in directory required
+- `recursive`, optional, True/False depending on if recursive scan for YAML or JSON files in directory required. Default is False.
+- `ignore-empty`, optional, True/False depends if one want to cause failure on empty files or not, default is True.
+- `schema-mapping`, optional, one can provide file-to-schema mapping, if specific files require specific schema.
 
 ## Results
 ### Success
@@ -95,5 +99,5 @@ On instance:
     {'field2': 'Value2'}
 ```
 
-## Supported & Used by
-* [Printify](https://github.com/printify)
+## Organisations Use
+* [Printify](https://printify.com)

action.yaml

@@ -7,7 +7,9 @@ branding:
   color: green
 inputs:
   json-schema-file:
-    description: 'JSON Schema file to validate against'
+    description: |
+      JSON Schema file to validate against. In case mapping is provided, schema
+      provided here will be used as a default (fallback) schema.
     required: true
   yaml-json-file-dir:
     description: |
@@ -26,6 +28,13 @@ inputs:
       otherwise empty files will be ignored.
     required: false
     default: 'true'
+  schema-mapping:
+    description: |
+      File/Directory and Schema mapping, provided in a one of formats:
+      1. /path/to/file:/path/to/schema.json
+      2. /path/to/dir/:/path/to/schema.json
+    required: false
+    default: ''
 runs:
   using: 'docker'
   image: 'Dockerfile'
@@ -34,3 +43,4 @@ runs:
     - ${{ inputs.yaml-json-file-dir }}
     - ${{ inputs.recursive }}
     - ${{ inputs.ignore-empty }}
+    - ${{ inputs.schema-mapping }}

test/test_sanity.py

@@ -2,86 +2,134 @@
 from pathlib import Path
 
 import validator
-import jsonschema
 
 
 class Test:
     abs_path = Path(__file__).parent
 
     def test1_load_schema(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        assert schema['title'] == 'TestConfig'
-        assert schema['description'] == 'Test Basic YAML File'
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        assert len(schemas.keys()) == 1
+        assert list(schemas.values())[0]['title'] == 'TestConfig'
+        assert list(schemas.values())[0]['description'] == 'Test Basic YAML File'
 
     def test2_get_files(self):
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/YAMLs', is_recursive=True)
+        files = validator.get_testing_filenames(f'{self.abs_path}/YAMLs', is_recursive=True)
         assert len(files) == 2
 
     def test3_validate_valid_file(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/YAMLs/valid.yaml', False)
-        assert validator.validate_files(files, schema)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/YAMLs/valid.yaml', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
+        assert validator.validate_files(files_schemas)
 
     def test4_validate_invalid_file(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/YAMLs/invalid.yaml', False)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/YAMLs/invalid.yaml', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
         try:
-            validator.validate_files(files, schema)
+            validator.validate_files(files_schemas)
             assert False
         except Exception as exc:
             assert len(exc.args) == 1
             assert len(exc.args[0]) == 1
             assert exc.args[0][0][1] == {'field2': 'Value2'}
 
     def test5_validate_valid_file_json(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/JSONs/valid.json', False)
-        assert validator.validate_files(files, schema)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/JSONs/valid.json', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
+        assert validator.validate_files(files_schemas)
 
     def test6_validate_invalid_file_json(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/JSONs/invalid.json', False)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/JSONs/invalid1.json', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
         try:
-            validator.validate_files(files, schema)
+            validator.validate_files(files_schemas)
             assert False
         except Exception as exc:
             assert len(exc.args) == 1
             assert len(exc.args[0]) == 1
             assert exc.args[0][0][1] == {'field2': 'Value2'}
 
     def test7_validate_folder_yaml(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/YAMLs/', False)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/YAMLs/', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
         try:
-            validator.validate_files(files, schema)
+            validator.validate_files(files_schemas)
             assert False
         except Exception as exc:
             assert len(exc.args) == 1
             assert len(exc.args[0]) == 1
             assert exc.args[0][0][1] == {'field2': 'Value2'}
 
     def test8_validate_folder_json(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/JSONs/', False)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/JSONs/', False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
         try:
-            validator.validate_files(files, schema)
+            validator.validate_files(files_schemas)
             assert False
         except Exception as exc:
             assert len(exc.args) == 1
             assert len(exc.args[0]) == 2
-            assert exc.args[0][1][1] == {'field2': 'Value2'}
-            assert exc.args[0][0][1] == {'field2': 'Value2_2'}
+            assert exc.args[0][0][1] == {'field2': 'Value2'}
+            assert exc.args[0][1][1] == {'field2': 'Value2_2'}
 
     def test9_validate_empty_json(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/emptyJSONs/empty.json', False, True)
-        validator.validate_files(files, schema)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/emptyJSONs/empty.json', False, True)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
+        validator.validate_files(files_schemas)
 
     def test9_validate_empty_json_fail(self):
-        schema = validator.load_schema(f'{self.abs_path}/schema/json_schema.json')
-        files = validator.get_yaml_json_files_list(f'{self.abs_path}/emptyJSONs/empty.json', False, False)
+        schemas = validator.get_all_schemas(
+            schema_file_path='',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/emptyJSONs/empty.json', False, False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, None)
+        try:
+            validator.validate_files(files_schemas)
+            assert False
+        except json.JSONDecodeError as exc:
+            assert True
+
+    def test10_validate_file_with_schema_map(self):
+        schemas = validator.get_all_schemas(
+            schema_file_path=f'{self.abs_path}/schema/json_schema.json',
+            default_schema_path=f'{self.abs_path}/schema/json_schema.json'
+        )
+        files = validator.get_testing_filenames(f'{self.abs_path}/emptyJSONs/empty.json', False, False)
+        files_schemas = validator.get_filenames_with_schema(files, schemas, f"{self.abs_path}/emptyJSONs/empty.json:{self.abs_path}/schema/json_schema.json")
         try:
-            validator.validate_files(files, schema)
+            validator.validate_files(files_schemas)
             assert False
         except json.JSONDecodeError as exc:
             assert True

validator.py

@@ -1,53 +1,87 @@
 import json
+import logging
 import os.path
 import pathlib
+import re
 import sys
+from typing import List, Tuple, Optional
 
 import yaml
 from jsonschema import validate
 from jsonschema.exceptions import ValidationError
 
 
-def load_schema(schema_file_path: str) -> json:
-    if os.path.exists(schema_file_path):
-        if os.path.isfile(schema_file_path):
-            with open(schema_file_path, 'r') as stream:
-                return json.loads("".join(stream.readlines()))
-        else:
-            raise f'Provided JSON Schema is not a file! Please provide legit JSON Schema file.'
-    else:
-        raise f'Provided JSON Schema file does not exist!'
-
-
-def get_yaml_json_files_list(files_paths_list: str, is_recursive: bool, ignore_empty_files: bool = False) -> list[str]:
-    yaml_input = list(files_paths_list.split(','))
-
-    yaml_files = []
-    for yaml_object in yaml_input:
-        if os.path.isdir(yaml_object):
-            yaml_files.extend(
-                list(
-                    map(
-                        lambda f: str(f),  # Convert all paths to string, instead of Posix Path
-                        filter(
-                            lambda f: str(f).endswith('.yaml') or str(f).endswith('.yml') or str(f).endswith('.json'),
-                            pathlib.Path(yaml_object).glob('**/*' if is_recursive else '*')
+def get_all_filenames(input_path: str, endings: List[str], is_recursive: bool) -> List[str]:
+    paths = list(input_path.split(','))
+
+    regex_endings = f'.*\\.({"|".join(endings)})'
+
+    output_files = []
+
+    for path in paths:
+        if os.path.exists(path):
+            if os.path.isdir(path):
+                output_files.extend(
+                    list(
+                        map(
+                            lambda f: str(f),  # Convert all paths to string, instead of Posix Path
+                            filter(
+                                lambda f: len(re.findall(regex_endings, str(f))) > 0,
+                                pathlib.Path(path).glob('**/*' if is_recursive else '*')
+                            )
                         )
                     )
                 )
-            )
-        elif os.path.isfile(yaml_object):
-            yaml_files.append(yaml_object)
+            elif os.path.isfile(path):
+                output_files.append(path)
+            else:
+                continue
+
+    return output_files
+
+
+def get_all_schemas(schema_file_path: str, default_schema_path: str) -> dict[str, dict]:
+    schema_files = get_all_filenames(schema_file_path, endings=['json'], is_recursive=False)
+    schemas = list(map(lambda x: (x, json.loads(''.join(open(x, 'r').readlines()))), schema_files))
+    schemas.append(('default', json.loads(''.join(open(default_schema_path, 'r').readlines()))))
+    return dict(schemas)
+
+
+def get_testing_filenames(files_paths_list: str, is_recursive: bool, ignore_empty_files: bool = False) -> list[str]:
+    yaml_files = get_all_filenames(files_paths_list, endings=['json', 'yaml', 'yml'], is_recursive=is_recursive)
 
     if ignore_empty_files:
         yaml_files = list(filter(lambda f: os.path.getsize(f) > 0, yaml_files))
 
-    return yaml_files
+    return sorted(yaml_files)
+
+
+def get_filenames_with_schema(
+        test_files: List[str],
+        schemas: dict[str, dict],
+        mapping_str: Optional[str]) -> List[Tuple[str, dict]]:
+
+    def map_schema(filename: str, schema_map: dict[str, str]) -> Tuple[str, dict]:
+        for s in schema_map.keys():
+            if filename in get_all_filenames(input_path=s, endings=['yaml', 'json', 'yml'], is_recursive=True):
+                return filename, schemas[schema_map[s]]
 
+        return filename, schemas['default']
 
-def validate_files(yaml_files: list, json_schema: json):
+    if mapping_str:
+        mapping = dict(list(map(lambda x: tuple_split(x, ':'), list(mapping_str.split(',')))))
+        files_schema = list(map(lambda x: map_schema(x, mapping), test_files))
+    else:
+        files_schema = list(map(lambda x: (x, schemas['default']), test_files))
+
+    return files_schema
+
+
+def validate_files(files_with_schema: List[Tuple[str, dict]]):
     failed = []
-    for yaml_file in yaml_files:
+    for file_with_schema in files_with_schema:
+        yaml_file = file_with_schema[0]
+        json_schema = file_with_schema[1]
         if os.path.exists(yaml_file):
             if os.path.isfile(yaml_file):
                 with open(yaml_file, 'r') as stream:
@@ -71,9 +105,41 @@ def validate_files(yaml_files: list, json_schema: json):
     return True
 
 
+def tuple_split(inp: str, separator: str) -> Tuple[str, str]:
+    """
+
+    :param inp: String in a form of <key><separator><value>, that will be split into (key, value) tuple.
+    :param separator: String representing separator.
+    :return: filename: str, schema: str
+    """
+    values = inp.split(separator)
+    return values[0], values[1]
+
+
 if __name__ == '__main__':
     args = sys.argv[1:]
 
-    schema = load_schema(args[0])
-    files = get_yaml_json_files_list(args[1], args[2].lower() == 'true')
-    validate_files(files, schema)
+    input_mapping = args[4]
+
+    input_schemas = {}
+
+    input_mapped_schemas = ''
+    if input_mapping:
+        logging.error(input_mapping)
+        input_mapped_schemas = ','.join(list(map(lambda x: tuple_split(x, ':')[1], input_mapping.split(','))))
+
+    input_schemas = get_all_schemas(schema_file_path=input_mapped_schemas, default_schema_path=args[0])
+
+    input_files = get_testing_filenames(
+        files_paths_list=args[1],
+        is_recursive=args[2].lower() == 'true',
+        ignore_empty_files=args[3].lower() == 'true'
+    )
+
+    input_files_with_schema = get_filenames_with_schema(
+        test_files=input_files,
+        schemas=input_schemas,
+        mapping_str=input_mapping
+    )
+
+    validate_files(input_files_with_schema)