Skip to content
Snippets Groups Projects
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
test_unit.py 12.14 KiB
# -*- coding: 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
#
"""Created on 14.12.2015.

@author: tf
"""
from __future__ import unicode_literals
import caosdb as h
from nose.tools import assert_true, assert_equal  # @UnresolvedImport


def setup_module():
    try:
        h.execute_query("FIND Simple*").delete()
    except BaseException:
        pass


def teardown_module():
    try:
        h.execute_query("FIND Simple*").delete()
    except BaseException:
        pass


def test_km_1():
    # insert with base unit and search for base unit and derived unit

    try:

        p = h.Property(
            name="SimpleDoubleProperty",
            datatype=h.DOUBLE,
            unit="m")
        p.insert()
        assert_true(p.is_valid())

        rt = h.RecordType(name="SimpleRecordType").add_property(p, value=3140)
        rt.insert()

        rt2 = h.execute_query("FIND SimpleRecordType", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)
        assert_equal(rt2.get_property("SimpleDoubleProperty").unit, "m")
        assert_equal(rt2.get_property("SimpleDoubleProperty").value, 3140.0)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3140.0m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3140m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3.14km'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_km_2():
    # insert with derived unit and search for base unit and derived unit

    try:

        p = h.Property(
            name="SimpleDoubleProperty",
            datatype="DOUBLE",
            unit="km")
        p.insert()
        assert_true(p.is_valid())

        rt = h.RecordType(name="SimpleRecordType").add_property(p, value=3.14)
        rt.insert()

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3.14km'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3140.0m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='3140m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_celsius_1():
    # insert with base unit and search for base unit and derived unit

    try:
        p = h.Property(
            name="SimpleDoubleProperty",
            datatype=h.DOUBLE,
            unit="K")
        p.insert()
        assert_true(p.is_valid())

        rt = h.RecordType(name="SimpleRecordType").add_property(p, value=0)
        rt.insert()

        rt2 = h.execute_query("FIND SimpleRecordType", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)
        assert_equal(rt2.get_property("SimpleDoubleProperty").unit, "K")
        assert_equal(rt2.get_property("SimpleDoubleProperty").value, 0.0)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='0.0K'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='0K'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='-273.15°C'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_celsius_2():
    # insert with derived unit and search for base unit and derived unit

    try:
        p = h.Property(
            name="SimpleDoubleProperty",
            datatype="DOUBLE",
            unit="°C")
        p.insert()
        assert_true(p.is_valid())

        rt = h.RecordType(name="SimpleRecordType").add_property(p, value=0)
        rt.insert()

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='0.0°C'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='0°C'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleDoubleProperty='273.15K'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_int():
    # insert with base unit and search for base unit and derived unit

    try:
        p = h.Property(
            name="SimpleIntegerProperty",
            datatype=h.INTEGER,
            unit="m")
        p.insert()
        assert_true(p.is_valid())

        rt = h.RecordType(name="SimpleRecordType").add_property(p, value=3140)
        rt.insert()

        rt2 = h.execute_query("FIND SimpleRecordType", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)
        assert_equal(rt2.get_property("SimpleIntegerProperty").unit, "m")
        assert_equal(rt2.get_property("SimpleIntegerProperty").value, 3140)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3140m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3140.0m'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3.14km'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='314000cm'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='314000.0cm'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_unknown_unit():
    # insert unknown unit

    try:
        p = h.Property(
            name="SimpleIntegerProperty",
            datatype=h.INTEGER,
            unit="my_unit")
        p.insert()
        assert_true(p.is_valid())
        assert_equal(len(p.get_warnings()), 1)
        assert_equal(
            p.get_warnings()[0].description,
            "Unknown unit. Values with this unit cannot be converted to other units when used in search queries.")

        rt = h.RecordType(
            name="SimpleRecordType").add_property(
            p, value="3140")
        rt.insert()

        rt2 = h.execute_query("FIND SimpleRecordType", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)
        assert_equal(
            rt2.get_property("SimpleIntegerProperty").unit,
            "my_unit")
        assert_equal(rt2.get_property("SimpleIntegerProperty").value, 3140)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3140'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3140my_unit'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

        rt2 = h.execute_query(
            "FIND SimpleRecordType.SimpleIntegerProperty='3140.0my_unit'", True)
        assert_true(rt2.is_valid())
        assert_true(rt2.id, rt.id)

    finally:
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass


def test_greatest():
    # insert unknown unit

    try:
        try:
            h.execute_query("FIND Simple*").delete()
        except BaseException:
            pass
        p = h.Property(
            name="SimpleDoubleProperty",
            datatype=h.DOUBLE,
            unit="m")
        p.insert()

        rt = h.RecordType(name="SimpleRecordType").add_property(p)
        rt.insert()

        rec1 = h.Record(name="SimpleRecord1").add_parent(
            rt).add_property(p, unit="km", value="1").insert()
        rec2 = h.Record(name="SimpleRecord2").add_parent(
            rt).add_property(p, unit="km", value="2").insert()
        rec3 = h.Record(
            name="SimpleRecord3").add_parent(rt).add_property(
            p, unit="m", value="200").insert()

        assert_true(rec1.is_valid())
        assert_true(rec2.is_valid())
        assert_true(rec3.is_valid())

        c = h.execute_query(
            "FIND SimpleRecord* WHICH HAS THE GREATEST SimpleDoubleProperty",
            unique=True)
        assert_equal(c.id, rec2.id)

        c = h.execute_query(
            "FIND SimpleRecord* WHICH HAS THE SMALLEST SimpleDoubleProperty",
            unique=True)
        assert_equal(c.id, rec3.id)

        rec4 = h.Record(
            name="SimpleRecord4").add_parent(rt).add_property(
            p, unit="bla", value="150").insert()
        assert_true(rec4.is_valid())

        c = h.execute_query(
            "FIND SimpleRecord* WHICH HAS THE GREATEST SimpleDoubleProperty")
        assert_equal(c[0].id, rec3.id)
        assert_equal(
            c.get_warnings()[0].description,
            "The filter POV(SimpleDoubleProperty,NULL,NULL) with the aggregate function 'max' could not match the values against each other with their units. The values had different base units. Only their numric value had been taken into account.")

        c = h.execute_query(
            "FIND SimpleRecord* WHICH HAS THE SMALLEST SimpleDoubleProperty",
            unique=True)
        assert_equal(c.id, rec1.id)
        assert_equal(
            c.get_warnings()[0].description,
            "The filter POV(SimpleDoubleProperty,NULL,NULL) with the aggregate function 'min' could not match the values against each other with their units. The values had different base units. Only their numric value had been taken into account.")

    finally:
        try:
            rec4.delete()
        except BaseException:
            pass
        try:
            rec3.delete()
        except BaseException:
            pass
        try:
            rec2.delete()
        except BaseException:
            pass
        try:
            rec1.delete()
        except BaseException:
            pass
        try:
            rt.delete()
        except BaseException:
            pass
        try:
            p.delete()
        except BaseException:
            pass