test14/dataladmetadatamodel/tests/test_datasettree.py

import subprocess
import tempfile
import unittest
from unittest.mock import patch

from dataladmetadatamodel.datasettree import DatasetTree
from dataladmetadatamodel.mapper.gitmapper.gitblobcache import hash_blob
from dataladmetadatamodel.metadatapath import MetadataPath
from dataladmetadatamodel.metadatarootrecord import MetadataRootRecord
from dataladmetadatamodel.tests.utils import (
    get_location,
    get_uuid,
    assert_dataset_trees_equal,
    create_dataset_tree
)


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

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


uuid_0 = get_uuid(0)


class TestDatasetTree(unittest.TestCase):
    def test_add_metadata(self):

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

        mrr = MetadataRootRecord(uuid_0, "00112233", None, None)
        dataset_tree = create_dataset_tree(
            dataset_tree_paths=paths,
            initial_mrr=mrr)

        returned_entries = tuple(dataset_tree.dataset_paths)

        returned_paths = [entry[0] for entry in returned_entries]
        self.assertEqual(sorted(paths), sorted(returned_paths))

        for entry in returned_entries:
            self.assertEqual(entry[1], mrr)

    def test_root_node(self):

        dataset_tree = DatasetTree()

        mrr = MetadataRootRecord(uuid_0, "00112233", None, None)
        dataset_tree.add_dataset(MetadataPath(""), mrr)
        self.assertEqual(dataset_tree.get_metadata_root_record(MetadataPath("")), mrr)

        returned_entries = tuple(dataset_tree.dataset_paths)
        self.assertEqual(len(returned_entries), 1)

        self.assertEqual(returned_entries[0][0], MetadataPath(""))
        self.assertEqual(returned_entries[0][1], mrr)


class TestReferenceCreation(unittest.TestCase):
    def test_object_reference_creation(self):

        dataset_tree = create_dataset_tree(dataset_test_paths)
        with \
                patch("dataladmetadatamodel.mapper.gitmapper."
                      "metadatamapper.git_save_str") as save_str, \
                patch("dataladmetadatamodel.mapper.gitmapper."
                      "mtreenodemapper.git_save_tree_node") as save_tree_node, \
                patch("dataladmetadatamodel.mapper.gitmapper."
                      "metadatarootrecordmapper.git_save_json") as save_json, \
                patch("dataladmetadatamodel.mapper.gitmapper."
                      "gitblobcache.git_save_file_list") as file_list_save, \
                patch("dataladmetadatamodel.mtreeproxy."
                      "add_tree_reference") as add_tree_ref:

            save_str.return_value = get_location(1)
            save_tree_node.return_value = get_location(2)
            save_json.return_value = get_location(3)
            file_list_save.side_effect = lambda r, l: [
                hash_blob(open(e, "rb").read())
                for e in l
            ]

            dataset_tree.write_out("/tmp/t1")

            # We expect one call for the dataset-tree itself
            # and one call for each file-tree, one of which
            # is anchored at each dataset path
            self.assertEqual(
                add_tree_ref.call_count,
                1 + len(dataset_test_paths)
            )


class TestDeepCopy(unittest.TestCase):

    def test_copy_from_memory(self):
        with \
                tempfile.TemporaryDirectory() as original_dir, \
                tempfile.TemporaryDirectory() as copy_dir:

            subprocess.run(["git", "init", original_dir])
            subprocess.run(["git", "init", copy_dir])

            dataset_tree = create_dataset_tree(dataset_test_paths,
                                               file_test_paths)

            dataset_tree_copy = dataset_tree.deepcopy("git", copy_dir)
            dataset_tree_copy.read_in()

            assert_dataset_trees_equal(
                self,
                dataset_tree,
                dataset_tree_copy,
                True)

    def test_copy_from_backend(self):
        with \
                tempfile.TemporaryDirectory() as original_dir, \
                tempfile.TemporaryDirectory() as copy_dir:

            subprocess.run(["git", "init", original_dir])
            subprocess.run(["git", "init", copy_dir])

            dataset_tree = create_dataset_tree(dataset_test_paths,
                                               file_test_paths)

            dataset_tree.write_out(original_dir)

            dataset_tree_copy = dataset_tree.deepcopy("git", copy_dir)
            dataset_tree_copy.read_in()

            assert_dataset_trees_equal(
                self,
                dataset_tree,
                dataset_tree_copy,
                True)


class TestSubTreeManipulation(unittest.TestCase):

    def get_mrr(self, n: int = 0) -> MetadataRootRecord:
        return MetadataRootRecord(uuid_0, f"00112233-{n}", None, None)

    def test_subtree_adding(self):
        mrr_1 = self.get_mrr(1)
        mrr_2 = self.get_mrr(2)

        tree = DatasetTree()
        tree.add_dataset(MetadataPath("a/b/c"), mrr_1)

        subtree = DatasetTree()
        subtree.add_dataset(MetadataPath("d/e/f"), mrr_2)

        tree.add_subtree(subtree, MetadataPath("a/x"))

        mrr = tree.get_metadata_root_record(MetadataPath("a/b/c"))
        self.assertEqual(mrr, mrr_1)

        mrr = tree.get_metadata_root_record(MetadataPath("a/x/d/e/f"))
        self.assertEqual(mrr, mrr_2)

    def test_subtree_adding_with_conversion(self):
        mrr_1 = self.get_mrr()
        mrr_2 = self.get_mrr()

        tree = DatasetTree()
        tree.add_dataset(MetadataPath("a/b/c"), mrr_1)

        subtree = DatasetTree()
        subtree.add_dataset(MetadataPath("e/f"), mrr_2)

        tree.add_subtree(subtree, MetadataPath("a/b/c/d"))

        node = tree.get_metadata_root_record(MetadataPath("a/b/c"))
        self.assertEqual(node, mrr_1)

        node = tree.get_metadata_root_record(MetadataPath("a/b/c/d/e/f"))
        self.assertEqual(node, mrr_2)

    def test_subtree_adding_on_existing_path(self):
        tree = DatasetTree()
        tree.add_dataset(MetadataPath("a/b/c/d"), self.get_mrr())

        subtree = DatasetTree()
        subtree.add_dataset(MetadataPath("e/f"), self.get_mrr())

        self.assertRaises(
            ValueError,
            tree.add_subtree,
            subtree, MetadataPath("a/b/c/d"))

    def test_subtree_deletion(self):
        mrr_1 = self.get_mrr()
        mrr_2 = self.get_mrr()

        tree = DatasetTree()
        tree.add_dataset(MetadataPath("a/b/c"), mrr_1)
        tree.add_dataset(MetadataPath("a/b/c/d/e/f"), mrr_2)

        self.assertIsNotNone(tree.get_metadata_root_record(MetadataPath("a/b/c/d/e/f")))
        tree.delete_subtree(MetadataPath("a/b/c/d/e/f"))
        self.assertIsNone(tree.get_metadata_root_record(MetadataPath("a/b/c/d/e/f")))
        self.assertIsNotNone(tree.mtree.get_object_at_path(MetadataPath("a/b/c/d/e")))

        tree.delete_subtree(MetadataPath("a/b/c"))
        self.assertIsNotNone(tree.mtree.get_object_at_path(MetadataPath("a/b")))
        self.assertIsNone(tree.get_metadata_root_record(MetadataPath("a/b/c")))
        self.assertIsNone(tree.get_metadata_root_record(MetadataPath("a/b/c/d")))
        self.assertIsNone(tree.get_metadata_root_record(MetadataPath("a/b/c/d/e")))


if __name__ == '__main__':
    unittest.main()