Testing with Python

Python tests use the Embedded engine’s server form over the REST API. The example uses only the standard library — unittest and urllib — so it runs anywhere Python 3 does.

The runnable project is testing/python in the examples download.

Start the engine

From the container image:

export TINYCONDUCTOR_TOKEN=test-token
docker run --rm -d --name tinyconductor-test -p 127.0.0.1:8080:8080 \
  -e CONDUCTOR_AUTH_CREDENTIALS_JSON='[{"token":"test-token","subject":"tests","tenantId":"default","kind":"service","actions":["*"]}]' \
  registry.tinyfactory.ai/tinyblox/tinyconductor:0.1.0 --mode embedded

Or from the release archive for your platform (downloads are listed in Embedded mode); on loopback it needs no credential:

./tinyconductor-0.1.0-linux-x86_64/tinyconductor --mode embedded

The test reads TINYCONDUCTOR_URL (default http://127.0.0.1:8080) and the optional TINYCONDUCTOR_TOKEN. The REST calls are listed in the Java guide.

The test

class OrderProcessTest(unittest.TestCase):
    def setUp(self):
        # Every test starts from an empty engine at the fixed epoch.
        call("POST", "/v1/embedded/reset")
        deploy("order.form", "application/json")
        deploy("order-decisions.dmn", "application/xml")
        deploy("order.bpmn", "application/xml")

    def test_order_completes_after_review_period(self):
        instance = call("POST", "/v1/process-instances", {
            "processDefinitionId": "order-process",
            "variables": {"orderId": 4711, "amount": 99.9},
        })
        key = instance["processInstanceKey"]  # keys are JSON strings

        complete_job("charge-card", {"charged": True})
        # The 24-hour review timer fires inside this call; no sleeping or polling.
        call("POST", "/v1/embedded/clock/advance-by", {"milliseconds": 86_400_000})
        complete_job("send-invoice", {})

        tasks = call("POST", "/v1/user-tasks/search", {"processInstanceKey": key})["items"]
        self.assertEqual([t["elementId"] for t in tasks], ["confirm-shipment"])
        call("POST", f"/v1/user-tasks/{tasks[0]['userTaskKey']}/completion", {"variables": {"shipped": True}})

        records = call("GET", "/v1/embedded/records")["records"]
        completed = {
            r["value"]["elementId"]
            for r in records
            if r["valueType"] == "PROCESS_INSTANCE"
            and r["intent"] == "ELEMENT_COMPLETED"
            and r["value"]["processInstanceKey"] == key
        }
        self.assertIn("order-process", completed)  # the process itself completed
        for element in ("charge-card", "send-invoice", "confirm-shipment"):
            self.assertIn(element, completed)

call sends JSON with urllib.request, deploy builds the multipart body by hand and complete_job activates one job and completes it; all three are in test_order_process.py. Run it:

cd tinyconductor-0.1.0-examples/testing/python
python3 -m unittest -v

To host the engine inside the Python process instead, load the C ABI with ctypes; see C and other FFI languages.