test_affiliation.py

# 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
#
"""Created on 09.03.2017.

@author: tf
"""
import caosdb as db
import os
from nose.tools import nottest, assert_true, assert_raises, assert_equal, with_setup, assert_is_not_none  # @UnresolvedImport


def setup_module():
    f = open(file_path, "w")
    f.write("hello, world!")
    f.close()


def teardown_module():
    os.remove(file_path)


rec_name = "TestRecordParent"
recty_name = "TestRecordTypeParent"
file_name = "TestFileParent"
prop_name = "TestPropertyParent"
file_path = "testfile.dat"


def setup():
    try:
        db.execute_query("FIND Test*").delete()
    except Exception as e:
        print(e)
    db.RecordType(name=recty_name).insert()
    db.Record(name=rec_name).add_parent(name=recty_name).insert()
    db.File(name=file_name, file=file_path, path="testfile.dat").insert()
    db.Property(name=prop_name, datatype=db.TEXT).insert()


def teardown():
    db.execute_query("FIND Test*").delete()


def test_affiliation_there():
    par = db.RecordType(name="TestRT1")
    ch = db.Record(name="TestRec").add_parent(par)
    print(ch)
    assert_is_not_none(ch.get_parent("TestRT1"))
    assert_true(hasattr(ch.get_parent("TestRT1"), "affiliation"))


@with_setup(setup, teardown)
def test_rec_rec_is_parthood():
    par = db.Record(name="TestRecordChild").add_parent(
        name=rec_name).insert().get_parent(rec_name)
    assert_equal(par.affiliation, "PARTHOOD")


@with_setup(setup, teardown)
def test_rec_rt_is_instantiation():
    par = db.Record(name="TestRecordChild").add_parent(
        name=recty_name).insert().get_parent(recty_name)
    assert_equal(par.affiliation, "INSTANTIATION")


@with_setup(setup, teardown)
def test_rec_prop_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.Record(name="TestRecordChild").add_parent(name=prop_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_rec_file_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.Record(name="TestRecordChild").add_parent(name=file_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_rt_rec_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.RecordType(
            name="TestRecordTypeChild").add_parent(
            name=rec_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_rt_rt_is_subtyping():
    par = db.RecordType(name="TestRecordTypeChild").add_parent(
        name=recty_name).insert().get_parent(recty_name)
    assert_equal(par.affiliation, "SUBTYPING")


@with_setup(setup, teardown)
def test_rt_prop_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.RecordType(
            name="TestRecordTypeChild").add_parent(
            name=prop_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_rt_file_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.RecordType(
            name="TestRecordTypeChild").add_parent(
            name=file_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_prop_rec_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.Property(
            name="TestPropertyChild",
            datatype=db.TEXT).add_parent(
            name=rec_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_prop_rt_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.Property(
            name="TestPropertyChild",
            datatype=db.TEXT).add_parent(
            name=recty_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_prop_prop_is_subtyping():
    par = db.Property(
        name="TestPropertyChild",
        datatype=db.TEXT).add_parent(
        name=prop_name).insert().get_parent(prop_name)
    assert_equal(par.affiliation, "SUBTYPING")


@with_setup(setup, teardown)
def test_prop_file_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.Property(
            name="TestPropertyChild",
            datatype=db.TEXT).add_parent(
            name=file_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_file_rec_is_parthood():
    par = db.File(
        name="TestFileChild",
        file=file_path,
        path="testfilechild.dat").add_parent(
        name=rec_name).insert().get_parent(rec_name)
    assert_equal(par.affiliation, "PARTHOOD")


@with_setup(setup, teardown)
def test_file_rt_is_instantiation():
    par = db.File(
        name="TestFileChild",
        file=file_path,
        path="testfilechild.dat").add_parent(
        name=recty_name).insert().get_parent(recty_name)
    assert_equal(par.affiliation, "INSTANTIATION")


@with_setup(setup, teardown)
def test_file_prop_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.File(
            name="TestFileChild",
            file=file_path,
            path="testfilechild.dat").add_parent(
            name=prop_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@with_setup(setup, teardown)
def test_file_file_is_invalid():
    with assert_raises(db.EntityError) as cm:
        db.File(
            name="TestFileChild",
            file=file_path,
            path="testfilechild.dat").add_parent(
            name=file_name).insert()
    assert_equal(
        cm.exception.get_errors()[0].msg,
        "Affiliation is not defined for this child-parent constellation.")


@nottest
def test_query_party_example():
    try:
        organisation = db.RecordType(name="Organisation").insert()
        partei = db.RecordType(
            name="Partei").add_parent(
            name="Organisation").insert()
        spd = db.Record(name="SPD").add_parent(name="Partei").insert()
        siggi = db.Record(name="SigmarGabriel").add_parent(name="SPD").insert()

        assert_equal(
            partei.get_parent("Organisation").affiliation,
            "SUBTYPING")
        assert_equal(spd.get_parent("Partei").affiliation, "INSTANTIATION")
        assert_equal(siggi.get_parent("SPD").affiliation, "PARTHOOD")

        q = db.execute_query("FIND Organisation")
        assert_equal(len(q), 3)
        assert_is_not_none(q.get_entity_by_id(organisation.id))
        assert_is_not_none(q.get_entity_by_id(partei.id))
        assert_is_not_none(q.get_entity_by_id(spd.id))

        q = db.execute_query("FIND Partei")
        assert_equal(len(q), 2)
        assert_is_not_none(q.get_entity_by_id(partei.id))
        assert_is_not_none(q.get_entity_by_id(spd.id))

        q = db.execute_query("FIND SPD")
        assert_equal(len(q), 2)
        assert_is_not_none(q.get_entity_by_id(spd.id))
        assert_is_not_none(q.get_entity_by_id(siggi.id))

    finally:
        try:
            siggi.delete()
        except BaseException:
            pass
        try:
            spd.delete()
        except BaseException:
            pass
        try:
            partei.delete()
        except BaseException:
            pass
        try:
            organisation.delete()
        except BaseException:
            pass