Merge branch 'f-fsm' into 'dev'

F fsm

See merge request !3

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added (for new features)
 
+* Tests for entity state [caosdb-server!62](https://gitlab.com/caosdb/caosdb-server/-/merge_requests/62)
 * Tests for version history
 * Tests for inheritance bug (caosdb/caosdb-server!54)
 * Tests for versioning

Makefile

@@ -49,6 +49,5 @@ help:
 autopep8:
 	autopep8 -ri tests
 
-
 # Meta target to call the other targets.
 all: autopep8 test

tests/test_affiliation.py

@@ -65,7 +65,6 @@ def teardown():
 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"))
 

tests/test_state.py

+import pytest
+import caosdb as db
+from caosdb import administration as admin
+
+_ORIGINAL_EXT_ENTITY_STATE = ""
+_DELETE_ROLES = ["reviewer", "team-leader", "normal"]
+_DELETE_USERS = ["reviewer_user", "team-leader_user", "normal_user"]
+# _DELETE_ROLES = []
+# _DELETE_USERS = []
+_PASSWORD = "1234asdf!P"
+
+
+def switch_to_admin_user():
+    db.configure_connection()
+
+
+def setup_users():
+    for role in ["reviewer", "team-leader", "normal"]:
+        _DELETE_ROLES.append(role)
+        db.administration._insert_role(name=role, description="A test role")
+
+        username = role + "_user"
+        _DELETE_USERS.append(username)
+        db.administration._insert_user(
+            name=username,
+            password=_PASSWORD,
+            status="ACTIVE")
+        db.administration._set_roles(username=username, roles=[role])
+
+    db.administration._set_permissions(
+        role="reviewer", permission_rules=[
+            db.administration.PermissionRule(
+                "Grant", "TRANSACTION:*"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:EditTransition"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:ReviewTransition"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:Transition1"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:Transition4"),
+
+        ])
+    db.administration._set_permissions(
+        role="normal", permission_rules=[
+            db.administration.PermissionRule(
+                "Grant", "TRANSACTION:*"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:Transition4"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:TRANSITION:EditTransition"),
+        ])
+    db.administration._set_permissions(
+        role="team-leader", permission_rules=[
+            db.administration.PermissionRule(
+                "Grant", "TRANSACTION:*"),
+            db.administration.PermissionRule(
+                "Grant", "STATE:*"),
+        ])
+
+
+def switch_to_test_user(role):
+    db.configure_connection(username=role + "_user",
+                            password=_PASSWORD,
+                            password_method="plain")
+
+
+def teardown_module():
+    for user in _DELETE_USERS:
+        try:
+            db.administration._delete_user(name=user)
+        except BaseException:
+            pass
+    for role in _DELETE_ROLES:
+        try:
+            db.administration._delete_role(name=role)
+        except BaseException:
+            pass
+
+    d = db.execute_query("FIND ENTITY WITH ID > 99")
+    if len(d) > 0:
+        d.delete(flags={"forceFinalState": "true"})
+    admin.set_server_property("EXT_ENTITY_STATE",
+                              _ORIGINAL_EXT_ENTITY_STATE)
+
+
+def setup_module():
+    global _ORIGINAL_EXT_ENTITY_STATE
+    try:
+        _ORIGINAL_EXT_ENTITY_STATE = admin.get_server_property(
+            "EXT_ENTITY_STATE")
+    except KeyError:
+        pass
+    teardown_module()
+    setup_users()
+
+    db.RecordType("State").insert()
+    db.RecordType("StateModel").insert()
+    db.RecordType("Transition").insert()
+    db.Property(name="from", datatype="State").insert()
+    db.Property(name="to", datatype="State").insert()
+    db.Property(name="initial", datatype="State").insert()
+    db.Property(name="final", datatype="State").insert()
+
+    st1 = db.Record(
+        "State1",
+        description="DescState1").add_parent("State").insert(
+        flags={
+            "ACL": None})
+    state_acl = db.ACL()
+    state_acl.grant(role="role1", permission="UPDATE:DESCRIPTION")
+    state_acl = db.State.create_state_acl(state_acl)
+    st1.acl = state_acl.combine(st1.acl)
+    st1.update_acl()
+
+    db.Record("State2", description="DescState2").add_parent("State").insert()
+    db.Record("State3", description="DescState3").add_parent("State").insert()
+    # 1->
+    db.Record("Transition1").add_parent("Transition").add_property(
+        "from", "State1").add_property("to", "State2").insert()
+    # 2->3
+    db.Record("Transition2").add_parent("Transition").add_property(
+        "from", "State2").add_property("to", "State3").insert()
+    # 3->1
+    db.Record("Transition3").add_parent("Transition").add_property(
+        "from", "State3").add_property("to", "State1").insert()
+    # 2->2
+    db.Record("Transition4").add_parent("Transition").add_property(
+        "from", "State2").add_property("to", "State2").insert()
+
+    db.Record("Model1").add_parent("StateModel").add_property(
+        "Transition",
+        datatype=db.LIST("Transition"),
+        value=[
+            "Transition1",
+            "Transition2",
+            "Transition3",
+            "Transition4"]).add_property(
+        "initial",
+        "State1").add_property(
+                "final",
+        "State1").insert()
+
+
+def teardown():
+    switch_to_admin_user()
+    d = db.execute_query("FIND TestRT")
+    if len(d) > 0:
+        d.delete(flags={"forceFinalState": "true"})
+
+
+def setup():
+    admin.set_server_property("EXT_ENTITY_STATE", "ENABLED")
+    teardown()
+    db.RecordType("TestRT").insert()
+
+
+def test_plugin_disabled():
+    """Test the behavior when the state machine extension is disabled on the server."""
+    admin.set_server_property("EXT_ENTITY_STATE", "DISABLED")
+
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    assert rec.get_property("State") is None
+
+    rec_insert = rec.insert(sync=False)
+    assert rec_insert.get_property("State") is None
+    assert rec_insert.state == rec.state
+
+    # however, state was not stored
+    rec_retrieve = db.Record.retrieve(rec_insert.id)
+    assert rec_retrieve.get_property("State") is None
+    assert rec_retrieve.state is None
+
+    # also, properties are not interpreted as state
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.add_property("State", "State2")
+    rec_insert = rec.insert(sync=False)
+
+    rec_retrieve = db.Record.retrieve(rec_insert.id)
+    assert rec_retrieve.get_property("State").value is not None
+    assert rec_retrieve.state is None
+
+
+def test_state_message():
+    """State is constant between creation, insertion, retrieval."""
+    state_id = db.execute_query("FIND Record State1", unique=True).id
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    assert rec.get_property("State") is None
+
+    rec_insert = rec.insert(sync=False)
+    assert rec_insert.get_property("State") is None
+    assert rec_insert.state == rec.state
+    assert rec_insert.state.description == "DescState1"
+    assert rec_insert.state.id == str(state_id)
+    assert rec_insert.state.transitions == set(
+        [db.Transition(name="Transition1", from_state="State1", to_state="State2")])
+
+    rec_retrieve = db.Record.retrieve(rec_insert.id)
+    assert rec_retrieve.get_property("State") is None
+    assert rec_retrieve.state == rec.state
+    assert rec_retrieve.state.description == "DescState1"
+    assert rec_retrieve.state.id == str(state_id)
+    assert rec_retrieve.state.transitions == set(
+        [db.Transition(name="Transition1", from_state="State1", to_state="State2")])
+
+    # test sparseState flag
+    rec_retrieve = db.Record(
+        id=rec_insert.id).retrieve(
+        flags={
+            "sparseState": "true"})
+    assert rec_retrieve.get_property("State") is None
+    assert rec_retrieve.state.id == str(state_id)
+    assert rec_retrieve.state.name is None
+    assert rec_retrieve.state.description is None
+    assert rec_retrieve.state.transitions is None
+
+
+def test_state_query():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec.insert()
+    assert rec.get_property("State") is None
+
+    result = db.execute_query("FIND TestRT WITH State = State1", unique=True)
+    assert result.id == rec.id
+
+    assert result.get_property("State") is None
+    assert result.state == db.State(model="Model1", name="State1")
+
+
+def test_state_transition():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec.insert()
+
+    rec.state = db.State(model="Model1", name="State2")
+    rec_update = rec.update(sync=False)
+
+    rec_retrieve = db.Record.retrieve(rec.id)
+    assert rec_retrieve.state == rec_update.state
+
+
+def test_transition_permissions():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+
+    rec.state = db.State(model="Model1", name="State1")
+    rec.insert()
+
+    rec_state_1 = db.Record.retrieve(rec.id)
+    assert rec_state_1.state == db.State(model="Model1", name="State1")
+    assert rec_state_1.state.transitions == set([db.Transition(name="Transition1",
+                                                               from_state="State1",
+                                                               to_state="State2")])
+    switch_to_test_user("team-leader")
+    rec_state_1 = db.Record.retrieve(rec.id)
+    assert rec_state_1.state == db.State(model="Model1", name="State1")
+    assert rec_state_1.state.transitions == set([db.Transition(name="Transition1",
+                                                               from_state="State1",
+                                                               to_state="State2")])
+    switch_to_test_user("reviewer")
+    rec_state_1 = db.Record.retrieve(rec.id)
+    assert rec_state_1.state == db.State(model="Model1", name="State1")
+    assert rec_state_1.state.transitions == set([db.Transition(name="Transition1",
+                                                               from_state="State1",
+                                                               to_state="State2")])
+    switch_to_test_user("normal")
+    rec_state_1 = db.Record.retrieve(rec.id)
+    assert rec_state_1.state == db.State(model="Model1", name="State1")
+    assert rec_state_1.state.transitions is None
+
+    rec.state = db.State(model="Model1", name="State2")
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this" in str(exc.value)
+
+    switch_to_test_user("reviewer")
+    rec.update(sync=False)
+
+    switch_to_test_user("team-leader")
+    rec_state_2 = db.Record.retrieve(rec.id)
+    assert rec_state_2.state == db.State(model="Model1", name="State2")
+    assert rec_state_2.state.transitions == set([db.Transition(name="Transition2",
+                                                               from_state="State2",
+                                                               to_state="State3"),
+                                                 db.Transition(name="Transition4",
+                                                               from_state="State2",
+                                                               to_state="State2")])
+
+    switch_to_test_user("reviewer")
+    rec_state_2 = db.Record.retrieve(rec.id)
+    assert rec_state_2.state == db.State(model="Model1", name="State2")
+    assert rec_state_2.state.transitions == set([db.Transition(name="Transition4",
+                                                               from_state="State2",
+                                                               to_state="State2")])
+
+    rec.state = db.State(model="Model1", name="State3")
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+
+    switch_to_test_user("team-leader")
+    rec.update(sync=False)
+    rec_state_3 = db.Record.retrieve(rec.id)
+    assert rec_state_3.state == db.State(model="Model1", name="State3")
+    assert rec_state_3.state.transitions == set([db.Transition(name="Transition3",
+                                                               from_state="State3",
+                                                               to_state="State1")])
+
+    switch_to_test_user("reviewer")
+    rec_state_3 = db.Record.retrieve(rec.id)
+    assert rec_state_3.state == db.State(model="Model1", name="State3")
+    assert rec_state_3.state.transitions is None
+
+
+def test_transition_not_allowed():
+    """Unallowed transitions return errors and do not update the entity."""
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec_insert = rec.insert(sync=False)
+
+    rec_insert.state = db.State(model="Model1", name="State3")
+    with pytest.raises(db.TransactionError):
+        rec_update = rec_insert.update(
+            sync=False, raise_exception_on_error=False)
+        assert len(rec_update.get_errors()) == 1
+        assert rec_update.get_errors(
+        )[0].description == "Transition not allowed."
+        db.common.models.raise_errors(rec_update)
+
+    rec_retrieve = db.Record.retrieve(rec_insert.id)
+    assert rec_retrieve.state == rec.state
+
+
+def test_wrong_initial():
+    """the first state has to be an initial state"""
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State2")
+    with pytest.raises(db.TransactionError):
+        rec.insert()
+
+    assert len(rec.get_errors()) == 1
+    assert rec.get_errors()[0].description == "Initial state not allowed."
+
+
+def test_wrong_final():
+    """deletion of the entity or the state is only possible in final states"""
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec.insert()
+
+    rec.state = db.State(model="Model1", name="State2")
+    rec.update()
+
+    with pytest.raises(db.TransactionError):
+        rec.delete()
+
+    assert len(rec.get_errors()) == 1
+    assert rec.get_errors()[0].description == "Final state not allowed."
+
+    rec.state = db.State(model="Model1", name="State3")
+    rec.update()
+    rec.state = db.State(model="Model1", name="State1")
+    rec.update()
+    rec.delete()
+
+
+def test_multiple_states():
+    """currently, only one state is allowed"""
+    rec = db.Record()
+    rec.add_parent("TestRT")
+
+    state1 = db.State(model="Model1", name="State1")
+    state2 = db.State(model="Model1", name="State2")
+
+    class TestState:
+        def to_xml(self, xml):
+            xml.append(state1.to_xml())
+            xml.append(state2.to_xml())
+
+        def clear_server_messages(self):
+            pass
+    rec.messages = TestState()
+    with pytest.raises(db.TransactionError):
+        rec_insert = rec.insert(sync=False)
+
+
+def test_broken_state_missing_model():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(name="State1", model=None)
+    with pytest.raises(db.TransactionError):
+        rec.insert()
+    assert len(rec.get_errors()) == 1
+    assert rec.get_errors()[0].description == "State model not specified."
+
+
+def test_broken_state_missing_state_name():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name=None)
+    with pytest.raises(db.TransactionError):
+        rec.insert()
+    assert len(rec.get_errors()) == 1
+    assert rec.get_errors()[0].description == "State not specified."
+
+
+def test_state_not_in_state_model():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="UnknownState")
+    with pytest.raises(db.TransactionError):
+        rec.insert()
+    assert len(rec.get_errors()) == 1
+    assert rec.get_errors()[
+        0].description == "State does not exist in this StateModel."
+
+
+def test_transition_without_state_change():
+    rec = db.Record()
+    rec.description = "old description"
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec_insert = rec.insert(sync=False)
+
+    # first update attempt (should fail, because 1->1 not allowed)
+    rec_insert.description = "updated description 1"
+    with pytest.raises(db.TransactionError):
+        # transition 1 -> 1 not allowed
+        rec_update = rec_insert.update(
+            sync=False, raise_exception_on_error=False)
+        assert len(rec_update.get_errors()) == 1
+        assert rec_update.get_errors(
+        )[0].description == "Transition not allowed."
+        db.common.models.raise_errors(rec_update)
+
+    # second update with transition to state2
+    rec_update = db.Record(id=rec_insert.id).retrieve()
+    assert rec_update.state == db.State(model="Model1", name="State1")
+    assert rec_update.description == "old description"
+
+    rec_update.description = "updated description 2"
+    rec_update.state = db.State(name="State2", model="Model1")
+    # transition 1 -> 2 is allowed.
+    rec_update.update()
+
+    # third update without state change
+    rec_update = db.Record(id=rec_insert.id).retrieve()
+    assert rec_update.state == db.State(model="Model1", name="State2")
+    assert rec_update.description == "updated description 2"
+
+    rec_update.description = "updated description 3"
+    # transition 2 -> 2 is also allowed.
+    rec_update.update()
+
+    rec_final = db.Record.retrieve(rec_insert.id)
+    assert rec_final.description == "updated description 3"
+    assert rec_final.state == db.State(model="Model1", name="State2")
+
+
+def test_transfer_state_acl():
+    rec = db.Record()
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    insert_rec = rec.insert(flags={"ACL": None})
+
+    state_acl = db.ACL().combine(db.get_global_acl())
+    state_acl.grant(role="role1", permission="UPDATE:DESCRIPTION")
+
+    # the acl has been transfered from the state record
+    assert insert_rec.acl == state_acl
+
+
+def test_full_edit_review_publish_cycle():
+    edit_state = db.Record(
+        "EditState",
+        description="Any user can edit, only team-leader can delete.").add_parent("State").insert(
+        flags={
+            "ACL": None})
+
+    edit_acl = db.ACL()
+    edit_acl.grant(role="team-leader", permission="*")
+    edit_acl.grant(role="reviewer", permission="UPDATE:*")
+    edit_acl.grant(role="normal", permission="UPDATE:*")
+    edit_acl = db.State.create_state_acl(edit_acl)
+    edit_state.acl = edit_acl.combine(edit_state.acl)
+    edit_state.update_acl()
+
+    review_state = db.Record(
+        "ReviewState",
+        description="Only users with the 'reviewer' role can edit, only team-leader can delete.").add_parent("State").insert(
+        flags={
+            "ACL": None})
+
+    review_acl = db.ACL()
+    review_acl.grant(role="team-leader", permission="*")
+    review_acl.grant(role="reviewer", permission="UPDATE:*")
+    review_acl = db.State.create_state_acl(review_acl)
+    review_state.acl = review_acl.combine(review_state.acl)
+    review_state.update_acl()
+
+    published_state = db.Record(
+        "PublishedState",
+        description="Entity is read-only for everyone.").add_parent("State").insert(
+        flags={
+            "ACL": None})
+
+    published_acl = db.ACL()
+    published_acl = db.State.create_state_acl(published_acl)
+    published_state.acl = published_acl.combine(published_state.acl)
+    published_state.update_acl()
+
+    db.Record("EditTransition").add_parent("Transition").add_property(
+        "from", "EditState").add_property("to", "EditState").insert()
+    db.Record("StartReviewTransition").add_parent("Transition").add_property(
+        "from", "EditState").add_property("to", "ReviewState").insert()
+    db.Record("ReviewTransition").add_parent("Transition").add_property(
+        "from", "ReviewState").add_property("to", "ReviewState").insert()
+    db.Record("RejectTransition").add_parent("Transition").add_property(
+        "from", "ReviewState").add_property("to", "EditState").insert()
+    db.Record("PublishTransition").add_parent("Transition").add_property(
+        "from", "ReviewState").add_property("to", "PublishedState").insert()
+    db.Record("UnpublishTransition").add_parent("Transition").add_property(
+        "from", "PublishedState").add_property("to", "EditState").insert()
+
+    db.Record("EditReviewPublish").add_parent("StateModel").add_property(
+        "Transition",
+        datatype=db.LIST("Transition"),
+        value=[
+            "EditTransition",
+            "StartReviewTransition",
+            "ReviewTransition",
+            "RejectTransition",
+            "PublishTransition",
+            "UnpublishTransition"]).add_property(
+        "initial",
+        "EditState").add_property(
+                "final",
+        "EditState").insert()
+    db.Property("TestProperty", datatype=db.TEXT).insert()
+
+    def val():
+        s = "val"
+        i = 0
+        while True:
+            i += 1
+            yield s + str(i)
+    val = val()
+    # tests begin
+
+    rec = db.Record().add_parent("TestRT")
+    rec.add_property("TestProperty", "val1")
+    rec.state = db.State(model="EditReviewPublish", name="EditState")
+    rec.insert()
+
+    # as team-leader
+    switch_to_test_user("team-leader")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # start review
+    switch_to_test_user("team-leader")
+    rec.state = db.State(model="EditReviewPublish", name="ReviewState")
+    rec.update()
+
+    # as team-leader
+    switch_to_test_user("team-leader")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    # reject
+    switch_to_test_user("team-leader")
+    rec.state = db.State(model="EditReviewPublish", name="EditState")
+    rec.update()
+
+    # as team-leader
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # start review
+    switch_to_test_user("team-leader")
+    rec.state = db.State(model="EditReviewPublish", name="ReviewState")
+    rec.update()
+
+    # as team-leader
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    # publish
+    switch_to_test_user("team-leader")
+    rec.state = db.State(model="EditReviewPublish", name="PublishedState")
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+        # updating the property and the state fails
+    assert "You are not allowed to do this." in str(exc.value)
+
+    rec = db.Record(id=rec.id).retrieve()
+    rec.state = db.State(model="EditReviewPublish", name="PublishedState")
+    rec.update(sync=False)
+
+    # as team-leader
+    rec.get_property("TestProperty").value = next(val)
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    # unpublish
+    switch_to_test_user("team-leader")
+    rec.state = db.State(model="EditReviewPublish", name="EditState")
+    rec.update()
+
+    # as team-leader
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as reviewer
+    switch_to_test_user("reviewer")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+    # as other user
+    switch_to_test_user("normal")
+    rec.get_property("TestProperty").value = next(val)
+    rec.update()
+
+
+def test_automatic_record_state():
+    rt = db.RecordType("StateFullRT").add_parent("TestRT")
+    rt.state = db.State(model="Model1", name="State1")
+    rt.insert()
+
+    rec = db.Record("TestRec").add_parent("StateFullRT")
+    rec.insert()
+
+    assert rec.state == db.State(model="Model1", name="State1")
+
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state == db.State(model="Model1", name="State1")
+
+
+def test_unauthorized_final():
+    rec = db.Record().add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec.insert()
+
+    switch_to_test_user("normal")
+    rec.state = None
+    with pytest.raises(db.TransactionError) as exc:
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state == db.State(model="Model1", name="State1")
+
+    switch_to_test_user("team-leader")
+    rec.update()
+
+    assert rec.state is None
+
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state is None
+
+
+def test_unauthorized_initial():
+    rec = db.Record().add_parent("TestRT")
+    rec.insert()
+
+    switch_to_test_user("normal")
+    rec.state = db.State(model="Model1", name="State1")
+    with pytest.raises(db.TransactionError) as exc:
+        # normal user lacks the permission for the initial state
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state is None
+
+    switch_to_test_user("team-leader")
+    with pytest.raises(db.TransactionError) as exc:
+        # it is not allowed to "steal" the entity with the state feature
+        rec.update(sync=False)
+    assert "You are not allowed to do this." in str(exc.value)
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state is None
+
+    # we need to give ownership to "team-leader"
+    switch_to_admin_user()
+    rec_update = db.Record(id=rec.id).retrieve(flags={"ACL": None})
+    rec_update.acl.grant(role="team-leader", permission="EDIT:ACL")
+    rec_update.acl.grant(role="team-leader", permission="RETRIEVE:ACL")
+    rec_update.update_acl()
+
+    switch_to_test_user("team-leader")
+    rec.update(sync=False)
+    assert rec.state == db.State(model="Model1", name="State1")
+
+    rec_retrieve = db.Record(id=rec.id).retrieve()
+    assert rec_retrieve.state == db.State(model="Model1", name="State1")
+
+
+@pytest.mark.xfail(
+    reason="This is a very special corner case bug with low severity")
+def test_transitions_included_after_empty_update():
+    rec = db.Record()
+    rec.description = "old description"
+    rec.add_parent("TestRT")
+    rec.state = db.State(model="Model1", name="State1")
+    rec_insert = rec.insert(sync=False)
+
+    assert rec_insert.state.transitions is not None
+    assert rec_insert.state.transitions == {db.Transition(name="Transition1",
+                                                          from_state="State1",
+                                                          to_state="State2")}
+
+    rec_insert.description = "new_description"
+    rec_insert.state = db.State(model="Model1", name="State2")
+    rec_update = rec_insert.update(sync=False)
+    assert rec_update.state.transitions is not None
+    assert rec_update.state.transitions == {db.Transition(name="Transition2",
+                                                          from_state="State2",
+                                                          to_state="State3"),
+                                            db.Transition(name="Transition4",
+                                                          from_state="State2",
+                                                          to_state="State2")}
+
+    rec_update_2 = rec_update.update(sync=False)
+
+    # this fails
+    assert rec_update_2.state.transitions is not None
+    assert rec_update_2.state.transitions == {db.Transition(name="Transition1",
+                                                            from_state="State1",
+                                                            to_state="State2"),
+                                              db.Transition(name="Transition4",
+                                                            from_state="State2",
+                                                            to_state="State2")}