diff --git a/.gitignore b/.gitignore index 87e0b75..1dda6bb 100644 --- a/.gitignore +++ b/.gitignore @@ -171,4 +171,7 @@ cython_debug/ # pytest coverage pytest.xml -pytest-coverage.txt \ No newline at end of file +pytest-coverage.txt + +# artifacts +artifacts \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json index f2223bb..b0a4109 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -57,7 +57,10 @@ "cwd": "${workspaceFolder}", "program": "${workspaceFolder}/.venv/bin/bam_masterdata", "justMyCode": false, - "args": ["fill_masterdata"] + "args": [ + "fill_masterdata", + // "--url=https://devel.datastore.bam.de/" + ] }, { "name": "BM export-to-json", diff --git a/bam_masterdata/cli/cli.py b/bam_masterdata/cli/cli.py index 5591693..b7741ff 100644 --- a/bam_masterdata/cli/cli.py +++ b/bam_masterdata/cli/cli.py @@ -1,10 +1,23 @@ import os +import subprocess import time from pathlib import Path import click +from decouple import config as environ +from openpyxl import Workbook +from bam_masterdata.cli.entities_to_excel import entities_to_excel +from bam_masterdata.cli.entities_to_json import entities_to_json from bam_masterdata.cli.fill_masterdata import MasterdataCodeGenerator +from bam_masterdata.logger import logger +from bam_masterdata.utils import ( + delete_and_create_dir, + import_module, + listdir_py_modules, +) + +DATAMODEL_DIR = os.path.join(".", "bam_masterdata", "datamodel") @click.group(help="Entry point to run `bam_masterdata` CLI commands.") @@ -14,19 +27,31 @@ def cli(): @cli.command( name="fill_masterdata", - help="Fill the masterdata from the openBIS instance specified in the `.env` in the bam_masterdata/datamodel/ subfolder.", + help="Fill the masterdata from the openBIS instance and stores it in the bam_masterdata/datamodel/ modules.", +) +@click.option( + "--url", + type=str, + required=False, + help=""" + (Optional) The URL of the openBIS instance from which to extract the data model. If not defined, + it is using the value of the `OPENBIS_URL` environment variable. + """, ) -def fill_masterdata(): +def fill_masterdata(url): start_time = time.time() # ! this takes a lot of time loading all the entities in Openbis - generator = MasterdataCodeGenerator() + # Use the URL if provided, otherwise fall back to defaults + if not url: + url = environ("OPENBIS_URL") + click.echo(f"Using the openBIS instance: {url}\n") + generator = MasterdataCodeGenerator(url=url) # Add each module to the `bam_masterdata/datamodel` directory - output_dir = os.path.join(".", "bam_masterdata", "datamodel") for module_name in ["property", "collection", "dataset", "object", "vocabulary"]: module_start_time = time.perf_counter() # more precise time measurement - output_file = Path(os.path.join(output_dir, f"{module_name}_types.py")) + output_file = Path(os.path.join(DATAMODEL_DIR, f"{module_name}_types.py")) # Get the method from `MasterdataCodeGenerator` code = getattr(generator, f"generate_{module_name}_types")() @@ -40,12 +65,76 @@ def fill_masterdata(): elapsed_time = time.time() - start_time click.echo(f"Generated all types in {elapsed_time:.2f} seconds\n\n") - # ! this could be automated in the CLI - click.echo( - "Don't forget to apply ruff at the end after generating the files by doing:\n" + try: + # Run ruff check + click.echo("Running `ruff check .`...") + subprocess.run(["ruff", "check", "."], check=True) + + # Run ruff format + click.echo("Running `ruff format .`...") + subprocess.run(["ruff", "format", "."], check=True) + except subprocess.CalledProcessError as e: + click.echo(f"Error during ruff execution: {e}", err=True) + else: + click.echo("Ruff checks and formatting completed successfully!") + + +@cli.command( + name="export_to_json", + help="Export entities to JSON files to the `./artifacts/` folder.", +) +def export_to_json(): + # Get the directories from the Python modules and the export directory for the static artifacts + export_dir = os.path.join(".", "artifacts") + + # Delete and create the export directory + delete_and_create_dir(directory_path=export_dir, logger=logger) + + # Get the Python modules to process the datamodel + py_modules = listdir_py_modules(directory_path=DATAMODEL_DIR, logger=logger) + + # Process each module using the `to_json` method of each entity + for module_path in py_modules: + entities_to_json(module_path=module_path, export_dir=export_dir, logger=logger) + + click.echo(f"All entity artifacts have been generated and saved to {export_dir}") + + +@cli.command( + name="export_to_excel", + help="Export entities to an Excel file in the path `./artifacts/masterdata.xlsx`.", +) +def export_to_excel(): + # Get the Python modules to process the datamodel + py_modules = listdir_py_modules(directory_path=DATAMODEL_DIR, logger=logger) + + # Load the definitions module classes + definitions_module = import_module( + module_path="./bam_masterdata/metadata/definitions.py" ) - click.echo(" ruff check .\n") - click.echo(" ruff format .\n") + + # Process the modules and save the entities to the openBIS masterdata Excel file + masterdata_file = os.path.join(".", "artifacts", "masterdata.xlsx") + wb = Workbook() + for i, module_path in enumerate(py_modules): + if i == 0: + ws = wb.active + else: + ws = wb.create_sheet() + ws.title = ( + os.path.basename(module_path) + .capitalize() + .replace(".py", "") + .replace("_", " ") + ) + entities_to_excel( + worksheet=ws, + module_path=module_path, + definitions_module=definitions_module, + ) + wb.save(masterdata_file) + + click.echo(f"All masterdata have been generated and saved to {masterdata_file}") if __name__ == "__main__": diff --git a/bam_masterdata/cli/entities_to_excel.py b/bam_masterdata/cli/entities_to_excel.py new file mode 100644 index 0000000..36a176e --- /dev/null +++ b/bam_masterdata/cli/entities_to_excel.py @@ -0,0 +1,99 @@ +import inspect +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from openpyxl.worksheet.worksheet import Worksheet + +from bam_masterdata.utils import import_module + + +def entities_to_excel( + worksheet: "Worksheet", + module_path: str, + definitions_module: Any, +) -> None: + """ + Export entities to the Excel file. The Python modules are imported using the function `import_module`, + and their contents are inspected (using `inspect`) to find the classes in the datamodel containing + `defs` and with a `to_json` method defined. Each row is then appended to the `worksheet`. + + Args: + worksheet (Worksheet): The worksheet to append the entities. + module_path (str): Path to the Python module file. + definitions_module (Any): The module containing the definitions of the entities. This is used + to match the header definitions of the entities. + """ + def_members = inspect.getmembers(definitions_module, inspect.isclass) + module = import_module(module_path=module_path) + + # Special case of `PropertyTypeDef` in `property_types.py` + if "property_types.py" in module_path: + for name, obj in inspect.getmembers(module): + if name.startswith("_") or name == "PropertyTypeDef": + continue + + # Entity title + worksheet.append([obj.excel_name]) + + # Entity header definitions and values + worksheet.append(obj.excel_headers) + row = [] + for f_set in obj.model_fields.keys(): + if f_set == "data_type": + val = obj.data_type.value + else: + val = getattr(obj, f_set) + row.append(val) + worksheet.append(row) + worksheet.append([""]) # empty row after entity definitions + return None + + # All other datamodel modules + for _, obj in inspect.getmembers(module, inspect.isclass): + # Ensure the class has the `to_json` method + if not hasattr(obj, "defs") or not callable(getattr(obj, "to_json")): + continue + + obj_instance = obj() + + # Entity title + obj_definitions = obj_instance.defs + worksheet.append([obj_definitions.excel_name]) + + # Entity header definitions and values + for def_name, def_cls in def_members: + if def_name == obj_definitions.name: + break + worksheet.append(obj_definitions.excel_headers) + header_values = [ + getattr(obj_definitions, f_set) for f_set in def_cls.model_fields.keys() + ] + worksheet.append(header_values) + + # Properties assignment for ObjectType, DatasetType, and CollectionType + if obj_instance.cls_name in ["ObjectType", "DatasetType", "CollectionType"]: + if not obj_instance.properties: + continue + worksheet.append(obj_instance.properties[0].excel_headers) + for prop in obj_instance.properties: + row = [] + for f_set in prop.model_fields.keys(): + if f_set == "data_type": + val = prop.data_type.value + else: + val = getattr(prop, f_set) + row.append(val) + worksheet.append(row) + # Terms assignment for VocabularyType + elif obj_instance.cls_name == "VocabularyType": + if not obj_instance.terms: + continue + worksheet.append(obj_instance.terms[0].excel_headers) + for term in obj_instance.terms: + worksheet.append( + getattr(term, f_set) for f_set in term.model_fields.keys() + ) + + # ? do the PropertyTypeDef need to be exported to Excel? + + worksheet.append([""]) # empty row after entity definitions diff --git a/bam_masterdata/cli/entities_to_json.py b/bam_masterdata/cli/entities_to_json.py new file mode 100644 index 0000000..f7e87c6 --- /dev/null +++ b/bam_masterdata/cli/entities_to_json.py @@ -0,0 +1,67 @@ +import inspect +import json +import os +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from structlog._config import BoundLoggerLazyProxy + +import click + +from bam_masterdata.utils import delete_and_create_dir, import_module + + +def entities_to_json( + module_path: str, export_dir: str, logger: "BoundLoggerLazyProxy" +) -> None: + """ + Export entities to JSON files. The Python modules are imported using the function `import_module`, + and their contents are inspected (using `inspect`) to find the classes in the datamodel containing + `defs` and with a `to_json` method defined. + + Args: + module_path (str): Path to the Python module file. + export_dir (str): Path to the directory where the JSON files will be saved. + logger (BoundLoggerLazyProxy): The logger to log messages. + """ + module = import_module(module_path=module_path) + # export to specific subfolders for each type of entity (each module) + module_export_dir = os.path.join( + export_dir, os.path.basename(module_path).replace(".py", "") + ) + delete_and_create_dir(directory_path=module_export_dir, logger=logger) + + # Special case of `PropertyTypeDef` in `property_types.py` + if "property_types.py" in module_path: + for name, obj in inspect.getmembers(module): + if name.startswith("_") or name == "PropertyTypeDef": + continue + try: + json_data = json.dumps(obj.model_dump(), indent=2) + output_file = os.path.join(module_export_dir, f"{obj.code}.json") + with open(output_file, "w", encoding="utf-8") as f: + f.write(json_data) + + click.echo(f"Saved JSON for class {name} to {output_file}") + except Exception as err: + click.echo(f"Failed to process class {name} in {module_path}: {err}") + return None + + # All other datamodel modules + for name, obj in inspect.getmembers(module, inspect.isclass): + # Ensure the class has the `to_json` method + if not hasattr(obj, "defs") or not callable(getattr(obj, "to_json")): + continue + + try: + # Instantiate the class and call the method + json_data = obj().to_json(indent=2) + + # Write JSON data to file + output_file = os.path.join(module_export_dir, f"{obj.defs.code}.json") + with open(output_file, "w", encoding="utf-8") as f: + f.write(json_data) + + click.echo(f"Saved JSON for class {name} to {output_file}") + except Exception as err: + click.echo(f"Failed to process class {name} in {module_path}: {err}") diff --git a/bam_masterdata/cli/fill_masterdata.py b/bam_masterdata/cli/fill_masterdata.py index 32bf1a2..793506a 100644 --- a/bam_masterdata/cli/fill_masterdata.py +++ b/bam_masterdata/cli/fill_masterdata.py @@ -3,6 +3,7 @@ import click from bam_masterdata.openbis import OpenbisEntities +from bam_masterdata.openbis.login import environ class MasterdataCodeGenerator: @@ -11,14 +12,14 @@ class MasterdataCodeGenerator: openBIS instance. """ - def __init__(self): + def __init__(self, url: str = ""): start_time = time.time() # * This part takes some time due to the loading of all entities from Openbis - self.properties = OpenbisEntities().get_property_dict() - self.collections = OpenbisEntities().get_collection_dict() - self.datasets = OpenbisEntities().get_dataset_dict() - self.objects = OpenbisEntities().get_object_dict() - self.vocabularies = OpenbisEntities().get_vocabulary_dict() + self.properties = OpenbisEntities(url=url).get_property_dict() + self.collections = OpenbisEntities(url=url).get_collection_dict() + self.datasets = OpenbisEntities(url=url).get_dataset_dict() + self.objects = OpenbisEntities(url=url).get_object_dict() + self.vocabularies = OpenbisEntities(url=url).get_vocabulary_dict() elapsed_time = time.time() - start_time click.echo( f"Loaded OpenBIS entities in `MasterdataCodeGenerator` initialization {elapsed_time:.2f} seconds\n" @@ -103,7 +104,7 @@ def add_properties( # ! patching dataType=SAMPLE instead of OBJECT if prop_data.get("dataType", "") == "SAMPLE": prop_data["dataType"] = "OBJECT" - lines.append(f" data_type=\"{prop_data.get('dataType', '')}\",") + lines.append(f' data_type="{prop_data.get("dataType", "")}",') property_label = (prop_data.get("label") or "").replace("\n", "\\n") lines.append(f' property_label="{property_label}",') description = ( @@ -163,7 +164,7 @@ def generate_property_types(self) -> str: # ! patching dataType=SAMPLE instead of OBJECT if data.get("dataType", "") == "SAMPLE": data["dataType"] = "OBJECT" - lines.append(f" data_type=\"{data.get('dataType', '')}\",") + lines.append(f' data_type="{data.get("dataType", "")}",') property_label = ( (data.get("label") or "").replace('"', '\\"').replace("\n", "\\n") ) @@ -222,7 +223,7 @@ def generate_collection_types(self) -> str: lines.append(f' description="""{description}""",') if data.get("validationPlugin") != "": lines.append( - f" validation_script=\"{data.get('validationPlugin')}\"," + f' validation_script="{data.get("validationPlugin")}",' ) lines.append(" )") lines.append("") @@ -327,7 +328,7 @@ def generate_object_types(self) -> str: ) lines.append(f' description="""{description}""",') lines.append( - f" generated_code_prefix=\"{data.get('generatedCodePrefix', '')}\"," + f' generated_code_prefix="{data.get("generatedCodePrefix", "")}",' ) lines.append(" )") lines.append("") diff --git a/bam_masterdata/metadata/definitions.py b/bam_masterdata/metadata/definitions.py index a9da1e4..5d63e8d 100644 --- a/bam_masterdata/metadata/definitions.py +++ b/bam_masterdata/metadata/definitions.py @@ -79,7 +79,7 @@ class EntityDef(BaseModel): @field_validator("code") @classmethod def validate_code(cls, value: str) -> str: - if not value or not re.match(r"^[A-Z_\$\.]+$", value): + if not value or not re.match(r"^[\w_\$\.\-\+]+$", value): raise ValueError( "`code` must follow the rules specified in the description: 1) Must be uppercase, " "2) separated by underscores, 3) start with a dollar sign if native to openBIS, " @@ -92,6 +92,31 @@ def validate_code(cls, value: str) -> str: def strip_description(cls, value: str) -> str: return value.strip() + @property + def name(self) -> str: + return self.__class__.__name__ + + @property + def excel_name(self) -> str: + """ + Returns the name of the entity in a format suitable for the openBIS Excel file. + """ + name_map = { + "CollectionTypeDef": "EXPERIMENT_TYPE", + "DataSetTypeDef": "DATASET_TYPE", + "ObjectTypeDef": "SAMPLE_TYPE", + "PropertyTypeDef": "PROPERTY_TYPE", + "VocabularyTypeDef": "VOCABULARY_TYPE", + } + return name_map.get(self.name) + + @property + def excel_headers(self) -> list[str]: + """ + Returns the headers for the entity in a format suitable for the openBIS Excel file. + """ + return [k.capitalize().replace("_", " ") for k in self.model_fields.keys()] + class BaseObjectTypeDef(EntityDef): """ diff --git a/bam_masterdata/metadata/entities.py b/bam_masterdata/metadata/entities.py index 416a3ce..7970b6e 100644 --- a/bam_masterdata/metadata/entities.py +++ b/bam_masterdata/metadata/entities.py @@ -52,6 +52,14 @@ def to_dict(self) -> dict: dump_json = self.to_json() return json.loads(dump_json) + @property + def cls_name(self) -> str: + """ + Returns the entity name of the class as a string to speed up checks. This is a property + to be overwritten by each of the abstract entity types. + """ + return self.__class__.__name__ + class ObjectType(BaseEntity): """ @@ -101,6 +109,13 @@ def model_validator_after_init(cls, data: Any) -> Any: return data + @property + def cls_name(self) -> str: + """ + Returns the entity name of the class as a string. + """ + return "ObjectType" + class VocabularyType(BaseEntity): """ @@ -140,14 +155,27 @@ def model_validator_after_init(cls, data: Any) -> Any: return data - -class PropertyType(BaseEntity): - pass + @property + def cls_name(self) -> str: + """ + Returns the entity name of the class as a string. + """ + return "VocabularyType" class CollectionType(ObjectType): - pass + @property + def cls_name(self) -> str: + """ + Returns the entity name of the class as a string. + """ + return "CollectionType" class DatasetType(ObjectType): - pass + @property + def cls_name(self) -> str: + """ + Returns the entity name of the class as a string. + """ + return "DatasetType" diff --git a/bam_masterdata/openbis/get_entities.py b/bam_masterdata/openbis/get_entities.py index 1ef76eb..339067d 100644 --- a/bam_masterdata/openbis/get_entities.py +++ b/bam_masterdata/openbis/get_entities.py @@ -1,4 +1,4 @@ -from bam_masterdata.openbis.login import ologin +from bam_masterdata.openbis.login import environ, ologin class OpenbisEntities: @@ -7,8 +7,8 @@ class OpenbisEntities: Python modules of `bam_masterdata/datamodel/`. """ - def __init__(self): - self.openbis = ologin() + def __init__(self, url: str = ""): + self.openbis = ologin(url=url) def _get_formatted_dict(self, entity_name: str): # entity_name is property_types, collection_types, dataset_types, object_types, or vocabularies @@ -29,6 +29,7 @@ def _assign_properties(self, entity_name: str, formatted_dict: dict) -> None: # Create a dictionary of properties using the correct permId properties = {} for entry in assignments_dict: + # ! This has changed and now permId does not exist on the property assignments!! property_perm_id = ( entry.get("permId", {}).get("propertyTypeId", {}).get("permId") ) @@ -59,6 +60,7 @@ def _assign_properties(self, entity_name: str, formatted_dict: dict) -> None: "plugin": entry.get("plugin", ""), } + # ! This has changed and now permId, label, and description do not exist on the property assignments!! for prop in assignments: properties[prop.permId].update( { diff --git a/bam_masterdata/openbis/login.py b/bam_masterdata/openbis/login.py index 34cb38e..a443c27 100644 --- a/bam_masterdata/openbis/login.py +++ b/bam_masterdata/openbis/login.py @@ -3,14 +3,16 @@ # Connect to openBIS -def ologin() -> Openbis: +def ologin(url: str = "") -> Openbis: """ Connect to openBIS using the credentials stored in the environment variables. + Args: + url (str): The URL of the openBIS instance. Defaults to the value of the `OPENBIS_URL` environment variable. + Returns: Openbis: Openbis object for the specific openBIS instance defined in `URL`. """ - url = environ("OPENBIS_URL") o = Openbis(url) o.login(environ("OPENBIS_USERNAME"), environ("OPENBIS_PASSWORD"), save_token=True) return o diff --git a/bam_masterdata/utils/__init__.py b/bam_masterdata/utils/__init__.py new file mode 100644 index 0000000..d7dec42 --- /dev/null +++ b/bam_masterdata/utils/__init__.py @@ -0,0 +1 @@ +from .utils import delete_and_create_dir, import_module, listdir_py_modules diff --git a/bam_masterdata/utils/utils.py b/bam_masterdata/utils/utils.py new file mode 100644 index 0000000..54b5d2e --- /dev/null +++ b/bam_masterdata/utils/utils.py @@ -0,0 +1,80 @@ +import glob +import importlib.util +import os +import shutil +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from structlog._config import BoundLoggerLazyProxy + + +def delete_and_create_dir(directory_path: str, logger: "BoundLoggerLazyProxy") -> None: + """ + Deletes the directory at `directory_path` and creates a new one in the same path. + + Args: + directory_path (str): The directory path to delete and create the folder. + logger (BoundLoggerLazyProxy): The logger to log messages. + """ + if not directory_path: + logger.warning( + "The `directory_path` is empty. Please, provide a proper input to the function." + ) + return None + + if os.path.exists(directory_path): + try: + shutil.rmtree(directory_path) # ! careful with this line + except PermissionError: + logger.error( + f"Permission denied to delete the directory at {directory_path}." + ) + return None + os.makedirs(directory_path) + + +def listdir_py_modules( + directory_path: str, logger: "BoundLoggerLazyProxy" +) -> list[str]: + """ + Recursively goes through the `directory_path` and returns a list of all .py files that do not start with '_'. + + Args: + directory_path (str): The directory path to search through. + logger (BoundLoggerLazyProxy): The logger to log messages. + + Returns: + list[str]: A list of all .py files that do not start with '_' + """ + if not directory_path: + logger.warning( + "The `directory_path` is empty. Please, provide a proper input to the function." + ) + return [] + + # Use glob to find all .py files recursively + files = glob.glob(os.path.join(directory_path, "**", "*.py"), recursive=True) + if not files: + logger.info("No Python files found in the directory.") + return [] + + # Filter out files that start with '_' + # ! sorted in order to avoid using with OS sorting differently + return sorted([f for f in files if not os.path.basename(f).startswith("_")]) + + +def import_module(module_path: str) -> Any: + """ + Dynamically imports a module from the given file path. + + Args: + module_path (str): Path to the Python module file. + + Returns: + module: Imported module object. + """ + module_name = os.path.splitext(os.path.basename(module_path))[0] + spec = importlib.util.spec_from_file_location(module_name, module_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module diff --git a/pyproject.toml b/pyproject.toml index ed7667c..2f7b5b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ maintainers = [ ] license = { file = "LICENSE" } dependencies = [ - "pybis<=1.36.3", + "pybis~=1.37", "openpyxl", "click", "pydantic", diff --git a/tests/cli/test_entities_to_excel.py b/tests/cli/test_entities_to_excel.py new file mode 100644 index 0000000..8928d47 --- /dev/null +++ b/tests/cli/test_entities_to_excel.py @@ -0,0 +1,81 @@ +import os + +import pytest +from openpyxl import Workbook + +from bam_masterdata.cli.entities_to_excel import entities_to_excel +from bam_masterdata.utils import import_module + + +@pytest.mark.parametrize( + "module_name, entity_name, entity_headers", + [ + ( + "collection_types", + "EXPERIMENT_TYPE", + ["Code", "Description", "Validation script"], + ), + # ('dataset_types', 'DATASET_TYPE', ['Code', 'Description', 'Validation script']), # ! this module does not have classes yet + ( + "object_types", + "SAMPLE_TYPE", + [ + "Code", + "Description", + "Validation script", + "Generated code prefix", + "Auto generated codes", + ], + ), + ( + "property_types", + "PROPERTY_TYPE", + [ + "Code", + "Description", + "Property label", + "Data type", + "Vocabulary code", + "Metadata", + "Dynamic script", + ], + ), + ( + "vocabulary_types", + "VOCABULARY_TYPE", + ["Code", "Description", "Url template"], + ), + ], +) +def test_entities_to_excel( + module_name: str, entity_name: str, entity_headers: list[str] +): + """Test the `entities_to_excel` function.""" + definitions_module = import_module( + module_path="./bam_masterdata/metadata/definitions.py" + ) + + # Get the Python modules to process the datamodel + module_path = os.path.join("./bam_masterdata/datamodel", f"{module_name}.py") + wb = Workbook() + ws = wb.active + ws.title = ( + os.path.basename(module_path).capitalize().replace(".py", "").replace("_", " ") + ) + + entities_to_excel( + worksheet=ws, + module_path=module_path, + definitions_module=definitions_module, + ) + + assert len(wb.worksheets) == 1 + # Header for type of entity + assert ws.cell(row=1, column=1).value == entity_name + # Headers for definitions + col = [] + for c in range(1, 101): + if not ws.cell(row=2, column=c).value: + break + col.append(ws.cell(row=2, column=c).value) + assert col == entity_headers diff --git a/tests/cli/test_entities_to_json.py b/tests/cli/test_entities_to_json.py new file mode 100644 index 0000000..6dcb66d --- /dev/null +++ b/tests/cli/test_entities_to_json.py @@ -0,0 +1,42 @@ +import json +import os +import shutil + +import pytest + +from bam_masterdata.cli.entities_to_json import entities_to_json +from bam_masterdata.logger import logger + + +@pytest.mark.parametrize( + "module_name", + [ + ("collection_types"), + # ('dataset_types', False), # ! this module does not have classes yet + ("object_types"), + ("property_types"), + ("vocabulary_types"), + ], +) +def test_entities_to_json(module_name: str): + """Test the `entities_to_json` function.""" + export_dir = "./tests/data/tmp/" + module_path = os.path.join("./bam_masterdata/datamodel", f"{module_name}.py") + + entities_to_json(module_path=module_path, export_dir=export_dir, logger=logger) + + module_export_dir = os.path.join(export_dir, module_name) + assert os.path.exists(export_dir) + assert len(os.listdir(module_export_dir)) > 0 + assert [".json" in f for f in os.listdir(module_export_dir)] + + for file in os.listdir(module_export_dir): + with open(os.path.join(module_export_dir, file)) as f: + data = json.load(f) + # making sure the data stored in json files is correct + if module_name == "property_types": + assert data["code"] == file.replace(".json", "") + else: + assert data["defs"]["code"] == file.replace(".json", "") + + shutil.rmtree(export_dir) # ! careful with this line diff --git a/tests/utils/test_utils.py b/tests/utils/test_utils.py new file mode 100644 index 0000000..0ea7a5f --- /dev/null +++ b/tests/utils/test_utils.py @@ -0,0 +1,96 @@ +import inspect +import os +import shutil + +import pytest + +from bam_masterdata.logger import logger +from bam_masterdata.utils import ( + delete_and_create_dir, + import_module, + listdir_py_modules, +) + + +@pytest.mark.parametrize( + "directory_path, dir_exists", + [ + # `directory_path` is empty + ("", False), + # `directory_path` does not exist and it is created + ("tests/data/tmp/", True), + ], +) +def test_delete_and_create_dir( + cleared_log_storage: list, directory_path: str, dir_exists: bool +): + """Tests the `delete_and_delete_dir` function.""" + delete_and_create_dir(directory_path=directory_path, logger=logger) + assert dir_exists == os.path.exists(directory_path) + if dir_exists: + shutil.rmtree(directory_path) # ! careful with this line + else: + assert len(cleared_log_storage) == 1 + assert cleared_log_storage[0]["level"] == "warning" + assert "directory_path" in cleared_log_storage[0]["event"] + + +@pytest.mark.parametrize( + "directory_path, listdir, log_message, log_message_level", + [ + # `directory_path` is empty + ( + "", + [], + "The `directory_path` is empty. Please, provide a proper input to the function.", + "warning", + ), + # No Python files found in the directory + ("./tests/data", [], "No Python files found in the directory.", "info"), + # Python files found in the directory + ( + "./tests/utils", + [ + "./tests/utils/test_utils.py", + ], + None, + None, + ), + ], +) +def test_listdir_py_modules( + cleared_log_storage: list, + directory_path: str, + listdir: list[str], + log_message: str, + log_message_level: str, +): + """Tests the `listdir_py_modules` function.""" + result = listdir_py_modules(directory_path=directory_path, logger=logger) + if not listdir: + assert cleared_log_storage[0]["event"] == log_message + assert cleared_log_storage[0]["level"] == log_message_level + # when testing locally and with Github actions the order of the files is different --> `result` is sorted, so we also sort `listdir` + assert result == sorted(listdir) + + +@pytest.mark.skip( + reason="Very annoying to test this function, as any module we can use to be tested will change a lot in the future." +) +def test_import_module(): + """Tests the `import_module` function.""" + # testing only the possitive results + module = import_module("./bam_data_store/utils/utils.py") + assert [f[0] for f in inspect.getmembers(module, inspect.ismodule)] == [ + "glob", + "importlib", + "os", + "shutil", + "sys", + ] + assert [f[0] for f in inspect.getmembers(module, inspect.isclass)] == [] + assert [f[0] for f in inspect.getmembers(module, inspect.isfunction)] == [ + "delete_and_create_dir", + "import_module", + "listdir_py_modules", + ]