test_utils.py

#!/usr/bin/env python
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2019 Henrik tom Wörden
#
# 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 logging
import unittest
from tempfile import NamedTemporaryFile

import caosdb as db
from caosadvancedtools.utils import (check_win_path, get_referenced_files,
                                     string_to_person)
from caosdb import RecordType, configure_connection, get_config
from caosdb.connection.mockup import MockUpResponse, MockUpServerConnection
from caosdb.exceptions import TransactionError


class ReferencesBaseTest(unittest.TestCase):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.entities = (
            '<Response><File name="test.npy" path="/some/path/test.npy'
            '" id="1234"/><Query string="find record" results="1">'
            '</Query></Response>')

    def setUp(self):
        conlogger = logging.getLogger("connection")
        conlogger.setLevel(level=logging.ERROR)
        autlogger = logging.getLogger("authentication")
        autlogger.setLevel(level=logging.ERROR)
        connection = configure_connection(
            url="unittests", username="testuser",
            password_method="plain",
            password="testpassword", timeout=200,
            implementation=MockUpServerConnection)

        connection._delegate_connection.resources.append(
            lambda **kwargs: MockUpResponse(200, {}, self.entities))

        self.logfile = NamedTemporaryFile()
        logger = logging.getLogger()
        logger.addHandler(logging.FileHandler(self.logfile.name))
        logger.setLevel(logging.DEBUG)

    def clear_log(self):
        with open(self.logfile.name, "w") as lf:
            lf.write("")

    def get_log(self):
        with open(self.logfile.name) as lf:
            log = lf.read()
        self.clear_log()

        return log

    def test_ref(self):
        self.clear_log()
        files = get_referenced_files("test.npy", prefix=None, filename=None,
                                     location=None)
        self.assertEqual(len(files), 1)
        self.assertEqual(files[0].path, "/some/path/test.npy")
        self.assertEqual(self.get_log(), "")


class ReferencesFailTest(ReferencesBaseTest):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.entities = (
            '<Response><File name="test.npy" path="/some/path/test.npy'
            '" id="1234"/><Query string="find record" results="1">'
            '</Query><Error code="0" description="An error occured during the'
            ' of this query. Maybe you use a wrong syntax?" /></Response>')

    def test_ref(self):
        self.clear_log()
        files = get_referenced_files("test.npy", prefix=None, filename=None,
                                     location=None)
        self.assertEqual(files, [])
        assert "does not allow a search" in self.get_log()


class ReferencesEmptyTest(ReferencesBaseTest):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.entities = (
            '<Response><Query string="find record" results="1">'
            '</Query></Response>')

    def test_ref(self):
        self.clear_log()
        files = get_referenced_files("test.npy", prefix=None, filename=None,
                                     location=None)
        self.assertEqual(files, [])
        assert "does not reference any " in self.get_log()


def is_dhornung(rec):
    if (rec.get_property("firstname").value == "Daniel"
            and rec.get_property("lastname").value == "Hornung"):

        return True

    return False


class PersonParserTest(unittest.TestCase):
    def test_simple(self):
        rec = string_to_person("Daniel Hornung")
        assert is_dhornung(rec)
        rec = string_to_person("Daniel Hornung (MPI)")
        assert is_dhornung(rec)
        rec = string_to_person("Henrik tom Wörden (MPI)")
        assert not is_dhornung(rec)


class PathTest(unittest.TestCase):
    def test_win(self):
        assert check_win_path(r"C:\hallo")
        assert check_win_path(r"\hallo")
        assert not check_win_path("/hallo")