Skip to content
Snippets Groups Projects
Verified Commit a0f4c836 authored by Timm Fitschen's avatar Timm Fitschen
Browse files

STY: autopep8'ed test_state.py

parent 704adae7
No related branches found
No related tags found
1 merge request!3F fsm
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
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment