68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
from workflow.models import PlannedGoalStep, WorkflowDefinition, WorkflowStepResult
|
|
from workflow.store import WorkflowStore
|
|
|
|
|
|
def _definition() -> WorkflowDefinition:
|
|
return WorkflowDefinition(
|
|
name="linear",
|
|
entry_step_id="first",
|
|
steps=[
|
|
PlannedGoalStep("first", "one", next_step_id="second"),
|
|
PlannedGoalStep("second", "two"),
|
|
],
|
|
)
|
|
|
|
|
|
def test_workflow_store_round_trips_definition_and_run(tmp_path) -> None:
|
|
store = WorkflowStore(tmp_path / "workflows.sqlite3")
|
|
definition = _definition()
|
|
|
|
store.save_definition(definition)
|
|
run = store.create_run(
|
|
definition.id,
|
|
{"contact": "Zhang San"},
|
|
device_id="phone",
|
|
)
|
|
loaded = store.get_run(run.id)
|
|
|
|
assert store.get_definition(definition.id) == definition
|
|
assert loaded is not None
|
|
assert loaded.status == "running"
|
|
assert loaded.current_step_id == "first"
|
|
assert loaded.variables == {"contact": "Zhang San"}
|
|
assert loaded.device_id == "phone"
|
|
|
|
|
|
def test_workflow_store_appends_step_results_in_order(tmp_path) -> None:
|
|
store = WorkflowStore(tmp_path / "workflows.sqlite3")
|
|
definition = _definition()
|
|
store.save_definition(definition)
|
|
run = store.create_run(definition.id, {})
|
|
|
|
store.append_step_result(run.id, WorkflowStepResult("first", "planned_goal", True))
|
|
store.append_step_result(run.id, WorkflowStepResult("second", "planned_goal", True))
|
|
|
|
loaded = store.get_run(run.id)
|
|
assert loaded is not None
|
|
assert [result.step_id for result in loaded.step_results] == ["first", "second"]
|
|
|
|
|
|
def test_workflow_store_survives_reopen(tmp_path) -> None:
|
|
db_path = tmp_path / "workflows.sqlite3"
|
|
first_store = WorkflowStore(db_path)
|
|
definition = _definition()
|
|
first_store.save_definition(definition)
|
|
run = first_store.create_run(definition.id, {"x": 1})
|
|
first_store.append_step_result(run.id, WorkflowStepResult("first", "planned_goal", True))
|
|
first_store.update_run(run.id, current_step_id="second", variables={"x": 2})
|
|
|
|
reopened = WorkflowStore(db_path)
|
|
loaded = reopened.get_run(run.id)
|
|
|
|
assert loaded is not None
|
|
assert loaded.current_step_id == "second"
|
|
assert loaded.variables == {"x": 2}
|
|
assert [result.step_id for result in loaded.step_results] == ["first"]
|