From a30d06dc44909e12c52b134536794b4af1a704b3 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:18:12 +0000 Subject: [PATCH 1/3] test: add published workflow identity and cold successor consumer --- .../workflow_identity_consumer/README.md | 22 +++ .../workflow_identity_consumer/compose.yml | 68 +++++++ .../workflow_identity_consumer/consumer.py | 166 ++++++++++++++++++ 3 files changed, 256 insertions(+) create mode 100644 tests/fixtures/workflow_identity_consumer/README.md create mode 100644 tests/fixtures/workflow_identity_consumer/compose.yml create mode 100644 tests/fixtures/workflow_identity_consumer/consumer.py diff --git a/tests/fixtures/workflow_identity_consumer/README.md b/tests/fixtures/workflow_identity_consumer/README.md new file mode 100644 index 0000000..f53f04d --- /dev/null +++ b/tests/fixtures/workflow_identity_consumer/README.md @@ -0,0 +1,22 @@ +# Published workflow identity consumer + +This fixture normally installs the selected published Python SDK and compares +its installed source with the exact immutable release. Set the required Compose +environment values, use a uniquely named project and a disposable data/state +directory, then start the stack. + +Run `/opt/venv/bin/python /fixture/consumer.py execute` in the client container, +followed by the same command with `cold` in a separate process. Every original +has a thirty-second whole observer and thirty-second run/execution deadlines. +The first workflow node, activity node and cold workflow node use separate +Worker instances and registrations. The final cold client verifies original +typed results and unchanged retained histories. + +The eight originals cover default/named namespaces, literal numeric workflow +IDs, activity suspension/cold replay, and continue-as-new successor identity. +Continue-as-new result inspection explicitly pins the successor run; it does +not treat an original run handle as a promise to follow a chain. + +Stop and audit the API/database, then remove the uniquely named project and all +task-owned data, dependencies and raw output. The fixture does not qualify any +native runtime or a broader SDK contract. diff --git a/tests/fixtures/workflow_identity_consumer/compose.yml b/tests/fixtures/workflow_identity_consumer/compose.yml new file mode 100644 index 0000000..290d06c --- /dev/null +++ b/tests/fixtures/workflow_identity_consumer/compose.yml @@ -0,0 +1,68 @@ +services: + bootstrap: + image: ${IDENTITY_SERVER_IMAGE:?Set the immutable published image} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + command: [server-bootstrap] + environment: &server_environment + DB_CONNECTION: sqlite + DB_DATABASE: /data/database.sqlite + DB_SQLITE_JOURNAL_MODE: DELETE + CACHE_STORE: file + QUEUE_CONNECTION: database + DW_MODE: service + DW_POLLING_CACHE_PATH: /data/poll-cache + DW_AUTH_DRIVER: token + DW_AUTH_TOKEN: identity-client + DW_ADMIN_TOKEN: identity-admin + DW_WORKER_TOKEN: identity-worker + DW_BOOTSTRAP_RETRIES: 1 + volumes: &server_volumes + - ${IDENTITY_PHASE_ROOT:?Set the disposable phase directory}/data:/data + + api: + image: ${IDENTITY_SERVER_IMAGE:?} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: *server_environment + volumes: *server_volumes + depends_on: + bootstrap: + condition: service_completed_successfully + + client: + image: ${IDENTITY_CLIENT_IMAGE:?Set the existing pinned Python image} + user: "1000:1000" + init: true + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: + HOME: /tmp + SERVER_URL: http://api:8080 + IDENTITY_COHORT: ${IDENTITY_COHORT:?Set a new cohort} + IDENTITY_SDK_VERSION: ${IDENTITY_SDK_VERSION:?Set the published version} + entrypoint: [sh, -ec] + command: + - | + python -m venv /opt/venv + /opt/venv/bin/pip install --index-url https://pypi.org/simple --no-cache-dir --quiet durable-workflow=="$$IDENTITY_SDK_VERSION" + exec sleep infinity + tmpfs: + - /opt/venv:rw,exec,uid=1000,gid=1000,size=268435456 + healthcheck: + test: [CMD, /opt/venv/bin/python, -c, "import durable_workflow"] + interval: 2s + timeout: 5s + retries: 30 + volumes: + - ${IDENTITY_PHASE_ROOT:?}/state:/state + - ${IDENTITY_EXPECTED_SOURCE:?Set the exact tagged package source}:/expected-source:ro + - ./consumer.py:/fixture/consumer.py:ro + depends_on: + api: + condition: service_healthy diff --git a/tests/fixtures/workflow_identity_consumer/consumer.py b/tests/fixtures/workflow_identity_consumer/consumer.py new file mode 100644 index 0000000..fd92c05 --- /dev/null +++ b/tests/fixtures/workflow_identity_consumer/consumer.py @@ -0,0 +1,166 @@ +"""Normally installed identity acceptance against a real published Server.""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import importlib.metadata +import json +import os +import sys +import time +from contextlib import asynccontextmanager +from pathlib import Path + +import durable_workflow +from durable_workflow import Client, Worker, activity, workflow + +STATE = Path('/state') +URL = os.environ['SERVER_URL'] +COHORT = os.environ['IDENTITY_COHORT'] + + +def verify_install(): + dist = importlib.metadata.distribution('durable-workflow') + assert dist.version == os.environ['IDENTITY_SDK_VERSION'] + assert '/opt/venv/' in str(Path(durable_workflow.__file__).resolve()) + hashes = [] + for entry in dist.files or []: + if entry.hash is None: + continue + assert entry.hash.mode == 'sha256' + path = Path(dist.locate_file(entry)) + digest = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()).decode().rstrip('=') + assert digest == entry.hash.value, str(entry) + hashes.append(str(entry)) + sources = list(Path('/expected-source').rglob('*.py')) + installed = Path(durable_workflow.__file__).parent + for source in sources: + assert source.read_bytes() == (installed / source.relative_to('/expected-source')).read_bytes(), str(source) + return {'version': dist.version, 'hashed_record_entries': len(hashes), 'matched_source_files': len(sources)} + + +@activity.defn(name='qualification.identity-echo') +def echo(value): + return value + + +@workflow.defn(name='qualification.workflow-identity') +class IdentityWorkflow: + def run(self, ctx, again=False, prior=None): + if again: + yield ctx.continue_as_new(False, [ctx.workflow_id, ctx.run_id]) + return None + before = [ctx.workflow_id, ctx.run_id] + value = yield ctx.schedule_activity('qualification.identity-echo', [before + [9007199254740993, None, 'λ']]) + return {'before': before, 'after': [ctx.workflow_id, ctx.run_id], 'prior': prior, 'echo': value} + + +class ObservedClient(Client): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.registered = asyncio.Event() + + async def register_worker(self, **kwargs): + reply = await super().register_worker(**kwargs) + self.registered.set() + return reply + + +@asynccontextmanager +async def node(namespace, queue, role): + async with ObservedClient(URL, token='identity-worker', namespace=namespace) as client: + worker = Worker( + client, task_queue=queue, + workflows=[IdentityWorkflow] if role != 'activity' else [], + activities=[echo] if role == 'activity' else [], + worker_id=f'{COHORT}-{namespace}-{queue}-{role}', + poll_timeout=1, heartbeat_interval=1, + ) + task = asyncio.create_task(worker.run()) + try: + await asyncio.wait_for(client.registered.wait(), 8) + yield + finally: + await worker.stop() + await asyncio.wait_for(task, 5) + + +async def wait_event(client, workflow_id, event_type): + while True: + desc = await client.describe_workflow(workflow_id) + history = await client.get_history(workflow_id, desc.run_id) + if any(event['event_type'] == event_type for event in history['events']): + return desc.run_id, history['events'] + await asyncio.sleep(0.1) + + +async def original(namespace, workflow_id, again): + started = time.monotonic() + queue = 'identity-' + workflow_id + async with Client(URL, token='identity-client', namespace=namespace) as client: + handle = await client.start_workflow( + workflow_type='qualification.workflow-identity', task_queue=queue, + workflow_id=workflow_id, input=[again], + run_timeout_seconds=30, execution_timeout_seconds=30, + ) + original_run_id = handle.run_id + async with node(namespace, queue, 'first'): + final_run_id, _ = await wait_event(client, workflow_id, 'ActivityScheduled') + async with node(namespace, queue, 'activity'): + await wait_event(client, workflow_id, 'ActivityCompleted') + async with node(namespace, queue, 'cold'): + result = await client.get_workflow_handle(workflow_id, run_id=final_run_id).result(timeout=20) + expected = [workflow_id, final_run_id] + assert result == { + 'before': expected, 'after': expected, + 'prior': [workflow_id, original_run_id] if again else None, + 'echo': expected + [9007199254740993, None, 'λ'], + }, result + assert (final_run_id != original_run_id) is again + original_history = (await client.get_history(workflow_id, original_run_id))['events'] + assert any(e['event_type'] == 'WorkflowContinuedAsNew' for e in original_history) is again + history = (await client.get_history(workflow_id, final_run_id))['events'] + assert sum(e['event_type'] == 'ActivityStarted' for e in history) == 1 + assert sum(e['event_type'] == 'WorkflowCompleted' for e in history) == 1 + elapsed = time.monotonic() - started + assert elapsed < 30 + return {'namespace': namespace, 'workflow_id': workflow_id, 'original_run_id': original_run_id, + 'run_id': final_run_id, 'continued': again, 'result': result, 'elapsed_seconds': elapsed, + 'history': history, 'original_history': original_history} + + +async def main(): + installation = verify_install() + stage = sys.argv[1] + path = STATE / 'originals.json' + if stage == 'execute': + async with Client(URL, token='identity-admin') as admin: + await admin.create_namespace(name='identity-named') + originals = [] + path.write_text('[]') + for namespace in ['default', 'identity-named']: + prefix = '' if namespace == 'default' else 'named-' + for suffix, again in [('0', False), ('17', False), ('017', False), ('continue', True)]: + result = await asyncio.wait_for(original(namespace, prefix + suffix, again), 30) + originals.append(result) + path.write_text(json.dumps(originals, ensure_ascii=False, indent=2)) + print(json.dumps({'workflow_id': result['workflow_id'], 'namespace': namespace, + 'elapsed_seconds': result['elapsed_seconds'], 'passed': True}), flush=True) + elif stage == 'cold': + originals = json.loads(path.read_text()) + assert len(originals) == 8 + for original in originals: + async with Client(URL, token='identity-client', namespace=original['namespace']) as client: + handle = client.get_workflow_handle(original['workflow_id'], run_id=original['run_id']) + assert await handle.result(timeout=5) == original['result'] + assert (await handle.get_history())['events'] == original['history'] + assert (await client.get_history(original['workflow_id'], original['original_run_id']))['events'] == original['original_history'] + else: + raise ValueError('Choose execute or cold.') + assert verify_install() == installation + print(json.dumps({'stage': stage, 'installation': installation, 'originals': len(originals)}), flush=True) + + +asyncio.run(main()) From bbabc6452bf05086784095277c2e75c6f7d8a3e0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:29:21 +0000 Subject: [PATCH 2/3] test: preserve identity producer callable in cold reader --- .../fixtures/workflow_identity_consumer/consumer.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/fixtures/workflow_identity_consumer/consumer.py b/tests/fixtures/workflow_identity_consumer/consumer.py index fd92c05..bf03faf 100644 --- a/tests/fixtures/workflow_identity_consumer/consumer.py +++ b/tests/fixtures/workflow_identity_consumer/consumer.py @@ -151,12 +151,12 @@ async def main(): elif stage == 'cold': originals = json.loads(path.read_text()) assert len(originals) == 8 - for original in originals: - async with Client(URL, token='identity-client', namespace=original['namespace']) as client: - handle = client.get_workflow_handle(original['workflow_id'], run_id=original['run_id']) - assert await handle.result(timeout=5) == original['result'] - assert (await handle.get_history())['events'] == original['history'] - assert (await client.get_history(original['workflow_id'], original['original_run_id']))['events'] == original['original_history'] + for record in originals: + async with Client(URL, token='identity-client', namespace=record['namespace']) as client: + handle = client.get_workflow_handle(record['workflow_id'], run_id=record['run_id']) + assert await handle.result(timeout=5) == record['result'] + assert (await handle.get_history())['events'] == record['history'] + assert (await client.get_history(record['workflow_id'], record['original_run_id']))['events'] == record['original_history'] else: raise ValueError('Choose execute or cold.') assert verify_install() == installation From 2ca597bbbf56dab15265306d6f467404a8437506 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 22 Oct 2026 08:37:13 +0000 Subject: [PATCH 3/3] style: wrap cold identity history assertion --- tests/fixtures/workflow_identity_consumer/consumer.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/fixtures/workflow_identity_consumer/consumer.py b/tests/fixtures/workflow_identity_consumer/consumer.py index bf03faf..bea8ae9 100644 --- a/tests/fixtures/workflow_identity_consumer/consumer.py +++ b/tests/fixtures/workflow_identity_consumer/consumer.py @@ -156,7 +156,9 @@ async def main(): handle = client.get_workflow_handle(record['workflow_id'], run_id=record['run_id']) assert await handle.result(timeout=5) == record['result'] assert (await handle.get_history())['events'] == record['history'] - assert (await client.get_history(record['workflow_id'], record['original_run_id']))['events'] == record['original_history'] + assert (await client.get_history( + record['workflow_id'], record['original_run_id'] + ))['events'] == record['original_history'] else: raise ValueError('Choose execute or cold.') assert verify_install() == installation