اختبار سير العمل غير المتزامن برصد محدود

allowed = {"queued": {"running", "cancelled"}, "running": {"succeeded", "failed"}, "succeeded": set(), "failed": set(), "cancelled": set()}
path = ["queued", "running", "succeeded"]
for before, after in zip(path, path[1:]):
    assert after in allowed[before]
assert "running" not in allowed["succeeded"]
print("workflow transition policy checked")

workflow transition policy checked

يفحص البرنامج الكامل مسارا بثلاث حالات وتراجعا محظورا واحدا. مجموعة الانتقالات سياسة مثال وليست دورة عامة للطوابير. لا يشغل معالجة خلفية أو تخزينا دائما.

def observe(states, limit):
    for state in states[:limit]:
        if state in {"succeeded", "failed", "cancelled"}:
            return state
    return "OBSERVATION_BUDGET_EXHAUSTED"

assert observe(["queued", "running", "succeeded"], 3) == "succeeded"
assert observe(["queued", "running", "failed"], 3) == "failed"
assert observe(["queued", "running"], 2) == "OBSERVATION_BUDGET_EXHAUSTED"
print("bounded observations checked")

bounded observations checked

يتخذ البرنامج الكامل ثلاثة قرارات حتمية على مصفوفات محلية. يعد الحد الملاحظات لا الثواني. لا يسبب تأخيرا ولا يدعي التزام وظيفة حقيقية بزمن أو توفر محدد.