From 6b83e18b538ce44f609f662c4cdf4ef9aa5154ef Mon Sep 17 00:00:00 2001
From: Timm Fitschen <t.fitschen@indiscale.com>
Date: Thu, 19 Nov 2020 19:30:07 +0100
Subject: [PATCH] WIP: fsm v0.2
---
src/caosdb/__init__.py | 2 +-
src/caosdb/common/models.py | 16 +++++
src/caosdb/common/state.py | 120 ++++++++++++++++++++++++++++++++++--
unittests/test_state.py | 53 +++++++++++++++-
4 files changed, 184 insertions(+), 7 deletions(-)
diff --git a/src/caosdb/__init__.py b/src/caosdb/__init__.py
index d59c60db..aed442e7 100644
--- a/src/caosdb/__init__.py
+++ b/src/caosdb/__init__.py
@@ -31,7 +31,7 @@ import caosdb.apiutils
from caosdb.common import administration
from caosdb.common.datatype import (BOOLEAN, DATETIME, DOUBLE, FILE, INTEGER,
REFERENCE, TEXT, LIST)
-from caosdb.common.state import State
+from caosdb.common.state import State, Transition
# Import of the basic API classes:
from caosdb.common.models import (ACL, ALL, FIX, NONE, OBLIGATORY, RECOMMENDED,
SUGGESTED, Container, DropOffBox, Entity,
diff --git a/src/caosdb/common/models.py b/src/caosdb/common/models.py
index eb625610..c3de01fa 100644
--- a/src/caosdb/common/models.py
+++ b/src/caosdb/common/models.py
@@ -3412,6 +3412,22 @@ class ACL():
self.deny(username=username, realm=realm, role=role,
permission=permission, priority=priority)
+ def combine(self, other):
+ """ Combine and return new instance."""
+ result = ACL()
+ result._grants.update(other._grants)
+ result._grants.update(self._grants)
+ result._denials.update(other._denials)
+ result._denials.update(self._denials)
+ result._priority_grants.update(other._priority_grants)
+ result._priority_grants.update(self._priority_grants)
+ result._priority_denials.update(other._priority_denials)
+ result._priority_denials.update(self._priority_denials)
+ return result
+
+ def __eq__(self, other):
+ return isinstance(other, ACL) and other._grants == self._grants and self._denials == other._denials and self._priority_grants == other._priority_grants and self._priority_denials == other._priority_denials
+
def is_empty(self):
return len(self._grants) + len(self._priority_grants) + \
len(self._priority_denials) + len(self._denials) == 0
diff --git a/src/caosdb/common/state.py b/src/caosdb/common/state.py
index eb81b343..de16c1ec 100644
--- a/src/caosdb/common/state.py
+++ b/src/caosdb/common/state.py
@@ -1,6 +1,81 @@
+import copy
from lxml import etree
+def _translate_to_state_acis(acis):
+ result = set()
+ for aci in acis:
+ aci = copy.copy(aci)
+ if aci.role:
+ aci.role = "?STATE?" + aci.role + "?"
+ result.add(aci)
+ return result
+
+
+class Transition:
+ """Transition
+
+ Represents allowed transitions from one state to another.
+
+ Properties
+ ----------
+ name : str
+ The name of the transition
+ description: str
+ The description of the transition
+ from_state : str
+ A state name
+ to_state : str
+ A state name
+ """
+
+ def __init__(self, name, from_state, to_state, description=None):
+ self._name = name
+ self._from_state = from_state
+ self._to_state = to_state
+ self._description = description
+
+ @property
+ def name(self):
+ return self._name
+
+ @property
+ def description(self):
+ return self._description
+
+ @property
+ def from_state(self):
+ return self._from_state
+
+ @property
+ def to_state(self):
+ return self._to_state
+
+ def __repr__(self):
+ return f'Transition(name="{self.name}", from_state="{self.from_state}", to_state="{self.to_state}", description="{self.description}")'
+
+ def __eq__(self, other):
+ return (isinstance(other, Transition)
+ and other.name == self.name
+ and other.to_state == self.to_state
+ and other.from_state == self.from_state)
+
+ def __hash__(self):
+ return 23472 + hash(self.name) + hash(self.from_state) + hash(self.to_state)
+
+ @staticmethod
+ def from_xml(xml):
+ to_state = [to.get("name") for to in xml
+ if to.tag.lower() == "tostate"]
+ from_state = [from_.get("name") for from_ in xml
+ if from_.tag.lower() == "fromstate"]
+ result = Transition(name=xml.get("name"),
+ description=xml.get("description"),
+ from_state=from_state[0] if from_state else None,
+ to_state=to_state[0] if to_state else None)
+ return result
+
+
class State:
"""State
@@ -15,16 +90,35 @@ class State:
Name of the State
model : str
Name of the StateModel
+ description : str
+ Description of the State (read-only)
+ id : str
+ Id of the undelying State record (read-only)
+ transitions : set of Transition
+ All transitions which are available from this state (read-only)
"""
def __init__(self, model, name):
self.name = name
self.model = model
+ self._id = None
+ self._description = None
+ self._transitions = None
+
+ @property
+ def id(self):
+ return self._id
+
+ @property
+ def description(self):
+ return self._description
+
+ @property
+ def transitions(self):
+ return self._transitions
def __eq__(self, other):
- return (other is not None
- and hasattr(other, "model")
- and hasattr(other, "name")
+ return (isinstance(other, State)
and self.name == other.name
and self.model == other.model)
@@ -62,4 +156,22 @@ class State:
"""
name = xml.get("name")
model = xml.get("model")
- return State(name=name, model=model)
+ result = State(name=name, model=model)
+ result._id = xml.get("id")
+ result._description = xml.get("description")
+ transitions = [Transition.from_xml(t) for t in xml if t.tag.lower() ==
+ "transition"]
+ if transitions:
+ result._transitions = set(transitions)
+
+ return result
+
+ @staticmethod
+ def create_state_acl(acl):
+ from .models import ACL
+ state_acl = ACL()
+ state_acl._grants = _translate_to_state_acis(acl._grants)
+ state_acl._denials = _translate_to_state_acis(acl._denials)
+ state_acl._priority_grants = _translate_to_state_acis(acl._priority_grants)
+ state_acl._priority_denials = _translate_to_state_acis(acl._priority_denials)
+ return state_acl
diff --git a/unittests/test_state.py b/unittests/test_state.py
index 37dc3c88..202c7a02 100644
--- a/unittests/test_state.py
+++ b/unittests/test_state.py
@@ -1,6 +1,7 @@
+import pytest
import caosdb as db
-from caosdb import State
-from caosdb.common.models import parse_xml
+from caosdb import State, Transition
+from caosdb.common.models import parse_xml, ACL
from lxml import etree
@@ -26,3 +27,51 @@ def test_entity_xml():
r = parse_xml(xml)
assert r.state == State(model="model1", name="state1")
+
+
+def test_description():
+ xml = b'<State name="state1" model="model1"/>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.description is None
+
+ with pytest.raises(AttributeError):
+ state.description = "test"
+
+ xml = b'<State name="state1" model="model1" description="test2"/>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.description == "test2"
+
+
+def test_id():
+ xml = b'<State name="state1" model="model1"/>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.id is None
+
+ with pytest.raises(AttributeError):
+ state.id = "2345"
+
+ xml = b'<State name="state1" model="model1" id="1234"/>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.id == "1234"
+
+
+def test_create_state_acl():
+ acl = ACL()
+ acl.grant(role="role1", permission="DO:IT")
+ acl.grant(role="?OWNER?", permission="DO:THAT")
+ state_acl = State.create_state_acl(acl)
+ assert state_acl.get_permissions_for_role("?STATE?role1?") == {"DO:IT"}
+ assert state_acl.get_permissions_for_role("?STATE??OWNER??") == {"DO:THAT"}
+
+
+def test_transitions():
+ xml = b'<State name="state1" model="model1"/>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.transitions is None
+
+ with pytest.raises(AttributeError):
+ state.transitions = []
+
+ xml = b'<State name="state1" model="model1" id="1234"><Transition name="t1"><FromState name="state1"/><ToState name="state2"/></Transition></State>'
+ state = State.from_xml(etree.fromstring(xml))
+ assert state.transitions == set([Transition(name="t1", from_state="state1", to_state="state2")])
--
GitLab