24 lines
860 B
Python
24 lines
860 B
Python
from state_center.app.web_state_machine import build_timeline
|
|
|
|
|
|
def test_build_timeline_marks_skipped_and_waiting_user():
|
|
job = {
|
|
"job_id": "JOB-1",
|
|
"source_type": "YYC",
|
|
"current_node_code": "SOURCE_ADD",
|
|
"status": "WAITING_USER",
|
|
"needs_source_add": 1,
|
|
}
|
|
job_nodes = [
|
|
{"node_code": "ORDER_FETCH", "attempt_no": 1, "status": "SUCCESS"},
|
|
{"node_code": "PRODUCT_MATCH", "attempt_no": 1, "status": "SUCCESS"},
|
|
]
|
|
|
|
timeline = build_timeline(job, job_nodes)
|
|
by_code = {item["node_code"]: item for item in timeline}
|
|
|
|
assert by_code["ORDER_FETCH"]["display_status"] == "SUCCESS"
|
|
assert by_code["OCR_EXTRACT"]["display_status"] == "SKIPPED"
|
|
assert by_code["SOURCE_ADD"]["display_status"] == "WAITING_USER"
|
|
assert by_code["CONTRACT_CREATE"]["display_status"] == "PENDING"
|