diff --git a/examples/elabftw/SKIP_CI b/examples/elabftw/SKIP_CI new file mode 100644 index 0000000..e69de29 diff --git a/tests/checks.py b/tests/checks.py new file mode 100644 index 0000000..9df0bc9 --- /dev/null +++ b/tests/checks.py @@ -0,0 +1,209 @@ +import json +import traceback +import tempfile +from pathlib import Path +from zipfile import ZIP_DEFLATED +from zipfile import ZipFile +from jsonschema import Draft202012Validator +from rocrate.rocrate import ROCrate +from rocrate_validator import services, models + +METADATA_FILE = 'ro-crate-metadata.json' + +def checkPypiRocrate(fileName, verbose=False): + """ Check if file is a valid ro-crate according to pypi rocrate package + https://pypi.org/project/rocrate/ + Args: + fileName: path to .eln file + verbose: bool if verbose output should be generated + Returns: + success: bool if check was successful + log: string with log information + """ + with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: + try: + dirpath = Path(tempfile.mkdtemp()) + elnFile.extractall(dirpath) + tempPath= [i for i in dirpath.iterdir() if i.is_dir()][0] + crate = ROCrate(tempPath) + log = "" + for e in crate.get_entities(): + if verbose: + log+= f' {e.id}: {e.type}\n' + return True, log + except Exception: + log = " ***** ERROR: Could not parse content of this file!! *****\n" + log+= f" Temporary folder: {tempPath}\n" + log+= traceback.format_exc() + return False, log + + +def checkParamMetadataJson(fileName): + """ Check if file is a valid ro-crate according to pypi rocrate package + Args: + fileName: path to .eln file + Returns: + success: bool if check was successful + log: string with log information + """ + # global variables worth discussion + ROCRATE_NOTE_MANDATORY = ['version','sdPublisher'] + DATASET_MANDATORY = ['name'] + DATASET_SUGGESTED = ['author','mentions', 'dateCreated', 'dateModified', 'identifier', 'text', 'keywords'] + FILE_MANDATORY = ['name'] + FILE_SUGGESTED = ['sha256', 'encodingFormat', 'contentSize', 'description'] + + # runtime global variables + OUTPUT_INFO = False + OUTPUT_COUNTS = False + KNOWN_KEYS = DATASET_MANDATORY+DATASET_SUGGESTED+FILE_MANDATORY+FILE_SUGGESTED+['@id', '@type'] + + def processNode(graph, nodeID): + """ + recursive function call to process each node + + Args: + graph: full graph + nodeID: id of node in graph + """ + globalSuccess = True + nodes = [ i for i in graph if '@id' in i and i['@id'] == nodeID] + if len(nodes)!=1: + print('**ERROR: all entries must only occur once in crate. check:', nodeID) + return + node = nodes[0] + # CHECK IF MANDATORY AND SUGGESTED KEYWORDS ARE PRESENT + if '@type' not in node: + print('**ERROR: all nodes must have @type. check:', nodeID) + return False + if node['@type'] == 'Dataset': + for key in DATASET_MANDATORY: + if key not in node: + print(f'**ERROR in dataset: "{key}" not in @id={node["@id"]}') + globalSuccess = False + for key in DATASET_SUGGESTED: + if key not in node and OUTPUT_INFO: + print(f'**INFO for dataset: "{key}" not in @id={node["@id"]}') + elif node['@type'] == 'File': + for key in FILE_MANDATORY: + if key not in node: + print(f'**ERROR in file: "{key}" not in @id={node["@id"]}') + globalSuccess = False + for key in FILE_SUGGESTED: + if key not in node and OUTPUT_INFO: + print(f'**INFO for file: "{key}" not in @id={node["@id"]}') + # CHECK PROPERTIES FOR ALL KEYS + if any(not str(i).strip() for i in node.values()): + print(f'**WARNING: {nodeID} contains empty values in the key-value pairs') + # SPECIFIC CHECKS ON CERTAIN KEYS + if isinstance(node.get('keywords', ''), list): + print(f'**ERROR: {nodeID} contains an array of keywords. Use comma or space separated string') + globalSuccess = False + # recurse to children + children = node.pop('hasPart') if 'hasPart' in node else [] + for child in children: + globalSuccess = processNode(graph, child['@id']) and globalSuccess + return globalSuccess + + success = True + log = '' + with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: + success = True + metadataJsonFile = [i for i in elnFile.namelist() if i.endswith(METADATA_FILE)][0] + metadataContent = json.loads(elnFile.read(metadataJsonFile)) + graph = metadataContent["@graph"] + # find information from master node + ro_crate_nodes = [i for i in graph if i["@id"] == METADATA_FILE] + if len(ro_crate_nodes) == 1: + for key in ROCRATE_NOTE_MANDATORY: + if key not in ro_crate_nodes[0]: + log += f'**ERROR: "{key}" not in @id={METADATA_FILE}' + else: + log += f'**ERROR: @id={METADATA_FILE} does not uniquely exist ' + success = False + main_node = [i for i in graph if i["@id"] == "./"][0] + + # iteratively go through graph + for partI in main_node['hasPart']: + success = processNode(graph, partI['@id']) and success + + # count occurances of all keys + counts = {} + for node in graph: + if node['@id'] in ['./',METADATA_FILE]: + continue + for key in node.keys(): + if key in counts: + counts[key] += 1 + else: + counts[key] = 1 + + view = [ (v,k) for k,v in counts.items() ] + view.sort(reverse=True) + if OUTPUT_COUNTS: + log += '===== Counts (* unspecified)' + for v,k in view: + prefix = ' ' if k in KNOWN_KEYS else ' * ' + log += f'{prefix}{k:15}: {v}' + return success, log + + +def checkSchema(fileName): + """ Check if file is a valid ro-crate according to THE-ELN-Consortium schema + Args: + fileName: path to .eln file + Returns: + success: bool if check was successful + log: string with log information + """ + log = '' + schema = json.load(open(Path(__file__).parent/'schema.json', 'r', encoding='utf-8')) + validator = Draft202012Validator(schema=schema) + validator.check_schema(schema=schema) + success = True + with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: + metadataJsonFile = [i for i in elnFile.namelist() if i.endswith(METADATA_FILE)][0] + metadataContent = json.loads(elnFile.read(metadataJsonFile)) + for error in sorted(validator.iter_errors(metadataContent), key=str): + log += f'- {error.message}' + success = False + return success, log + + +def checkValidator(fileName): + """ Check if file is a valid ro-crate according to rocrate-validator + Args: + fileName: path to .eln file + Returns: + success: bool if check was successful + log: string with log information + """ + log = '' + success = True + with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: + dirpath = Path(tempfile.mkdtemp())/Path(fileName).parent.name + dirpath.mkdir(parents=True, exist_ok=True) + elnFile.extractall(dirpath) + rocrate_dir= [i for i in dirpath.iterdir() if i.is_dir()][0] + + # short stop-gap measure because the validator currently does not support the latest ro-crate version + with open(rocrate_dir/METADATA_FILE, 'r', encoding='utf-8') as f: + metadata = f.read() + metadata = metadata.replace('conformsTo":{"@id":"https:\/\/w3id.org\/ro\/crate\/1.2"}', + 'conformsTo":{"@id":"https:\/\/w3id.org\/ro\/crate\/1.1"}') + with open(rocrate_dir/METADATA_FILE, 'w', encoding='utf-8') as f: + f.write(metadata) + + # start validation + settings = services.ValidationSettings( + rocrate_uri=rocrate_dir, + profile_identifier='ro-crate-1.1', + requirement_severity=models.Severity.REQUIRED, + ) + result = services.validate(settings) + if result.has_issues(): + log += f'{fileName} is not valid\n' + for issue in result.get_issues(): + log += f"Detected issue of severity {issue.severity.name} with check \"{issue.check.identifier}\": {issue.message}\n" + success = False + return success, log diff --git a/tests/test_00_pypi_rocrate.py b/tests/test_00_pypi_rocrate.py index 97be534..630e74c 100644 --- a/tests/test_00_pypi_rocrate.py +++ b/tests/test_00_pypi_rocrate.py @@ -5,15 +5,41 @@ """ import os import json -import unittest, traceback -import tempfile +import unittest from pathlib import Path -from zipfile import ZIP_DEFLATED -from zipfile import ZipFile -from rocrate.rocrate import ROCrate +from checks import checkPypiRocrate + +def generalizedTest(checkFunction, label): + """ + generalized test function to reduce code duplication + + Args: + checkFunction: function to be tested + label: label to be used in logging.json + """ + if Path('tests/logging.json').exists(): + logJson = json.load(open('tests/logging.json')) + else: + logJson = {} + success = True + for root, _, files in os.walk(".", topdown=False): + if 'SKIP_CI' in files: + continue + for name in files: + if not name.endswith('.eln'): + continue + fileName = os.path.join(root, name) + print(f'\n\nTest 00: {fileName}') + successI, log = checkFunction(fileName) + print(log) + if fileName not in logJson: + logJson[fileName] = {label:successI} + else: + logJson[fileName] = logJson[fileName] | {label:successI} + success = success and successI + json.dump(logJson, open('tests/logging.json', 'w')) + assert success -LABEL = 'pypi_rocrate' -verbose = False class Test_1(unittest.TestCase): """ @@ -23,43 +49,4 @@ def test_main(self): """ main function """ - # log-file - if Path('tests/logging.json').exists(): - logJson = json.load(open('tests/logging.json')) - else: - logJson = {} - - success = True - for root, _, files in os.walk(".", topdown=False): - if 'SKIP_CI' in files: - continue - for name in files: - if not name.endswith('.eln'): - continue - fileName = os.path.join(root, name) - print(f'\n\nTest 00: {fileName}') - with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: - dirName = os.path.splitext(os.path.basename(fileName))[0] - try: - dirpath = Path(tempfile.mkdtemp()) - elnFile.extractall(dirpath) - tempPath= [i for i in dirpath.iterdir() if i.is_dir()][0] - crate = ROCrate(tempPath) - for e in crate.get_entities(): - if verbose: - print(f' {e.id}: {e.type}') - if fileName not in logJson: - logJson[fileName] = {LABEL:True} - else: - logJson[fileName] = logJson[fileName] | {LABEL:True} - except Exception: - print(" ***** ERROR: Could not parse content of this file!! *****") - print(f" Temporary folder: ",tempPath) - print(traceback.format_exc()) - if fileName not in logJson: - logJson[fileName] = {LABEL:False} - else: - logJson[fileName] = logJson[fileName] | {LABEL:False} - success = False - json.dump(logJson, open('tests/logging.json', 'w')) - assert success + generalizedTest(checkPypiRocrate, 'pypi_rocrate') diff --git a/tests/test_01_params_metadata_json.py b/tests/test_01_params_metadata_json.py index dbf09f9..240cdfa 100644 --- a/tests/test_01_params_metadata_json.py +++ b/tests/test_01_params_metadata_json.py @@ -1,13 +1,8 @@ #!/usr/bin/python3 """ This tests against rules that we as the ELN consortium set for ourselves """ -import os -import json -from pathlib import Path import unittest -from zipfile import ZIP_DEFLATED -from zipfile import Path as ZPath -from zipfile import ZipFile - +from checks import checkParamMetadataJson +from test_00_pypi_rocrate import generalizedTest class Test_2(unittest.TestCase): """ @@ -17,123 +12,4 @@ def test_main(self): """ main function """ - # global variables worth discussion - ROCRATE_NOTE_MANDATORY = ['version','sdPublisher'] - DATASET_MANDATORY = ['name'] - DATASET_SUGGESTED = ['author','mentions', 'dateCreated', 'dateModified', 'identifier', 'text', 'keywords'] - FILE_MANDATORY = ['name'] - FILE_SUGGESTED = ['sha256', 'encodingFormat', 'contentSize', 'description'] - - # runtime global variables - METADATA_FILE = 'ro-crate-metadata.json' - OUTPUT_INFO = False - OUTPUT_COUNTS = False - KNOWN_KEYS = DATASET_MANDATORY+DATASET_SUGGESTED+FILE_MANDATORY+FILE_SUGGESTED+['@id', '@type'] - LABEL = 'params_metadata_json' - - # log-file - if Path('tests/logging.json').exists(): - logJson = json.load(open('tests/logging.json')) - else: - logJson = {} - - def processNode(graph, nodeID): - """ - recursive function call to process each node - - Args: - graph: full graph - nodeID: id of node in graph - """ - globalSuccess = True - nodes = [ i for i in graph if '@id' in i and i['@id'] == nodeID] - if len(nodes)!=1: - print('**ERROR: all entries must only occur once in crate. check:', nodeID) - return - node = nodes[0] - # CHECK IF MANDATORY AND SUGGESTED KEYWORDS ARE PRESENT - if '@type' not in node: - print('**ERROR: all nodes must have @type. check:', nodeID) - return False - if node['@type'] == 'Dataset': - for key in DATASET_MANDATORY: - if key not in node: - print(f'**ERROR in dataset: "{key}" not in @id={node["@id"]}') - globalSuccess = False - for key in DATASET_SUGGESTED: - if key not in node and OUTPUT_INFO: - print(f'**INFO for dataset: "{key}" not in @id={node["@id"]}') - elif node['@type'] == 'File': - for key in FILE_MANDATORY: - if key not in node: - print(f'**ERROR in file: "{key}" not in @id={node["@id"]}') - globalSuccess = False - for key in FILE_SUGGESTED: - if key not in node and OUTPUT_INFO: - print(f'**INFO for file: "{key}" not in @id={node["@id"]}') - # CHECK PROPERTIES FOR ALL KEYS - if any(not str(i).strip() for i in node.values()): - print(f'**WARNING: {nodeID} contains empty values in the key-value pairs') - # SPECIFIC CHECKS ON CERTAIN KEYS - if isinstance(node.get('keywords', ''), list): - print(f'**ERROR: {nodeID} contains an array of keywords. Use comma or space separated string') - globalSuccess = False - # recurse to children - children = node.pop('hasPart') if 'hasPart' in node else [] - for child in children: - globalSuccess = processNode(graph, child['@id']) and globalSuccess - return globalSuccess - - for root, _, files in os.walk(".", topdown=False): - if 'SKIP_CI' in files: - continue - for name in files: - if not name.endswith('.eln'): - continue - fileName = os.path.join(root, name) - print(f'\n\nTest 01: {fileName}') - with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: - success = True - metadataJsonFile = [i for i in elnFile.namelist() if i.endswith(METADATA_FILE)][0] - metadataContent = json.loads(elnFile.read(metadataJsonFile)) - graph = metadataContent["@graph"] - # find information from master node - ro_crate_nodes = [i for i in graph if i["@id"] == METADATA_FILE] - if len(ro_crate_nodes) == 1: - for key in ROCRATE_NOTE_MANDATORY: - if key not in ro_crate_nodes[0]: - print(f'**ERROR: "{key}" not in @id={METADATA_FILE}') - else: - print(f'**ERROR: @id={METADATA_FILE} does not uniquely exist ') - success = False - main_node = [i for i in graph if i["@id"] == "./"][0] - - # iteratively go through graph - for partI in main_node['hasPart']: - success = processNode(graph, partI['@id']) and success - if fileName not in logJson: - logJson[fileName] = {LABEL:success} - else: - logJson[fileName] = logJson[fileName] | {LABEL:success} - - # count occurances of all keys - counts = {} - for node in graph: - if node['@id'] in ['./',METADATA_FILE]: - continue - for key in node.keys(): - if key in counts: - counts[key] += 1 - else: - counts[key] = 1 - - view = [ (v,k) for k,v in counts.items() ] - view.sort(reverse=True) - if OUTPUT_COUNTS: - print('===== Counts (* unspecified)') - for v,k in view: - prefix = ' ' if k in KNOWN_KEYS else ' * ' - print(f'{prefix}{k:15}: {v}') - print('\n\nSuccess:', success) - json.dump(logJson, open('tests/logging.json', 'w')) - assert success #if this fails on your local test, great. It is a summary such that github actions report correctly + generalizedTest(checkParamMetadataJson, 'params_metadata_json') diff --git a/tests/test_02_schema.py b/tests/test_02_schema.py index 170ae21..13c00f2 100644 --- a/tests/test_02_schema.py +++ b/tests/test_02_schema.py @@ -3,16 +3,9 @@ Validate if rocrate of pypi can open and parse it. This is a test if we follow general ro-crate guidelines. https://pypi.org/project/rocrate/ """ -import os -import json import unittest -from pathlib import Path -from zipfile import ZIP_DEFLATED -from zipfile import ZipFile -from jsonschema import Draft202012Validator - -LABEL = 'schema' -METADATA_FILE = 'ro-crate-metadata.json' +from checks import checkSchema +from test_00_pypi_rocrate import generalizedTest class Test_1(unittest.TestCase): """ @@ -22,32 +15,4 @@ def test_main(self): """ main function """ - # log-file - if Path('tests/logging.json').exists(): - logJson = json.load(open('tests/logging.json')) - else: - logJson = {} - - schema = json.load(open('tests/schema.json', 'r', encoding='utf-8')) - validator = Draft202012Validator(schema=schema) - validator.check_schema(schema=schema) - success = True - for root, _, files in os.walk(".", topdown=False): - if 'SKIP_CI' in files: - continue - for name in files: - successFile = True - if not name.endswith('.eln'): - continue - fileName = os.path.join(root, name) - print(f'\nTest 02: {root}{name}') - with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: - metadataJsonFile = [i for i in elnFile.namelist() if i.endswith(METADATA_FILE)][0] - metadataContent = json.loads(elnFile.read(metadataJsonFile)) - for error in sorted(validator.iter_errors(metadataContent), key=str): - print(f'- {error.message}') - success = False - successFile = False - logJson[fileName] = logJson.get(fileName,{}) | {LABEL: successFile} - json.dump(logJson, open('tests/logging.json', 'w')) - assert success + generalizedTest(checkSchema, 'schema') diff --git a/tests/test_03_validator.py b/tests/test_03_validator.py index e849c23..f2ad431 100644 --- a/tests/test_03_validator.py +++ b/tests/test_03_validator.py @@ -3,18 +3,9 @@ Validate if rocrate of pypi can open and parse it. This is a test if we follow general ro-crate guidelines. https://pypi.org/project/rocrate/ """ -import os -import json -import logging -import tempfile import unittest -from pathlib import Path -from zipfile import ZIP_DEFLATED -from zipfile import ZipFile -from rocrate_validator import services, models - -LABEL = 'validator' -METADATA_FILE = 'ro-crate-metadata.json' +from checks import checkValidator +from test_00_pypi_rocrate import generalizedTest class Test_1(unittest.TestCase): """ @@ -24,50 +15,4 @@ def test_main(self): """ main function """ - # log-file - logging.getLogger('rdflib').setLevel(logging.ERROR) # get rid of verbose rdflib warnings, we concentrate on validator directly - if Path('tests/logging.json').exists(): - logJson = json.load(open('tests/logging.json')) - else: - logJson = {} - - for root, _, files in os.walk(".", topdown=False): - for name in files: - if 'SKIP_CI' in files: - continue - if not name.endswith('.eln'): - continue - fileName = os.path.join(root, name) - print(f'\n\nTest 03: {fileName}') - with ZipFile(fileName, 'r', compression=ZIP_DEFLATED) as elnFile: - dirpath = Path(tempfile.mkdtemp())/Path(fileName).parent.name - dirpath.mkdir(parents=True, exist_ok=True) - elnFile.extractall(dirpath) - rocrate_dir= [i for i in dirpath.iterdir() if i.is_dir()][0] - - # short stop-gap measure because the validator currently does not support the latest ro-crate version - with open(rocrate_dir/METADATA_FILE, 'r', encoding='utf-8') as f: - metadata = f.read() - metadata = metadata.replace('conformsTo":{"@id":"https:\/\/w3id.org\/ro\/crate\/1.2"}', - 'conformsTo":{"@id":"https:\/\/w3id.org\/ro\/crate\/1.1"}') - with open(rocrate_dir/METADATA_FILE, 'w', encoding='utf-8') as f: - f.write(metadata) - - # start validation - settings = services.ValidationSettings( - rocrate_uri=rocrate_dir, - profile_identifier='ro-crate-1.1', - requirement_severity=models.Severity.REQUIRED, - ) - result = services.validate(settings) - if not result.has_issues(): - success = True - else: - print(f'{fileName} is not valid') - for issue in result.get_issues(): - print(f"Detected issue of severity {issue.severity.name} with check \"{issue.check.identifier}\": {issue.message}") - success = False - - logJson[fileName] = logJson.get(fileName,{}) | {LABEL: success} - json.dump(logJson, open('tests/logging.json', 'w')) - assert success + generalizedTest(checkValidator, 'validator')