test_permissions.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 30.01.2017.

@author: tf
"""

from __future__ import absolute_import
from nose.tools import assert_true, assert_equal, assert_raises, assert_false, assert_is_none, assert_is_not_none, nottest, with_setup  # @UnresolvedImport
from pytest import raises
import caosdb as db
from nose.tools import (assert_equal, assert_false,  # @UnresolvedImport
                        assert_is_none, assert_is_not_none, assert_raises,
                        assert_true, nottest, with_setup)

from .test_misc import test_info

test_user = "test_user"
test_role = "test_role"
test_pw = "passphrase1P!"


def setup_module():
    insert_test_user()


def teardown_module():
    switch_to_admin_user()
    db.administration._delete_user(name="test_user")
    db.administration._delete_role(name=test_role)
    import os
    try:
        os.remove("test.dat")
    except BaseException:
        pass
    try:
        os.remove("test2.dat")
    except BaseException:
        pass


@nottest
def switch_to_test_user():
    db.configure_connection(username=test_user, password=test_pw,
                            password_method="plain")


def switch_to_admin_user():
    db.configure_connection()


def deny_permission(entity, permission, username=test_user, priority=False):
    switch_to_admin_user()
    entity.retrieve_acl()
    entity.deny(
        realm="CaosDB",
        username=username,
        permission=permission,
        priority=priority)
    entity.update_acl()
    entity.acl = None
    switch_to_test_user()


def grant_permission(entity, permission, username=test_user, priority=False,
                     switch=True):
    if switch:
        switch_to_admin_user()
    entity.retrieve_acl()
    entity.grant(
        realm="CaosDB",
        username=username,
        permission=permission,
        priority=priority)
    ret = entity.update_acl()
    entity.acl = None
    if switch:
        switch_to_test_user()
    return ret


@nottest
def insert_test_user():
    try:
        db.administration._delete_user(name=test_user)
    except BaseException:
        pass
    try:
        db.administration._delete_role(name=test_role)
    except BaseException:
        pass

    db.administration._insert_user(
        name=test_user,
        password=test_pw,
        status="ACTIVE")
    db.administration._insert_role(name=test_role, description="A test role")
    db.administration._set_roles(username=test_user, roles=[test_role])
    set_transaction_permissions_test_role()


def set_transaction_permissions_test_role():
    switch_to_admin_user()
    db.administration._set_permissions(
        role=test_role, permission_rules=[
            db.administration.PermissionRule(
                "Grant", "TRANSACTION:*")])


def revoke_permissions_test_role():
    switch_to_admin_user()
    db.administration._set_permissions(
        role=test_role, permission_rules=[])


def teardown():
    setup()


def setup():
    switch_to_admin_user()
    try:
        db.execute_query("FIND Test*").delete()
    except BaseException:
        pass
    test_info()


@with_setup(setup, teardown)
def test_basic_acl_stuff():
    p = db.Property(
        name="TestProperty",
        datatype=db.TEXT).insert(
        flags={
            "ACL": None})
    assert_true(p.is_valid())
    assert_is_not_none(p.acl)

    p.retrieve(flags={"ACL": None})
    assert_true(p.is_valid())
    assert_is_not_none(p.acl)
    assert_true(isinstance(p.acl, db.ACL))
    assert_false(p.acl.is_empty())

    user_acl = p.acl.get_acl_for_user(
        db.get_config().get("Connection", "username"))
    print(user_acl)

    assert_is_not_none(user_acl)
    assert_true(isinstance(user_acl, db.ACL))
    assert_false(user_acl.is_empty())

    user_permissions = p.acl.get_permissions_for_user(
        db.get_config().get("Connection", "username"))
    assert_is_not_none(user_permissions)
    assert_true(isinstance(user_permissions, set))
    assert_true(len(user_permissions) > 0)

    assert_true("DELETE" in user_permissions)

    other_role_permissions = p.acl.get_permissions_for_role("other_role")
    assert_false("DELETE" in other_role_permissions)


@with_setup(setup, teardown)
def test_query():
    person = db.RecordType("TestPerson").insert()
    db.Property("TestFirstName", datatype=db.TEXT).insert()
    db.Property("TestConductor", datatype=person).insert()

    dan = db.Record(
        name="TestDaniel").add_property(
        name="TestFirstName",
        value="Daniel").add_parent(person).insert()
    exp = db.RecordType(
        name="TestExperiment").add_property(
        name="TestConductor",
        value=dan.id).insert()

    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor->TestPerson",
            unique=True).id,
        exp.id)
    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor=TestPerson",
            unique=True).id,
        exp.id)
    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor=" + str(dan.id),
            unique=True).id, exp.id)
    assert_equal(
        db.execute_query(
            "FIND TestExperiment",
            unique=True).id,
        exp.id)
    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
            unique=True).id,
        exp.id)

    '''success'''
    grant_permission(person, "RETRIEVE:*")
    grant_permission(dan, "RETRIEVE:*")
    grant_permission(exp, "RETRIEVE:*")

    switch_to_test_user()
    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
            unique=True).id,
        exp.id)

    '''failure - dan'''
    deny_permission(dan, "RETRIEVE:*")
    switch_to_test_user()

    with assert_raises(db.TransactionError) as cm:
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor WHICH has a TestFirstName=Daniel",
            unique=True)
    assert_equal(cm.exception.msg, "No such entity found.")
    '''... but works without the which clause'''
    assert_equal(
        db.execute_query(
            "FIND TestExperiment",
            unique=True).id,
        exp.id)
    '''and with the id'''
    assert_equal(
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor=" + str(dan.id),
            unique=True).id, exp.id)

    '''failure - exp'''
    grant_permission(dan, "RETRIEVE:*")
    deny_permission(exp, "RETRIEVE:*")
    switch_to_test_user()

    with assert_raises(db.TransactionError) as cm:
        db.execute_query(
            "FIND TestExperiment WHICH HAS A TestConductor=TestDaniel",
            unique=True)
    assert_equal(cm.exception.msg, "No such entity found.")

    with assert_raises(db.TransactionError) as cm:
        db.execute_query("FIND TestExperiment", unique=True)
    assert_equal(cm.exception.msg, "No such entity found.")


@with_setup(setup, teardown)
def test_update_acl():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())
    assert_is_none(p.acl)
    p.retrieve_acl()
    assert_is_not_none(p.acl)

    assert_true(
        "USE:AS_DATA_TYPE" in p.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    p.acl.deny(
        username=db.get_config().get("Connection", "username"),
        permission="USE:AS_DATA_TYPE")
    assert_false(
        "USE:AS_DATA_TYPE" in p.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    '''Success'''
    p.update(flags={"ACL": None})
    assert_true(p.is_valid())
    assert_is_not_none(p.acl)
    assert_false(
        "USE:AS_DATA_TYPE" in p.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    p2 = db.execute_query(
        "FIND TestProperty",
        unique=True,
        flags={
            "ACL": None})
    assert_true(p2.is_valid())
    assert_is_not_none(p2.acl)
    assert_true(
        "USE:AS_PROPERTY" in p2.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    assert_false(
        "USE:AS_DATA_TYPE" in p2.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    assert_true(
        "EDIT:ACL" in p2.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    p2.acl.deny(
        username=db.get_config().get("Connection", "username"),
        permission="EDIT:ACL")
    assert_false(
        "EDIT:ACL" in p2.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    '''Success'''
    p2.update()

    p3 = db.execute_query(
        "FIND TestProperty",
        unique=True,
        flags={
            "ACL": None})
    assert_true(p3.is_valid())
    assert_is_not_none(p3.acl)
    assert_true(
        "USE:AS_PROPERTY" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    assert_false(
        "USE:AS_DATA_TYPE" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    assert_false(
        "EDIT:ACL" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    switch_to_test_user()
    '''Failure'''
    p.retrieve_acl()
    p3.acl.deny(
        username=db.get_config().get("Connection", "username"),
        permission="USE:AS_PROPERTY")
    try:
        p3.update_acl()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    p3 = db.execute_query(
        "FIND TestProperty",
        unique=True,
        flags={
            "ACL": None})
    assert_true(p3.is_valid())
    assert_is_not_none(p3.acl)
    assert_true(
        "USE:AS_PROPERTY" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    assert_false(
        "USE:AS_DATA_TYPE" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))
    assert_false(
        "EDIT:ACL" in p3.acl.get_permissions_for_user(
            db.get_config().get("Connection", "username")))

    '''Failure'''
    switch_to_test_user()
    p3.acl.grant(username=test_user, permission="EDIT:ACL")
    try:
        p3.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")


@with_setup(setup, teardown)
def test_update_name():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:*")
    grant_permission(p, "UPDATE:*")

    '''Success'''
    p.name = "TestPropertyNew"
    assert_is_none(p.acl)
    p.update()

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    assert_equal(p2.name, "TestPropertyNew")

    '''Failure'''
    deny_permission(p, "UPDATE:NAME")

    p.retrieve()
    assert_false("UPDATE:NAME" in p.permissions)

    p.name = "TestPropertyEvenNewer"
    try:
        p.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p2.is_valid())
    assert_equal(p2.name, "TestPropertyNew")


@with_setup(setup, teardown)
def test_update_desc():
    p = db.Property(
        name="TestProperty",
        description="Description",
        datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:*")
    grant_permission(p, "UPDATE:DESCRIPTION")

    '''Success'''
    switch_to_test_user()
    assert_equal(p.description, "Description")
    p.description = "DescriptionNew"
    p.update()

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    assert_equal(p2.description, "DescriptionNew")

    deny_permission(p, "UPDATE:DESCRIPTION")

    '''Failure'''
    p.description = "DescriptionEvenNewer"
    try:
        p.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p2.is_valid())
    assert_equal(p2.description, "DescriptionNew")


@with_setup(setup, teardown)
def test_update_data_type():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p, "UPDATE:DATA_TYPE")

    '''Success'''
    assert_equal(p.datatype, db.TEXT)
    p.datatype = db.INTEGER
    p.update()

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    assert_equal(p2.datatype, db.INTEGER)

    deny_permission(p, "UPDATE:DATA_TYPE")

    '''Failure'''
    p.datatype = db.DOUBLE
    try:
        p.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p2.is_valid())
    assert_equal(p2.datatype, db.INTEGER)


@with_setup(setup, teardown)
def test_update_role():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p, "UPDATE:ROLE")

    '''Success'''
    rt = db.RecordType(name="TestProperty").retrieve()
    rt.update()

    rt2 = db.execute_query("FIND TestProperty", unique=True)
    assert_true(isinstance(rt2, db.RecordType))

    deny_permission(p, "UPDATE:ROLE")

    '''Failure'''
    rec = db.Record(name="TestProperty").retrieve()
    try:
        rec.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND Test*", unique=True)
    assert_true(rt2.is_valid())
    assert_true(isinstance(rt2, db.RecordType))


@with_setup(setup, teardown)
def test_update_move_file():
    upload_file = open("test.dat", "w")
    upload_file.write("hello world\n")
    upload_file.close()

    f = db.File(
        name="TestFile",
        path="/permissiontestfiles/test.dat",
        file="test.dat").insert()
    assert_true(f.is_valid())

    grant_permission(f, "RETRIEVE:ENTITY")
    grant_permission(f, "UPDATE:FILE:MOVE")

    '''SUCCESS'''
    f.path = "/otherpermissiontestfiles/test.dat"
    f.update()

    f2 = db.execute_query("FIND TestFile", unique=True)
    assert_equal(f2.path, "/otherpermissiontestfiles/test.dat")

    deny_permission(f, "UPDATE:FILE:MOVE")

    '''FAILURE'''
    f.path = "/againotherpermissiontestfiles/test.dat"
    try:
        f.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    f2 = db.execute_query("FIND TestFile", unique=True)
    assert_equal(f2.path, "/otherpermissiontestfiles/test.dat")


@with_setup(setup, teardown)
def test_update_add_file():
    upload_file = open("test.dat", "w")
    upload_file.write("test update add file: #here#\n")
    upload_file.close()

    f = db.File(name="TestFile").insert()
    assert_true(f.is_valid())

    grant_permission(f, "RETRIEVE:ENTITY")
    '''FAILURE'''

    f.path = "/permissiontestfiles/newtest.dat"
    f.file = upload_file
    try:
        f.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    f2 = db.execute_query("FIND TestFile", unique=True)
    assert_is_none(f2.path)

    '''SUCCESS'''
    grant_permission(f, "UPDATE:FILE:ADD")

    f2 = db.execute_query("FIND TestFile", unique=True)
    f2.path = "/permissiontestfiles/newtest.dat"
    f2.file = upload_file
    f2.update()
    assert_true(f2.is_valid())

    f2 = db.execute_query("FIND TestFile", unique=True)
    assert_equal(f2.path, "/permissiontestfiles/newtest.dat")


@with_setup(setup, teardown)
def test_update_change_file():
    upload_file = open("test.dat", "w")
    upload_file.write("hello world\n")
    upload_file.close()

    upload_file2 = open("test2.dat", "w")
    upload_file2.write("hello universe\n")
    upload_file2.close()

    f = db.File(
        name="TestFile",
        file=upload_file,
        path="permissiontestfiles/test.dat").insert()
    assert_true(f.is_valid())
    grant_permission(f, "RETRIEVE:ENTITY")
    grant_permission(f, "RETRIEVE:FILE")

    '''FAILURE'''
    deny_permission(f, "UPDATE:FILE:REMOVE")

    f.file = upload_file2
    try:
        f.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    f2 = db.execute_query("FIND TestFile", unique=True)
    download_file = f2.download()
    assert_equal(db.File._get_checksum(download_file),
                 db.File._get_checksum(upload_file))

    '''SUCCESS'''
    print('#################################################################')
    print('#################################################################')
    print(f.acl)
    grant_permission(f, "UPDATE:FILE:REMOVE")
    print(f.acl)
    grant_permission(f, "UPDATE:FILE:ADD")
    print(f.acl)

    f2 = db.execute_query("FIND TestFile", unique=True)
    f2.file = upload_file2
    f2.update()
    assert_true(f2.is_valid())

    f2 = db.execute_query("FIND TestFile", unique=True)
    download_file = f2.download()
    assert_equal(db.File._get_checksum(download_file),
                 db.File._get_checksum(upload_file2))


@with_setup(setup, teardown)
def test_update_add_property():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())
    p2 = db.Property(name="TestProperty2", datatype=db.TEXT).insert()
    assert_true(p2.is_valid())
    rt = db.RecordType(name="TestRecordType").insert()
    assert_true(rt.is_valid())

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_properties()), 0)

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p2, "RETRIEVE:ENTITY")
    grant_permission(rt, "RETRIEVE:ENTITY")

    grant_permission(p, "USE:*")
    grant_permission(p2, "USE:*")
    grant_permission(rt, "UPDATE:PROPERTY:ADD")

    '''Success - add p to rt'''
    rt.add_property(name="TestProperty", id=p.id).update()

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_properties()), 1)
    assert_is_not_none(rt.get_property("TestProperty"))

    deny_permission(rt, "UPDATE:PROPERTY:ADD")

    '''Failure - add p to rt'''
    rt.add_property(name="TestProperty", id=p.id)
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_properties()), 1)
    assert_is_not_none(rt2.get_property("TestProperty"))

    '''Failure - add p2 to rt'''
    rt.add_property(name="TestProperty2", id=p2.id)
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_properties()), 1)
    assert_is_not_none(rt2.get_property("TestProperty"))


@with_setup(setup, teardown)
def test_update_remove_property():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())
    p2 = db.Property(name="TestProperty2", datatype=db.TEXT).insert()
    assert_true(p2.is_valid())
    rt = db.RecordType(
        name="TestRecordType").add_property(
        name="TestProperty").add_property(
            name="TestProperty2").insert()
    assert_true(rt.is_valid())

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_properties()), 2)
    assert_is_not_none(rt.get_property("TestProperty"))
    assert_is_not_none(rt.get_property("TestProperty2"))

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p2, "RETRIEVE:ENTITY")
    grant_permission(rt, "RETRIEVE:ENTITY")

    grant_permission(p, "USE:*")
    grant_permission(p2, "USE:*")
    grant_permission(rt, "UPDATE:PROPERTY:REMOVE")

    '''Success - remove p2 from rt'''
    rt.remove_property("TestProperty2").update()

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_properties()), 1)
    assert_is_not_none(rt2.get_property("TestProperty"))
    assert_is_none(rt2.get_property("TestProperty2"))

    deny_permission(rt, "UPDATE:PROPERTY:REMOVE")

    '''Failure - remove p from rt'''
    rt.remove_property("TestProperty")
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_properties()), 1)
    assert_is_not_none(rt2.get_property("TestProperty"))


@with_setup(setup, teardown)
def test_update_add_parent():
    rt = db.RecordType(name="TestRecordType").insert()
    assert_true(rt.is_valid())
    par1 = db.RecordType(name="TestRecordTypePar1").insert()
    assert_true(par1.is_valid())
    par2 = db.RecordType(name="TestRecordTypePar2").insert()
    assert_true(par2.is_valid())

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_parents()), 0)

    grant_permission(rt, "RETRIEVE:ENTITY")
    grant_permission(par1, "RETRIEVE:ENTITY")
    grant_permission(par2, "RETRIEVE:ENTITY")
    grant_permission(par1, "USE:*")
    grant_permission(par2, "USE:*")
    grant_permission(rt, "UPDATE:PARENT:ADD")

    '''Success - add par1 to rt'''
    rt.add_parent(name="TestRecordTypePar1", id=par1.id).update()

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_parents()), 1)
    assert_is_not_none(rt2.get_parent("TestRecordTypePar1"))

    deny_permission(rt, "UPDATE:PARENT:ADD")

    '''Failure - add par1 to rt'''
    rt.add_parent(name="TestRecordTypePar1", id=par1.id)
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_parents()), 1)
    assert_is_not_none(rt2.get_parent("TestRecordTypePar1"))

    '''Failure - add par2 to rt'''
    rt.add_parent(name="TestRecordTypePar2", id=par2.id)
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt2 = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt2.get_parents()), 1)
    assert_is_not_none(rt2.get_parent("TestRecordTypePar1"))
    assert_is_none(rt2.get_parent("TestRecordTypePar2"))


@with_setup(setup, teardown)
def test_update_remove_parent():
    par1 = db.RecordType(name="TestRecordTypePar1").insert()
    assert_true(par1.is_valid())
    par2 = db.RecordType(name="TestRecordTypePar2").insert()
    assert_true(par2.is_valid())
    rt = db.RecordType(name="TestRecordType").add_parent(
        par1).add_parent(par2).insert()
    assert_true(rt.is_valid())

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_parents()), 2)
    assert_is_not_none(rt.get_parent("TestRecordTypePar1"))
    assert_is_not_none(rt.get_parent("TestRecordTypePar2"))

    grant_permission(rt, "RETRIEVE:ENTITY")
    grant_permission(par1, "RETRIEVE:ENTITY")
    grant_permission(par2, "RETRIEVE:ENTITY")
    grant_permission(par1, "USE:*")
    grant_permission(par2, "USE:*")
    grant_permission(rt, "UPDATE:PARENT:REMOVE")

    '''Success'''
    rt.remove_parent("TestRecordTypePar2").update()

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_parents()), 1)
    assert_is_not_none(rt.get_parent("TestRecordTypePar1"))
    assert_is_none(rt.get_parent("TestRecordTypePar2"))

    deny_permission(rt, "UPDATE:PARENT:REMOVE")

    '''Failure'''
    rt.remove_parent("TestRecordTypePar1")
    try:
        rt.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    rt = db.execute_query("FIND TestRecordType", unique=True)
    assert_equal(len(rt.get_parents()), 1)
    assert_is_not_none(rt.get_parent("TestRecordTypePar1"))
    assert_is_none(rt.get_parent("TestRecordTypePar2"))


@with_setup(setup, teardown)
def test_update_value():
    p = db.Property(
        name="TestProperty",
        datatype=db.TEXT,
        value="Value").insert()
    assert_true(p.is_valid())

    p = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    assert_equal(p.value, "Value")

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p, "UPDATE:VALUE")

    '''Success'''
    p.value = "NewValue"
    p.update()

    p = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    assert_equal(p.value, "NewValue")

    deny_permission(p, "UPDATE:VALUE")

    '''Failure'''
    p.value = "EvenNewerValue"
    try:
        p.update()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    p2 = db.execute_query("FIND Test*", unique=True)
    assert_true(p2.is_valid())
    assert_equal(p2.value, "NewValue")


@with_setup(setup, teardown)
def test_deletion():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:ENTITY")
    '''Failure'''

    p = db.execute_query("FIND Test*", unique=True)
    assert_true(p.is_valid())
    try:
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    '''Success'''
    grant_permission(p, "DELETE")

    p.delete()
    assert_equal(
        p.get_messages()[0].description,
        "This entity has been deleted successfully.")


@with_setup(setup, teardown)
def test_retrieve_acl():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    '''Success'''
    grant_permission(p, "RETRIEVE:*")
    p.retrieve_acl()

    '''Failure'''
    deny_permission(p, "RETRIEVE:ACL")

    try:
        p.retrieve_acl()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")


@with_setup(setup, teardown)
def test_retrieve_history():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    grant_permission(p, "RETRIEVE:ENTITY")

    '''Failure'''
    try:
        p.retrieve(flags={"H": None})
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    '''Success'''
    grant_permission(p, "RETRIEVE:HISTORY")

    p.retrieve(flags={"H": None})
    assert_equal(p.messages["History"], ('Insert', None))


@with_setup(setup, teardown)
def test_retrieve_entity():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()

    assert_true(p.is_valid())

    ''' Success'''
    grant_permission(p, "RETRIEVE:*")
    db.Property(name="TestProperty").retrieve()

    '''Failure'''
    deny_permission(p, "RETRIEVE:ENTITY")

    try:
        p.retrieve()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")


@with_setup(setup, teardown)
def test_retrieve_owner():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())

    '''Failure'''
    grant_permission(p, "RETRIEVE:ENTITY")
    deny_permission(p, "RETRIEVE:OWNER")

    try:
        p.retrieve(flags={"owner": None})
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, "You are not allowed to do this.")

    '''Success'''
    grant_permission(p, "RETRIEVE:OWNER")

    p.retrieve(flags={"owner": None})


@with_setup(setup, teardown)
def test_download_file():
    upload_file = open("test.dat", "w")
    upload_file.write("hello world\n")
    upload_file.close()

    f = db.File(
        name="TestFile",
        file=upload_file,
        path="permissiontestfiles/test.dat").insert()
    assert_true(f.is_valid())

    '''FAILURE'''
    f2 = db.execute_query("FIND TestFile", unique=True)

    grant_permission(f2, "RETRIEVE:ENTITY")
    deny_permission(f2, "RETRIEVE:FILE")

    try:
        f.download()
    except db.TransactionError as e:
        assert_equal(
            e.msg,
            "Request failed. The response returned with status 403.")

    '''SUCCESS'''
    grant_permission(f2, "RETRIEVE:FILE")

    f2 = db.execute_query("FIND TestFile", unique=True)
    download_file = f2.download()
    assert_equal(db.File._get_checksum(download_file),
                 db.File._get_checksum(upload_file))


@with_setup(setup, teardown)
def test_grant_priority_permission():
    p = db.Property(name="TestProperty2", datatype=db.TEXT).insert()

    switch_to_test_user()
    try:
        # other user cannot delete this.
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')

    deny_permission(p, "DELETE")

    try:
        # still not working
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')

    # now its working
    grant_permission(p, "DELETE", priority=True)
    p.delete()


@with_setup(setup, teardown)
def test_deny_priority_permission():
    p = db.Property(name="TestProperty1", datatype=db.TEXT).insert()

    switch_to_test_user()
    try:
        # other user cannot delete this.
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')

    deny_permission(p, "DELETE")

    try:
        # still not working
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')

    # now its working
    grant_permission(p, "DELETE")
    deny_permission(p, "DELETE", priority=True)
    try:
        # still not working
        p.delete()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')


@with_setup(setup, teardown)
def test_change_priority_permission():
    entity = db.Property(name="TestProperty1", datatype=db.TEXT).insert()
    grant_permission(entity, "*")

    entity.retrieve(flags={"ACL": None})
    # try and grant a priority permission with the test_user
    entity.acl.grant(
        username="test_user",
        permission="USE:AS_PROPERTY",
        priority=True)

    try:
        entity.update_acl()
        assert_true(False)
    except db.TransactionError as e:
        assert_equal(e.msg, 'You are not allowed to do this.')


@with_setup(setup, teardown)
def test_permissions_there():
    entity = db.Property(name="TestProperty1", datatype=db.TEXT).insert()
    assert_is_not_none(entity.permissions)

    entity2 = db.Property(id=entity.id).retrieve()
    assert_is_not_none(entity2.permissions)

    assert_true(entity2.is_permitted("RETRIEVE:ACL"))


@with_setup(setup, teardown)
def test_grant_nonsense_permission():
    entity = db.Property(name="TestProperty1", datatype=db.TEXT).insert()
    entity.retrieve_acl()
    entity.grant(username="someone", permission="PAINT:BLUE")
    assert_raises(db.TransactionError, entity.update_acl)


@with_setup(setup, teardown)
def test_global_acl_there():
    assert_is_not_none(db.get_global_acl())
    assert_true(isinstance(db.get_global_acl(), db.ACL))
    assert_is_not_none(db.get_known_permissions())
    assert_true(isinstance(db.get_known_permissions(), db.Permissions))
    print(db.get_known_permissions())
    print(db.get_global_acl())


@with_setup(setup, teardown)
def test_use_as_property():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())
    switch_to_test_user()
    rt = db.RecordType(name="TestRecordType").add_property(name="TestProperty")

    deny_permission(p, "USE:AS_PROPERTY")
    '''Failure'''
    deny_permission(p, "USE:AS_PROPERTY")
    with assert_raises(db.TransactionError) as cm:
        rt.insert()
    ex = cm.exception
    assert_equal(ex.msg, "Entity has unqualified properties.")
    assert_equal(rt.get_property("TestProperty").get_errors()[0].code, "403")

    '''Success'''
    grant_permission(p, "USE:AS_PROPERTY")
    rt2 = db.RecordType(
        name="TestRecordType").add_property(
        name="TestProperty").insert()
    assert_true(rt2.is_valid())


@with_setup(setup, teardown)
def test_use_as_data_type():
    dt_name = "TestPersonRecordType"
    dt = db.RecordType(name=dt_name).insert()

    '''success'''
    grant_permission(dt, "RETRIEVE:ENTITY")
    grant_permission(dt, "USE:AS_DATA_TYPE")
    db.Property(name="TestConductorProperty", datatype=dt_name).insert()

    deny_permission(dt, "USE:AS_DATA_TYPE")
    with assert_raises(db.TransactionError) as cm:
        db.Property(name="TestConductorProperty2", datatype=dt_name).insert()
    assert_equal(cm.exception.msg, "You are not allowed to do this.")


@with_setup(setup, teardown)
def test_use_as_reference():

    p = db.RecordType(name="TestRecordTypeForReference").insert()
    rec = db.Record().add_parent(parent=p).insert()

    grant_permission(p, "RETRIEVE:ENTITY")
    grant_permission(p, "USE:*")
    grant_permission(rec, "RETRIEVE:ENTITY")
    grant_permission(rec, "USE:*")

    rt = db.RecordType(
        name="TestRecordType").add_property(
        name="TestRecordTypeForReference",
        value=rec).insert()
    assert_true(rt.is_valid())

    deny_permission(rec, "USE:AS_REFERENCE")
    rt2 = db.RecordType(
        name="TestRecordType2").add_property(
        name="TestRecordTypeForReference",
        value=rec)
    with assert_raises(db.TransactionError) as cm:
        rt2.insert()
    assert_equal(cm.exception.msg, "Entity has unqualified properties.")
    assert_equal(rt2.get_property(p.name).get_errors()[0].code, "403")


@with_setup(setup, teardown)
def test_use_as_parent():
    p = db.Property(name="TestProperty", datatype=db.TEXT).insert()
    assert_true(p.is_valid())
    switch_to_test_user()

    p2 = db.Property(
        name="TestPropertyChild",
        datatype=db.TEXT).add_parent(
        name="TestProperty")

    deny_permission(p, "USE:AS_PARENT")
    '''Failure'''
    deny_permission(p, "USE:AS_PARENT")
    with assert_raises(db.TransactionError) as cm:
        p2.insert()
    assert_equal(cm.exception.msg, "Entity has unqualified parents.")
    assert_equal(p2.get_parent("TestProperty").get_errors()[0].code, "403")

    '''Success'''
    grant_permission(p, "USE:AS_PARENT")
    p3 = db.Property(
        name="TestPropertyChild").add_parent(
        name="TestProperty").insert()
    assert_true(p3.is_valid())


@with_setup(setup, teardown)
def test_access_control_job_bug():
    """test_access_control_job_bug.

    The AccessControl class attempts to retrieve an entity with an id < 0,
    which causes a DataTruncation Error in the SQL backend.

    This happens if a user does not have the permission "TRANSACTION:INSERT"
    """
    revoke_permissions_test_role()
    switch_to_test_user()
    with raises(db.AuthorizationException):
        rt1 = db.RecordType(name="TestRT").add_parent(id=-5).insert()
    set_transaction_permissions_test_role()


@with_setup(setup, teardown)
def test_check_entity_acl_roles():
    '''Test the CheckEntityACLRoles class.

    Insert an entity.
    With "CHECK_ENTITY_ACL_ROLES_MODE=MUST":
    Add permissions for a non-existing user role -> error
    Set "CHECK_ENTITY_ACL_ROLES_MODE=SHOULD":
    Add permissions for a non-existing user role -> warning
    reset to CHECK_ENTITY_ACL_ROLES_MODE=MUST
    '''

    reset = db.administration.get_server_property(
        "CHECK_ENTITY_ACL_ROLES_MODE")
    assert reset == "MUST"

    p = db.Property(name="TestP", datatype=db.TEXT,
                    description="test_check_entity_acl_roles").insert()

    # test failure with error
    with raises(db.TransactionError) as cm:
        grant_permission(p, "USE:AS_PARENT", username="asdf-non-existing",
                         switch=False)
    errors = cm.value.entity.get_errors()
    assert errors[0].description == "User Role does not exist."
    db.administration.set_server_property(
        "CHECK_ENTITY_ACL_ROLES_MODE", "SHOULD")

    # test success with warning
    ret = grant_permission(p, "USE:AS_PROPERTY",
                           username="asdf-non-existing", switch=False)
    assert ret.get_warnings()[0].description == "User Role does not exist."

    db.administration.set_server_property("CHECK_ENTITY_ACL_ROLES_MODE", reset)