diff --git a/tests/test_state.py b/tests/test_state.py
index ab734071b69e6578633fdbb8827a668ba078aac2..490b091f9d6c1585cb80d4d288a35e1f88e0ebb2 100644
--- a/tests/test_state.py
+++ b/tests/test_state.py
@@ -1,11 +1,13 @@
import pytest
import caosdb as db
+
def teardown_module():
d = db.execute_query("FIND ENTITY WITH ID > 99")
if len(d) > 0:
d.delete(flags={"forceFinalState": "true"})
+
def setup_module():
teardown_module()
db.RecordType("State").insert()
@@ -20,25 +22,43 @@ def setup_module():
db.Record("State2").add_parent("State").insert()
db.Record("State3").add_parent("State").insert()
# 1->
- db.Record("Transition1").add_parent("Transition").add_property("from", "State1").add_property("to", "State2").insert()
+ 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()
+ 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()
+ 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("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()
- 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():
d = db.execute_query("FIND TestRT")
if len(d) > 0:
d.delete(flags={"forceFinalState": "true"})
+
def setup():
teardown()
db.RecordType("TestRT").insert()
+
def test_state_message():
rec = db.Record()
rec.add_parent("TestRT")
@@ -53,6 +73,7 @@ def test_state_message():
assert rec_retrieve.get_property("State") is None
assert rec_retrieve.state == rec.state
+
def test_state_query():
rec = db.Record()
rec.add_parent("TestRT")
@@ -60,7 +81,10 @@ def test_state_query():
rec.insert()
assert rec.get_property("State") is None
- assert db.execute_query("FIND TestRT WITH State = State1", unique=True).id == rec.id
+ assert db.execute_query(
+ "FIND TestRT WITH State = State1",
+ unique=True).id == rec.id
+
def test_state_transition():
rec = db.Record()
@@ -74,6 +98,7 @@ def test_state_transition():
rec_retrieve = db.Record.retrieve(rec.id)
assert rec_retrieve.state == rec_update.state
+
def test_transition_not_allowed():
"""unallowed transitions return errors and do not update the entity"""
rec = db.Record()
@@ -83,14 +108,17 @@ def test_transition_not_allowed():
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)
+ 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."
+ 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()
@@ -102,6 +130,7 @@ def test_wrong_initial():
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()
@@ -124,6 +153,7 @@ def test_wrong_final():
rec.update()
rec.delete()
+
def test_multiple_states():
"""currently, only one state is allowed"""
rec = db.Record()
@@ -131,10 +161,12 @@ def test_multiple_states():
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()
@@ -142,6 +174,7 @@ def test_multiple_states():
rec_insert = rec.insert(sync=False)
print(rec_insert)
+
def test_broken_state_missing_model():
rec = db.Record()
rec.add_parent("TestRT")
@@ -151,6 +184,7 @@ def test_broken_state_missing_model():
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")
@@ -160,6 +194,7 @@ def test_broken_state_missing_state_name():
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")
@@ -167,7 +202,9 @@ def test_state_not_in_state_model():
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."
+ assert rec.get_errors()[
+ 0].description == "State does not exist in this StateModel."
+
def test_transition_with_out_state_change():
rec = db.Record()
@@ -177,12 +214,14 @@ def test_transition_with_out_state_change():
rec_insert = rec.insert(sync=False)
# first update attempt (should fail, because 1->1 not allowed)
- rec_insert.description="updated description 1"
+ 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)
+ 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."
+ assert rec_update.get_errors(
+ )[0].description == "Transition not allowed."
db.common.models.raise_errors(rec_update)
# second update with transition to state2