Mapper unit testing

docker-compose.yml

@@ -1,5 +1,6 @@
 services:
   python:
+    container_name: rikolti
     build:
       context: ./
       dockerfile: ./Dockerfile.dev

metadata_mapper/mappers/ucsd_blacklight/ucsd_blacklight_mapper.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 
-class UcsdBlacklightMapper(Record):
+class UcsdBlacklightRecord(Record):
     BASE_URL = "https://library.ucsd.edu/dc/object/"
 
     BASE_ARK = "ark:/20775/"
@@ -357,7 +357,7 @@ def identifier_content_match(validation_def: dict, rikolti_value: Any,
 
 
 class UcsdBlacklightVernacular(Vernacular):
-    record_cls = UcsdBlacklightMapper
+    record_cls = UcsdBlacklightRecord
     validator = UcsdBlacklightValidator
 
     def parse(self, api_response) -> list:

metadata_mapper/requirements.txt

@@ -4,3 +4,7 @@ lxml
 sickle
 MarkupSafe
 python-dotenv
+faker
+pytest
+pytest-assume
+requests-mock

metadata_mapper/test/conftest.py

@@ -0,0 +1,3 @@
+def pytest_addoption(parser):
+    parser.addoption("--mappers", action="store", default=None)
+    parser.addoption("--mapper", action="store", default=None)

metadata_mapper/test/helpers/base_helper.py

@@ -0,0 +1,133 @@
+import importlib
+import os
+
+from datetime import datetime
+from faker import Faker
+from random import randint
+from typing import Any, Union
+
+
+class BaseTestHelper:
+    """
+    Generates fake data for use in mapper unit tests.
+
+    By default, this class uses DEFAULT_SCHEMA as its schema definition.
+    Any subclass can define SCHEMA, and it will be merged into
+    DEFAULT_SCHEMA to generate an appropriate schema for any given mapper.
+
+    If a field requires special logic to generate, define a method named
+    "generate_{field_name}" where {field_name} equals the key in SCHEMA
+    that you want to generate the value for.
+    """
+
+    # SCHEMAs will be merged together in order to generate
+    # the final fixture schema.
+    SCHEMA = {}
+
+    @classmethod
+    def for_mapper(cls, module_parts: list[str]) -> type["BaseTestHelper"]:
+        helper_path = None
+        module_len = len(module_parts)
+        while module_len and not helper_path:
+            helper_path = (
+                "metadata_mapper/test/helpers/"
+                + "/".join(module_parts[:module_len]).replace("_mapper", "")
+                + "_helper.py"
+            )
+            if not os.path.exists(helper_path):
+                helper_path = None
+                module_len = module_len - 1
+
+        if helper_path:
+            helper_module_parts = [
+                p.replace("_mapper", "_helper") for p in module_parts
+            ]
+            helper_class_name = (
+                "".join(
+                    [
+                        word.title()
+                        for word in module_parts[-1].replace("_mapper", "").split("_")
+                    ]
+                )
+                + "TestHelper"
+            )
+            helper_module = importlib.import_module(
+                f".helpers.{'.'.join(helper_module_parts)}",
+                package="rikolti.metadata_mapper.test",
+            )
+            return getattr(helper_module, helper_class_name)
+        else:
+            return None
+
+    def __init__(self, request_mock=None):
+        self.request_mock = request_mock
+        self.faker = Faker()
+        self.collection_id = self.faker.pyint()
+        self.static = {}
+        self.setup_mocks()
+
+    def setup_mocks(self):
+        pass
+
+    def instantiate_record(self, record_class) -> type["Record"]:
+        fixture = self.generate_fixture()
+        instance = record_class(self.collection_id, fixture)
+        self.prepare_record(instance)
+        return instance
+
+    def prepare_record(self, record) -> None:
+        record.legacy_couch_db_id = "asdf--123123"
+
+    def generate_fixture(self, schema_index: int = 0) -> dict[str, Any]:
+        """
+        Generates a test data fixture.
+        """
+        schema = self.merged_schema(schema_index)
+
+        return {
+            key: self.generate_value_for(key, type) for (key, type) in schema.items()
+        }
+
+    def merged_schema(self, schema_index: int = 0) -> dict[str, Any]:
+        inheritance_chain = list(reversed(type(self).__mro__))
+        superschemas = [
+            super(cls, self).SCHEMA
+            for cls in inheritance_chain
+            if hasattr(super(cls, self), "SCHEMA")
+        ]
+
+        schema = {}
+        for superschema in superschemas:
+            schema = {**schema, **superschema}
+
+        return {**schema, **self.SCHEMA}
+
+    def generate_value_for(
+        self,
+        field_name: str = None,
+        expected_type: Union[type, list, str] = str,
+    ) -> Any:
+        if isinstance(expected_type, str):
+            return getattr(self, expected_type)()
+        elif isinstance(expected_type, type):
+            return self.generate_value_of_type(expected_type)
+        elif isinstance(expected_type, list):
+            return [self.generate_value_for(item) for item in expected_type]
+
+    def generate_value_of_type(self, type: type) -> Any:
+        if type == str:
+            return self.faker.pystr()
+        elif type == datetime:
+            return self.faker.date()
+
+    # Helper methods
+
+    def splittable_string(self) -> str:
+        """Generate a string with semicolons to be split on"""
+        return ";".join([self.faker.pystr() for _ in range(0, randint(1, 3))])
+
+    def list_of_splittable_strings(self) -> list[str]:
+        """
+        Generate content to be split and flattened by mapper#split_and_flatten.
+        """
+        return [self.splittable_string() for _ in range(0, randint(1, 3))]

metadata_mapper/test/helpers/oai/cca_vault_helper.py

@@ -0,0 +1,8 @@
+from .oai_helper import OaiTestHelper
+
+class CcaVaultTestHelper(OaiTestHelper):
+
+  SCHEMA = {
+    "language": "list_of_splittable_strings",
+    "source": [str]
+  }

metadata_mapper/test/helpers/oai/content_dm/contentdm_helper.py

@@ -0,0 +1,38 @@
+import re
+
+from ..oai_helper import OaiTestHelper
+from .....mappers.oai.content_dm.contentdm_mapper import ContentdmRecord
+
+
+class ContentdmTestHelper(OaiTestHelper):
+    SCHEMA = {
+        "contributor": "list_of_splittable_strings",
+        "coverage": "splittable_string",
+        "creator": "list_of_splittable_strings",
+        "identifier": "contentdm_identifier",
+        "language": "list_of_splittable_strings",
+        "spatial": "splittable_string",
+        "subject": "list_of_splittable_strings",
+        "type": "list_of_splittable_strings",
+    }
+
+    def setup_mocks(self):
+        matcher = re.compile('utils/ajaxhelper')
+        response = {
+          "imageinfo": {
+              "height": self.faker.pyint(),
+              "width": self.faker.pyint()
+          }
+        }
+        self.request_mock.register_uri('GET', matcher, json=response)
+
+    def contentdm_identifier(self):
+        # See ContentdmMapper#get_identifier_parts for required formated
+        value = f"http://{self.faker.domain_name()}/" + \
+                f"{ContentdmRecord.identifier_match}/" + \
+                (f"{self.faker.pystr()}/") + \
+                f"{self.collection_id}/" + \
+                f"{self.faker.pystr()}/" + \
+                str(self.faker.pyint())
+
+        return [value]

metadata_mapper/test/helpers/oai/content_dm/csudh_helper.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+from random import randint
+from typing import Any
+
+from .contentdm_helper import ContentdmTestHelper
+
+
+class CsudhTestHelper(ContentdmTestHelper):
+    SCHEMA = {
+        "bibliographicCitation": str,
+        "title": "csudh_title",
+    }
+
+    def csudh_title(self):
+        return [f"csudh-{self.faker.pystr()}", self.faker.pystr()]

metadata_mapper/test/helpers/oai/oai_helper.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from random import randint
+
+from ..base_helper import BaseTestHelper
+
+class OaiTestHelper(BaseTestHelper):
+
+  SCHEMA = {
+    "contributor": str,
+    "creator": str,
+    "date": [datetime] * randint(1, 9),
+    "description": [str] * randint(1, 3),
+    "extent": str,
+    "format": [str] * randint(1, 2),
+    "id": str,
+    "identifier": [str] * randint(1, 2),
+    "provenance": str,
+    "publisher": str,
+    "relation": [str] * randint(1, 14),
+    "rights": [str] * randint(1, 2),
+    "spatial": [str] * randint(1, 2),
+    "subject": str,
+    "temporal": str,
+    "title": str,
+    "type": str
+  }
+
+  def prepare_record(self, record) -> None:
+    record.select_id(["id"])

metadata_mapper/test/test_mapper.py

@@ -0,0 +1,115 @@
+import importlib
+import os
+import pytest
+import re
+import requests_mock
+import traceback
+
+from textwrap import dedent
+
+from .helpers.base_helper import BaseTestHelper
+from ..mappers.mapper import Record
+
+
+class TestMapper:
+    DEFAULT_TEST_METHOD_NAME = "_test_generic_mapper"
+
+    def find_mappers_to_test(self, start_path="metadata_mapper/mappers"):
+        ret = {}
+
+        for dir in os.scandir(start_path):
+            if dir.is_dir():
+                ret = {**ret, **self.find_mappers_to_test(dir.path)}
+            elif dir.is_file() and dir.name.endswith("_mapper.py"):
+                path_regex_result = re.search("([\\w\\/]+?_mapper).py", dir.path)
+                if path_regex_result:
+                    full_mapper_path = (
+                        path_regex_result[1].replace("/", ".").lstrip(".")
+                    )
+                    module_parts = [
+                        path
+                        for path in full_mapper_path.split(".")
+                        if path not in ["metadata_mapper", "mappers"]
+                    ]
+                    mapper_name = module_parts[-1]
+                    mapper_path = ".".join(module_parts)
+
+                    if f"test_{mapper_name}" in locals().keys():
+                        ret[mapper_path] = getattr(self, f"test_{mapper_name}")
+                    else:
+                        ret[mapper_path] = getattr(self, self.DEFAULT_TEST_METHOD_NAME)
+
+        return ret
+
+    def get_record(self, module_parts, module) -> Record:
+        mapper_name = module_parts[-1].replace("_mapper", "")
+        class_name = f"{self.camelize(mapper_name)}Record"
+        return getattr(module, class_name)
+
+    def camelize(self, words: str) -> str:
+        return "".join([word.title() for word in words.split("_")])
+
+    def _test_generic_mapper(self, record_class, helper):
+        try:
+            instance = helper.instantiate_record(record_class)
+            try:
+                instance.to_UCLDC()
+            except Exception as exc:
+                pytest.assume(
+                    False,
+                    dedent(f"""\n**{type(instance).__name__}** raised error '{exc}'
+                    at time of mapping.\n Here's the backtrace:\n
+                    {traceback.format_exc()}"""),
+                )
+        except Exception as exc:
+            pytest.assume(
+                False,
+                dedent(f"""\n**{record_class.__name__}** raised error '{exc}'
+                at time of initialization.\n Here's the backtrace:\n
+                {traceback.format_exc()}"""),
+            )
+
+    # Test methods (invoked by pytest)
+
+    # This will loop through all mappers that don't have explicit test methods and
+    # run them with default data
+    def test_mappers(self, pytestconfig):
+        with requests_mock.Mocker() as r_mock:
+            default_test_method = getattr(self, self.DEFAULT_TEST_METHOD_NAME)
+
+            mapper_filter = [
+                mapper
+                for mapper in re.split(
+                    r"[,;]",
+                    pytestconfig.getoption("mappers")
+                    or pytestconfig.getoption("mapper")
+                    or "",
+                )
+                if mapper
+            ] or None
+
+            mappers = [
+                mapper
+                for mapper, method in self.find_mappers_to_test().items()
+                if method == default_test_method
+            ]
+
+            for mapper in mappers:
+                module_parts = mapper.split(".")
+                if (
+                    mapper_filter
+                    and module_parts[-1] not in mapper_filter
+                    and module_parts[-1].replace("_mapper", "") not in mapper_filter
+                ):
+                    continue
+
+                module = importlib.import_module(
+                    f".mappers.{'.'.join(module_parts)}",
+                    package="rikolti.metadata_mapper",
+                )
+                helper_class = BaseTestHelper.for_mapper(module_parts)
+
+                if helper_class:
+                    helper = helper_class(r_mock)
+                    record_class = self.get_record(module_parts, module)
+                    default_test_method(record_class, helper)