test_cfood.py

#!/usr/bin/env python
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header
import re
import unittest

import caosdb as db
from caosadvancedtools.cfood import (AbstractCFood, AbstractFileCFood, CMeal,
                                     assure_has_parent, assure_has_property,
                                     assure_object_is_in_list,
                                     get_entity_for_path)
from caosadvancedtools.crawler import FileCrawler
from caosadvancedtools.example_cfood import ExampleCFood

PATTERN = "h.*"


class ExampleCFoodMeal(AbstractFileCFood, CMeal):
    matching_groups = ["test"]

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # add the constructor of CMeal
        CMeal.__init__(self)

    @classmethod
    def match_item(cls, item):
        """ standard match_match, but returns False if a suitable cfood exists """

        if cls.has_suitable_cfood(item):
            return False

        return re.match(cls.get_re(), item) is not None

    def looking_for(self, crawled_file):
        """ standard looking_for, but returns True if the file matches all
        groups"""

        if self.belongs_to_meal(crawled_file):
            return True

        return super().looking_for(crawled_file)

    @staticmethod
    def get_re():
        return r"/(?P<test>[a-z]*)/"

    def create_identifiables(self):
        pass

    def update_identifiables(self):
        pass


class SimpleCFood(AbstractFileCFood):

    @staticmethod
    def get_re():
        return PATTERN


class DependendCFood(AbstractCFood):
    existing = []

    @classmethod
    def match_item(cls, item):
        if len(cls.existing) == 0:
            return True

    def __init__(self, *args,  **kwargs):
        super().__init__(*args,  **kwargs)
        DependendCFood.existing.append(self)

    def create_identifiables(self):
        pass

    def update_identifiables(self):
        pass


class CFoodReTest(unittest.TestCase):
    def test(self):
        self.assertEqual(SimpleCFood.get_re(), PATTERN)
        self.assertEqual(SimpleCFood._pattern, None)
        self.assertTrue(SimpleCFood.match_item("hallo"))
        # TODO the caching is of compiled re is disabled currently
        # self.assertIsNotNone(SimpleCFood._pattern)
        self.assertTrue(SimpleCFood.match_item("hallo"))
        self.assertFalse(SimpleCFood.match_item("allo"))


class InsertionTest(unittest.TestCase):
    def test_contained_in_list(self):
        entity_with_list = db.Record()
        prop_name = "list_prop"
        to_be_updated = []
        assure_object_is_in_list(1005, entity_with_list, prop_name,
                                 to_be_updated)
        assert to_be_updated[0] is entity_with_list
        assert (to_be_updated[0].get_property(prop_name).value[0] == 1005)
        to_be_updated = []
        assure_object_is_in_list(1005, entity_with_list, prop_name,
                                 to_be_updated)
        assert len(to_be_updated) == 0

    def test_parent(self):
        entity = db.Record()
        to_be_updated = []
        assure_has_parent(entity, "parent", to_be_updated)
        assert to_be_updated[0] is entity
        assert (to_be_updated[0].get_parents()[0].name == "parent")
        to_be_updated = []
        assure_has_parent(entity, "parent", to_be_updated)
        assert len(to_be_updated) == 0

    def test_has_property(self):
        """Test properties with string, int, float, and Boolean values"""
        entity = db.Record()
        to_be_updated = []
        int_name = "Test int"
        types_and_values = {
            int_name: ("INT", 5),
            "Test float": (db.DOUBLE, 3.14),
            "Test bool": (db.BOOLEAN, True),
            "Test string": (db.TEXT, "bla")
        }

        for name, ty_val in types_and_values.items():
            entity.add_property(name=name, datatype=ty_val[0],
                                value=ty_val[1])
            assure_has_property(entity=entity, name=name,
                                value=ty_val[1], to_be_updated=to_be_updated)
            assert len(to_be_updated) == 0
        new_int = 6
        assure_has_property(entity=entity, name=int_name,
                            value=new_int, to_be_updated=to_be_updated)
        assert to_be_updated[0] is entity


class DependendTest(unittest.TestCase):
    def test(self):
        self.assertTrue(DependendCFood.match_item(None))
        cf = DependendCFood(None)
        self.assertFalse(DependendCFood.match_item(None))


class ExampleTest(unittest.TestCase):
    def test(self):
        path = "/data/rabbit/2019-03-03/README.md"
        cf = ExampleCFood(crawled_path=path)
        self.assertIsNotNone(ExampleCFood.match_item(path))
        self.assertEqual(cf.match.group('species'), 'rabbit')
        self.assertEqual(cf.match.group('date'), '2019-03-03')


class MealTest(unittest.TestCase):
    def test(self):
        # path should match
        self.assertTrue(ExampleCFoodMeal.match_item("/this/file"))
        # create an instance
        c = ExampleCFoodMeal("/this/file")
        # same prefix should no longer match
        self.assertFalse(ExampleCFoodMeal.match_item("/this/other"))
        # but instance should be looking for this prefix
        self.assertTrue(c.looking_for("/this/other"))
        # class should still match other prefixes
        self.assertTrue(ExampleCFoodMeal.match_item("/that/file"))


class FileCacheTest(unittest.TestCase):
    def test(self):
        self.assertRaises(Exception, get_entity_for_path, "/lol")
        FileCrawler(cfood_types=[], files=[db.File(path="/lol")])
        get_entity_for_path("/lol")