Open Knowledge Validator

tools/ok_validator/README.md

@@ -0,0 +1,2 @@
+## Open Knowledge Validator
+

tools/ok_validator/pyproject.toml

@@ -0,0 +1,31 @@
+[tool.poetry]
+name="okvalidator"
+version="0.1.0"
+description="An Open Knowledge validator"
+authors = [
+    "Helpful Engineering",
+]
+readme="README.md"
+
+repository = "https://github.com/helpfulengineering/project-data-platform/tools/ok_validator"
+
+classifiers = [
+  "Intended Audience :: Developers",
+  "Topic :: Software Development :: Open Knowledge :: Supply Chain",
+  "License :: OSI Approved :: MIT License",
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.7",
+  "Programming Language :: Python :: 3.8",
+  "Programming Language :: Python :: 3.9",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3 :: Only",
+]
+
+
+[tool.poetry.dependencies]
+python = "^3.7"
+pyyaml = "6.0"
+
+[tool.poetry.group.dev.dependencies]
+pytest = "^7.2.1"

tools/ok_validator/src/okh.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+from .validate import OKValidator
+
+
+class OKH(BaseModel, extra="allow"):
+    title: str
+    bom: list[str]
+
+
+class OKHValidator(OKValidator):
+    def __init__(self):
+        super().__init__(OKH)

tools/ok_validator/src/validate.py

@@ -0,0 +1,51 @@
+"""Open knowledge validator"""
+import json
+from pathlib import Path
+from typing import Type, TypeVar, Union
+from pydantic import BaseModel
+from pydantic_yaml import parse_yaml_file_as, parse_yaml_raw_as
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class OKValidator:
+    def __init__(self, model_type: Type[T]):
+        self.model_type = model_type
+
+    def validate(self, src: Union[str, Path, dict], raise_exception=False) -> bool:
+        """
+        Validate the YAML source, which can be a file path, YAML content string,
+        Path object, or YAML dictionary.
+
+        Args:
+            src: The string path or parsed yaml contents represented as dictionary.
+            raise_exception: Raises an exception if true otherwise returns a boolean result.
+        """
+        if not isinstance(src, Union[str, dict, Path]):
+            return self.return_value_or_error(
+                ValueError(
+                    "`src` should be one of the following: a string path, "
+                    "a Path object, or Yaml dict",
+                ),
+                raise_exception,
+            )
+        try:
+            if isinstance(src, dict):
+                parse_yaml_raw_as(self.model_type, json.dumps(src))
+            else:
+                parse_yaml_file_as(self.model_type, src)
+            return True
+        except Exception as err:
+            return self.return_value_or_error(err, raise_exception)
+
+    @staticmethod
+    def return_value_or_error(error: Exception, raise_exception: bool = False):
+        """Return a bool or raise an exception.
+
+        Args:
+            error: Exception to raise.
+            raise_exception: If set to true, the provided exception will be raised.
+        """
+        if not raise_exception:
+            return False
+        raise error

tools/ok_validator/tests/conftest.py

@@ -0,0 +1,41 @@
+import pytest
+
+from tools.ok_validator.src.okh import OKHValidator
+
+
+@pytest.fixture
+def okh_string():
+    return """
+title: mock okh title.
+description: mock description for okh.
+bom: 
+  - mock bom field.
+        """
+
+
+@pytest.fixture
+def okh_yaml_file(tmp_path, okh_string):
+    okh_file = tmp_path / "okh.yaml"
+    okh_file.touch()
+    okh_file.write_text(okh_string)
+
+    return okh_file
+
+
+@pytest.fixture
+def okh_dict():
+    return {
+        "title": "mock okh title.",
+        "description": "mock description for okh.",
+        "bom": ["mock bom field."],
+    }
+
+
+@pytest.fixture
+def okh_dict_partial():
+    return {"title": "mock okh title."}
+
+
+@pytest.fixture
+def okh_validator():
+    return OKHValidator()

tools/ok_validator/tests/test_okh.py

@@ -0,0 +1,25 @@
+"""Unit tests for okh validator"""
+import pytest
+
+
+@pytest.mark.parametrize("src_fixture", ["okh_yaml_file", "okh_dict"])
+def test_okh_validate(okh_validator, src_fixture, request):
+    """Test that the OKH validator return the correct boolean
+    on validate.
+
+    Args:
+        okh_validator: OKHValidator instance.
+        src_fixture: src dict or file containing all required OKH fields.
+    """
+    assert okh_validator.validate(request.getfixturevalue(src_fixture))
+
+
+def test_okh_validate_partial_fields(okh_validator, okh_dict_partial):
+    """Test that the OKH validator return the correct boolean
+    on validate.
+
+    Args:
+        okh_validator: OKHValidator instance.
+        okh_dict_partial: Yaml dict with missing required OKH fields.
+    """
+    assert not okh_validator.validate(okh_dict_partial)