generated from coulomb/repo-seed
95 lines
2.4 KiB
Python
95 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssertionDef:
|
|
id: str
|
|
target: str
|
|
op: str
|
|
value: Any = None
|
|
description: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "AssertionDef":
|
|
return cls(
|
|
id=str(data["id"]),
|
|
target=str(data.get("target", "")),
|
|
op=str(data["op"]),
|
|
value=data.get("value"),
|
|
description=str(data.get("description", "")),
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkstationDef:
|
|
name: str
|
|
entry_assertions: list[AssertionDef] = field(default_factory=list)
|
|
exit_assertions: list[AssertionDef] = field(default_factory=list)
|
|
description: str = ""
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "WorkstationDef":
|
|
return cls(
|
|
name=str(data["name"]),
|
|
description=str(data.get("description", "")),
|
|
entry_assertions=[
|
|
AssertionDef.from_dict(item)
|
|
for item in data.get("entry_assertions", []) or []
|
|
],
|
|
exit_assertions=[
|
|
AssertionDef.from_dict(item)
|
|
for item in data.get("exit_assertions", []) or []
|
|
],
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FlowDef:
|
|
id: str
|
|
entity_type: str
|
|
workstations: list[WorkstationDef]
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "FlowDef":
|
|
return cls(
|
|
id=str(data["id"]),
|
|
entity_type=str(data["entity_type"]),
|
|
workstations=[
|
|
WorkstationDef.from_dict(item)
|
|
for item in data.get("workstations", []) or []
|
|
],
|
|
)
|
|
|
|
def workstation(self, name: str) -> WorkstationDef | None:
|
|
return next((item for item in self.workstations if item.name == name), None)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AssertionResult:
|
|
id: str
|
|
passed: bool
|
|
target: str
|
|
op: str
|
|
expected: Any = None
|
|
actual: list[Any] = field(default_factory=list)
|
|
description: str = ""
|
|
reason: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UnreachableWorkstation:
|
|
workstation: str
|
|
blocking: AssertionResult
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FlowResult:
|
|
current_workstation: str
|
|
exit_blocked: bool
|
|
blocking_assertions: list[AssertionResult]
|
|
reachable: list[str]
|
|
unreachable: list[UnreachableWorkstation]
|