test14/dataladmetadatamodel/tests/utils.py

import time
import unittest
from typing import (
    cast,
    Any,
    List,
    Optional,
    Union,
)
from uuid import UUID

from dataladmetadatamodel.datasettree import DatasetTree
from dataladmetadatamodel.filetree import FileTree
from dataladmetadatamodel.mappableobject import MappableObject
from dataladmetadatamodel.metadata import Metadata
from dataladmetadatamodel.metadatapath import MetadataPath
from dataladmetadatamodel.metadatarootrecord import MetadataRootRecord
from dataladmetadatamodel.mtreeproxy import MTreeProxy
from dataladmetadatamodel.uuidset import UUIDSet
from dataladmetadatamodel.versionlist import VersionList
from dataladmetadatamodel.mapper.reference import Reference


uuid_pattern = "9900{:04x}00000000000000000000{:04x}"
version_pattern = "ea{:04x}000000000000000000000000000000{:04x}"


default_file_tree_paths = [
    MetadataPath("a/b/c"),
    MetadataPath("a/b/a"),
    MetadataPath("b"),
    MetadataPath("c/d/e"),
    MetadataPath("a/x")]

default_dataset_tree_paths = [
    MetadataPath("d1"),
    MetadataPath("d1/d1.1"),
    MetadataPath("d2"),
    MetadataPath("d2/d2.1/d2.1.1"),
    MetadataPath("d3/d3.1")]


def get_uuid(n: int) -> UUID:
    return UUID(uuid_pattern.format(n, n))


def get_location(n: int) -> str:
    return version_pattern.format(n, n)


def get_version(n: int) -> str:
    return get_location(n)


def assert_equal(test_case: unittest.TestCase,
                 object_a: Any,
                 object_b: Any,
                 _: bool,
                 ):
    test_case.assertEqual(object_a, object_b)


def assert_metadata_equal(test_case: unittest.TestCase,
                          object_a: Any,
                          object_b: Any,
                          _: bool,
                          ):

    test_case.assertEqual(object_a, object_b)


def assert_mrr_equal(test_case: unittest.TestCase,
                     a_mrr: MetadataRootRecord,
                     b_mrr: MetadataRootRecord,
                     a_purge_unsafe: bool,
                     ):

    test_case.assertEqual(a_mrr.dataset_identifier, b_mrr.dataset_identifier)
    test_case.assertEqual(a_mrr.dataset_version, b_mrr.dataset_version)
    assert_mappable_objects_equal(
        test_case,
        a_mrr.dataset_level_metadata,
        b_mrr.dataset_level_metadata,
        a_purge_unsafe,
        assert_metadata_equal)

    assert_mappable_objects_equal(
        test_case,
        a_mrr._file_tree,
        b_mrr._file_tree,
        a_purge_unsafe,
        assert_file_trees_equal)


def assert_mappable_objects_equal(test_case: unittest.TestCase,
                                  a_object: Union[MappableObject, MTreeProxy],
                                  b_object: Union[MappableObject, MTreeProxy],
                                  a_purge_unsafe: bool,
                                  equality_asserter):

    a_object = a_object.read_in()
    b_object = b_object.read_in()

    equality_asserter(test_case, a_object, b_object, a_purge_unsafe)

    if a_purge_unsafe is False:
        a_object.purge()
    b_object.purge()


def assert_file_trees_equal(test_case: unittest.TestCase,
                            a: FileTree,
                            b: FileTree,
                            unsafe: bool
                            ):

    a_entries = list(a.get_paths_recursive())
    b_entries = list(b.get_paths_recursive())

    # Compare paths
    test_case.assertListEqual(
        sorted(list(map(lambda x: x[0], a_entries))),
        sorted(list(map(lambda x: x[0], b_entries))))

    # Compare metadata elements
    for a_object, b_object in zip(
            map(lambda x: x[1], a_entries),
            map(lambda x: x[1], b_entries)):

        # TODO: proper metadata comparison instead of assertEqual
        assert_mappable_objects_equal(
            test_case,
            a_object,
            b_object,
            unsafe,
            assert_equal)


def assert_dataset_trees_equal(test_case: unittest.TestCase,
                               a: DatasetTree,
                               b: DatasetTree,
                               a_purge_unsafe: bool
                               ):

    a_entries = list(a.dataset_paths)
    b_entries = list(b.dataset_paths)

    # Compare paths
    test_case.assertListEqual(
        list(map(lambda x: x[0], a_entries)),
        list(map(lambda x: x[0], b_entries)))

    # Compare metadata_root_records
    for a_mrr, b_mrr in zip(
            map(lambda x: x[1], a_entries),
            map(lambda x: x[1], b_entries)):

        assert_mrr_equal(
            test_case,
            a_mrr.read_in(),
            b_mrr.read_in(),
            a_purge_unsafe)


def assert_mrrs_equal(test_case: unittest.TestCase,
                      a: MetadataRootRecord,
                      b: MetadataRootRecord,
                      unsafe: bool
                      ):

    # Compare dataset level metadata
    assert_mappable_objects_equal(
        test_case,
        a.dataset_level_metadata,
        b.dataset_level_metadata,
        unsafe,
        assert_equal)

    # Compare file trees
    assert_mappable_objects_equal(
        test_case,
        a._file_tree,
        b._file_tree,
        unsafe,
        assert_file_trees_equal
    )

    # Compare the remaining metadata
    test_case.assertEqual(a.dataset_identifier, b.dataset_identifier)
    test_case.assertEqual(a.dataset_version, b.dataset_version)


def assert_version_lists_equal(test_case: unittest.TestCase,
                               a: VersionList,
                               b: VersionList,
                               unsafe: bool
                               ):

    a_entries = [a.get_versioned_element(pdv) for pdv in a.versions()]
    b_entries = [b.get_versioned_element(pdv) for pdv in b.versions()]

    for a_entry, b_entry in zip(a_entries, b_entries):
        test_case.assertEqual(a_entry[0], b_entry[0])
        test_case.assertEqual(a_entry[1], b_entry[1])
        test_case.assertIsInstance(a_entry[2], (DatasetTree, MetadataRootRecord))
        test_case.assertIsInstance(b_entry[2], (DatasetTree, MetadataRootRecord))
        if isinstance(a_entry[2], DatasetTree):
            assert_dataset_trees_equal(
                test_case,
                cast(DatasetTree, a_entry[2]),
                cast(DatasetTree, b_entry[2]),
                unsafe)
        else:
            assert_mrr_equal(
                test_case,
                cast(MetadataRootRecord, a_entry[2]),
                cast(MetadataRootRecord, b_entry[2]),
                unsafe)


def assert_uuid_sets_equal(test_case: unittest.TestCase,
                           a_uuid_set: UUIDSet,
                           b_uuid_set: UUIDSet):

    for dataset_id in a_uuid_set.uuids():
        a_version_list = a_uuid_set.get_version_list(dataset_id)
        b_version_list = b_uuid_set.get_version_list(dataset_id)
        assert_version_lists_equal(test_case,
                                   a_version_list,
                                   b_version_list,
                                   True)


def create_file_tree(paths: Optional[List[MetadataPath]] = None) -> FileTree:

    paths = paths or default_file_tree_paths

    file_tree = FileTree()
    for path in paths:
        metadata = Metadata()
        file_tree.add_metadata(path, metadata)
    return file_tree


def create_file_tree_with_metadata(paths: List[MetadataPath],
                                   metadata: List[Metadata]) -> FileTree:

    assert len(paths) == len(metadata)

    file_tree = FileTree()
    for path, md in zip(paths, metadata):
        file_tree.add_metadata(path, md)

    return file_tree


def create_dataset_tree(dataset_tree_paths: Optional[List[MetadataPath]] = None,
                        file_tree_paths: Optional[List[MetadataPath]] = None,
                        initial_mrr: Optional[MetadataRootRecord] = None
                        ) -> DatasetTree:

    dataset_tree_paths = dataset_tree_paths or default_dataset_tree_paths
    file_tree_paths = file_tree_paths or default_file_tree_paths

    dataset_tree = DatasetTree()
    for index, path in enumerate(dataset_tree_paths):

        metadata = Metadata()
        file_tree = create_file_tree(file_tree_paths)

        if initial_mrr is None:
            mrr = MetadataRootRecord(
                get_uuid(index),
                get_version(index),
                metadata,
                file_tree)
        else:
            mrr = initial_mrr

        dataset_tree.add_dataset(path, mrr)

    return dataset_tree


class MMDummy:
    def __init__(self,
                 info: str = "",
                 is_mapped: bool = False,
                 copied_from: Optional["MMDummy"] = None):
        self.info = info or f"MMDummy created at {time.time()}"
        self.mapped = is_mapped
        self.copied_from = copied_from

    def deepcopy(self, *args, **kwargs):
        return MMDummy(self.info, False, self)

    def read_in(self, backend_type="git") -> Any:
        self.mapped = True
        return self

    def write_out(self,
                  destination: Optional[str] = None,
                  backend_type: str = "git",
                  force_write: bool = False) -> Reference:
        return Reference(
            "MMDummy",
            get_location(0x51))

    def purge(self, force: bool = False):
        self.mapped = False
        return