diff --git a/tests/fixtures/entry_recovery_http/README.md b/tests/fixtures/entry_recovery_http/README.md new file mode 100644 index 0000000..9bc4bc1 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/README.md @@ -0,0 +1,89 @@ +# Published entry-recovery Waterline fixture + +This revision uses normally installed Python SDK 2.8.0, published Server 2.10.1 +and published Waterline 2.12.1. It extends the earlier HTTP fixture below with +real browser interaction; no route or response is mocked. Run only +`required-before` and `required-after` for this two-original UI cohort. The first +uses desktop/dark and the second mobile/light. The real selected-run Repair +action replaces the fixture's direct repair request. The original deadlines and +same-run assertions are unchanged. The stopped-state audit expects two completed +originals, two deliberate repairs and no registered workers. + +Use `mcr.microsoft.com/playwright/python:v1.55.0-noble` for the client/browser +and install `requirements.txt` normally into `/opt/venv`. The published SDK wheel +SHA-256 is `908267666c35890fce13ab9c8e62a5c93abac78bdffbb0e56423eb3b3e57a77b`. +Mount these fixture files individually under `/phase`, with only the disposable +`data` and `consumer` directories writable. Keep SDK source off `PYTHONPATH`. +The packaged viewer resolves as `http://viewer:8080/waterline`, uses the same +synthetic API credentials and must be configured with +`WATERLINE_ACCESS_MODE=operator`. Require both services and the browser to be +ready before admitting either original. Preserve concise failed observations; +retire images pulled solely for the phase, browsers, dependencies and all raw +state immediately after review. + +## Historical HTTP cohort + +This is the five-original installed-package acceptance fixture for +[shared issue 221](https://github.com/durable-workflow/.github/issues/221). +It uses a normal PyPI installation, not this checkout's runtime code. + +The recorded cohort uses Python SDK 2.6.2 (`1b03ed2a`) and Server 2.9.9, +image index `sha256:d45d197e9556292ec096531030717e43d69ea4ea9ab197170971492f14cc0869`. +That Server image is historical: use synthetic data on an isolated internal +Docker network without published host ports when reproducing this exact tuple. +Production deployments should use a current patched Server release. + +## Preparation + +Copy these files into a disposable directory, mounted as `/phase` in the client, +and create its `data` and `consumer` subdirectories. The API resolves as +`http://api:8080` on their private network. Mount only `data` at `/data` in the +published API image, preserving its packaged bootstrap/seeder files. + +Bootstrap a fresh SQLite database with `DB_CONNECTION=sqlite`, +`DB_DATABASE=/data/database.sqlite`, `CACHE_STORE=file`, +`QUEUE_CONNECTION=database`, `DW_POLLING_CACHE_PATH=/data/poll-cache`, +`DW_AUTH_TOKEN=general221-fixture`, `DW_ADMIN_TOKEN=general221-fixture`, and +`DW_WORKER_TOKEN=general221-worker`. These are synthetic fixture credentials. +Run the image's `server-bootstrap` and its normal Apache entry point. +Require `/api/ready` to return 200 before business admission. + +Install `requirements.txt` into `/opt/venv` in the client. Save the PyPI wheel +as `/phase/sdk.whl`; its SHA-256 is +`28379c16bcd598c9aa545071d074bee297e449158a550e580d1df0ef9fc27ab8`. +Run `verify_install.py` before and after the cohort. It checks the installed +production files against both their RECORD hashes and the published wheel. +Do not put an SDK checkout on `PYTHONPATH`. An executable tmpfs is needed if +the venv, including extension modules, lives in tmpfs. + +## Execute and inspect + +For each of `required-before`, `required-after`, `optional`, `body-before`, and +`body-after`, use a fresh cohort identifier and run: + +```sh +EXPECTED_BINDING_REFUSAL=1 /opt/venv/bin/python /phase/consumer.py CASE COHORT +``` + +Every original retains a 30-second run and 60-second execution deadline, and +each observer has its original 30-second total bound. The two changed required +signatures must refuse without entering the body. Supported run-scoped repair +and compatible code must finish the same original with unchanged history +prefix and deadlines. The optional argument completes normally; genuine body +TypeErrors remain terminal business failures. The fixture records every +business Worker exit and actual task submission, including failure reason. + +For each completed original, start a separate process with +`/opt/venv/bin/python /phase/cold.py WORKFLOW_ID RUN_ID`. Its result must retain +the exact signed-int64 maximum, null and Unicode values. + +Stop the API before SQL inspection. Run `audit.py` with the entire phase mounted +kernel read-only at `/phase`, a disposable writable `/audit`, no network and no +SDK dependency. It copies the stopped database/WAL, opens that copy read-only, +checks five originals and two deliberate repairs, and requires source and +copied database/WAL bytes to stay exact. Only a transient SQLite index may be +created in the disposable copy. Its report is written to stdout. + +Retain concise results and source identities in the owning issue. Remove all +owned containers, network, dependencies, database, wheel and raw traces after +inspection; preserve borrowed images and other owners' resources. diff --git a/tests/fixtures/entry_recovery_http/audit.py b/tests/fixtures/entry_recovery_http/audit.py new file mode 100644 index 0000000..458e8d5 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/audit.py @@ -0,0 +1,89 @@ +import datetime +import hashlib +import json +import os +import shutil +import sqlite3 +from pathlib import Path + +root = Path('/phase') +path = root / 'data/database.sqlite' +files = [p for p in (path, Path(str(path) + '-wal')) if p.exists()] +hashes = {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in files} +copy = Path('/audit') +for source in files: + shutil.copyfile(source, copy / source.name) +db = sqlite3.connect('file:' + str(copy / path.name) + '?mode=ro', uri=True) +db.row_factory = sqlite3.Row +db.execute('PRAGMA query_only=ON') +results = [] +for evidence_path in sorted((root / 'consumer').glob('*/evidence.json')): + evidence = json.loads(evidence_path.read_text()) + run_id = evidence['after']['run_id'] + run = dict(db.execute('SELECT * FROM workflow_runs WHERE id=?', (run_id,)).fetchone()) + tasks = [dict(row) for row in db.execute( + 'SELECT id,status,attempt_count,repair_count FROM workflow_tasks WHERE workflow_run_id=? ORDER BY id', + (run_id,), + )] + history = [dict(row) for row in db.execute( + 'SELECT sequence,event_type FROM workflow_history_events WHERE workflow_run_id=? ORDER BY sequence', (run_id,), + )] + assert run['status'] == evidence['after']['status'] + assert [(r['sequence'], r['event_type']) for r in history] == [ + (r['sequence'], r['event_type']) for r in evidence['history']['events'] + ] + assert evidence['prefix']['events'] == evidence['history']['events'][:len(evidence['prefix']['events'])] + started = datetime.datetime.fromisoformat(run['started_at']) + assert (datetime.datetime.fromisoformat(run['run_deadline_at']) - started).total_seconds() == 30 + assert (datetime.datetime.fromisoformat(run['execution_deadline_at']) - started).total_seconds() == 60 + assert started.microsecond != 0 + if run['status'] == 'completed': + assert all(task['status'] == 'completed' for task in tasks), tasks + else: + assert all(task['status'] in ('completed', 'failed') for task in tasks), tasks + assert sum(row['event_type'] == 'WorkflowFailed' for row in history) == 1 + repaired = evidence.get('refusal') is not None + assert sum(task['repair_count'] for task in tasks) == int(repaired), tasks + assert sum(task['attempt_count'] - 1 for task in tasks) == int(repaired), tasks + assert sum(row['event_type'] == 'WorkflowCompleted' for row in history) == int(run['status'] == 'completed') + if repaired: + assert not evidence['refusal']['is_terminal'] + assert not any(row['event_type'] == 'WorkflowFailed' for row in history) + assert all(code == 0 for code in evidence['worker_exits']) + results.append({'case': evidence['case'], 'cohort': evidence['cohort'], 'run_id': run_id, + 'status': run['status'], 'tasks': len(tasks), 'repairs': int(repaired), + 'events': len(history), 'worker_exits': evidence['worker_exits'], + 'run_deadline_at': run['run_deadline_at'], 'execution_deadline_at': run['execution_deadline_at']}) +assert len(results) == 2, results +failed_originals = [] +for original_path in sorted((root / 'consumer').glob('*/original.json')): + if (original_path.parent / 'evidence.json').exists(): + continue + original = json.loads(original_path.read_text()) + run_id = original['before']['run_id'] + run = db.execute('SELECT status FROM workflow_runs WHERE id=?', (run_id,)).fetchone() + tasks = [dict(row) for row in db.execute( + 'SELECT status,attempt_count,repair_count FROM workflow_tasks WHERE workflow_run_id=?', (run_id,), + )] + history = [row[0] for row in db.execute( + 'SELECT event_type FROM workflow_history_events WHERE workflow_run_id=? ORDER BY sequence', (run_id,), + )] + assert run['status'] == 'pending', (run_id, dict(run)) + assert tasks == [{'status': 'failed', 'attempt_count': 1, 'repair_count': 0}], tasks + assert history == ['StartAccepted', 'WorkflowStarted'], history + assert (original_path.parent / 'waterline-failure.txt').exists() + failed_originals.append({'run_id': run_id, 'status': run['status'], 'repairs': 0, 'events': len(history)}) +assert db.execute('SELECT COUNT(*) FROM workflow_runs').fetchone()[0] == len(results) + len(failed_originals) +assert sum(row['status'] == 'completed' for row in results) == 2 +assert sum(row['status'] == 'failed' for row in results) == 0 +assert db.execute('SELECT COUNT(*) FROM workflow_worker_registrations').fetchone()[0] == 0 +db.close() +assert hashes == {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in files} +assert hashes == {p.name: hashlib.sha256((copy / p.name).read_bytes()).hexdigest() for p in files} +print(json.dumps({'sdk': '2.8.0', 'server': '2.10.1', 'waterline': '2.12.1', 'cases': results, + 'waterline_asset_source': os.environ.get('WATERLINE_ASSET_SOURCE', 'published'), + 'failed_originals': failed_originals, + 'events': sum(r['events'] for r in results), + 'tasks': sum(r['tasks'] for r in results), 'repairs': sum(r['repairs'] for r in results), + 'workers': sum(len(r['worker_exits']) for r in results), 'database_and_wal_unchanged': True, + 'source_hashes': hashes}, indent=2)) diff --git a/tests/fixtures/entry_recovery_http/cold.py b/tests/fixtures/entry_recovery_http/cold.py new file mode 100644 index 0000000..2902640 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/cold.py @@ -0,0 +1,17 @@ +import asyncio +import json +import sys + +from durable_workflow import Client + + +async def main(): + workflow_id, run_id = sys.argv[1:] + async with Client('http://api:8080', token='general221-fixture') as client: + handle = client.get_workflow_handle(workflow_id, run_id=run_id, workflow_type='general221.entry') + result = await handle.result(timeout=3) + assert result == {'amount': 7, 'typed': [9223372036854775807, None, 'Київ']}, result + print(json.dumps({'run_id': run_id, 'typed_result': 'exact'})) + + +asyncio.run(main()) diff --git a/tests/fixtures/entry_recovery_http/compose.yml b/tests/fixtures/entry_recovery_http/compose.yml new file mode 100644 index 0000000..9f05ed8 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/compose.yml @@ -0,0 +1,92 @@ +services: + bootstrap: + image: ${ENTRY_SERVER_IMAGE:?Set the immutable published Server 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_POLLING_CACHE_PATH: /data/poll-cache + DW_AUTH_DRIVER: token + DW_AUTH_TOKEN: general221-fixture + DW_ADMIN_TOKEN: general221-fixture + DW_WORKER_TOKEN: general221-worker + volumes: + - ${ENTRY_PHASE_ROOT:?Set a disposable phase directory}/data:/data + + api: + image: ${ENTRY_SERVER_IMAGE:?} + user: "1000:1000" + cpus: 1 + mem_limit: 512m + memswap_limit: 512m + environment: *server_environment + volumes: + - ${ENTRY_PHASE_ROOT:?}/data:/data + depends_on: + bootstrap: + condition: service_completed_successfully + + viewer: + image: ${ENTRY_WATERLINE_IMAGE:?Set the immutable published Waterline image} + cpus: 0.5 + mem_limit: 256m + memswap_limit: 256m + init: true + environment: + WATERLINE_SERVER_ENDPOINT: http://api:8080 + WATERLINE_SERVER_TOKEN: general221-fixture + WATERLINE_NAMESPACE: default + WATERLINE_ACCESS_MODE: operator + WATERLINE_ALLOW_UNAUTHENTICATED: "true" + volumes: + - viewer-data:/data + depends_on: + api: + condition: service_healthy + + client: + image: ${ENTRY_BROWSER_IMAGE:?Set the pinned Python Playwright image} + user: "1000:1000" + init: true + cpus: 2 + mem_limit: 2g + memswap_limit: 2g + shm_size: 256m + environment: + HOME: /tmp + EXPECTED_BINDING_REFUSAL: "1" + WATERLINE_URL: http://viewer:8080/waterline + entrypoint: [sh, -ec] + command: + - | + python -m venv --without-pip /opt/venv + python -m pip --python /opt/venv/bin/python install --index-url https://pypi.org/simple --no-cache-dir --quiet -r /phase/requirements.txt + /opt/venv/bin/python /phase/setup.py + exec sleep infinity + tmpfs: + - /opt/venv:rw,exec,uid=1000,gid=1000,size=536870912 + volumes: + - ${ENTRY_PHASE_ROOT:?}:/phase + - ./requirements.txt:/phase/requirements.txt:ro + - ./setup.py:/phase/setup.py:ro + - ./consumer.py:/phase/consumer.py:ro + - ./worker.py:/phase/worker.py:ro + - ./ui.py:/phase/ui.py:ro + - ./cold.py:/phase/cold.py:ro + - ./audit.py:/phase/audit.py:ro + - ./verify_install.py:/phase/verify_install.py:ro + depends_on: + api: + condition: service_healthy + viewer: + condition: service_healthy + +volumes: + viewer-data: diff --git a/tests/fixtures/entry_recovery_http/consumer.py b/tests/fixtures/entry_recovery_http/consumer.py new file mode 100644 index 0000000..0f306ac --- /dev/null +++ b/tests/fixtures/entry_recovery_http/consumer.py @@ -0,0 +1,202 @@ +import asyncio +import hashlib +import importlib.metadata +import json +import os +import signal +import sys +import time +from pathlib import Path + +import httpx +from ui import repair_in_waterline + +from durable_workflow import Client + +ROOT = Path('/phase/consumer') +BASE = '''import os +from pathlib import Path +from durable_workflow import workflow +@workflow.defn(name="general221.entry") +class WaitingWorkflow: + def __init__(self): + self.ready = False + @workflow.signal("ready") + def receive(self): + self.ready = True + def run(self, ctx, amount): +{body} +''' +WAIT = ' yield ctx.wait_condition(lambda: self.ready, key="ready")\n' +RETURN = ' return {"amount": amount, "typed": [9223372036854775807, None, "Київ"]}\n' +BODY_MARKER = ' Path(os.environ["BODY_TRACE"]).write_text("entered")\n' + + +def records(path): + return [json.loads(line) for line in path.read_text().splitlines()] if path.exists() else [] + + +async def until(check, label, timeout=12): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + result = await check() + if result: + return result + await asyncio.sleep(0.05) + raise AssertionError(label) + + +async def main(): + assert importlib.metadata.version('durable-workflow') == '2.8.0' + case, cohort = sys.argv[1:] + identity = cohort + '-' + case + directory = ROOT / identity + directory.mkdir(parents=True) + module = directory / 'definition.py' + after_wait = case in ('required-after', 'body-after') + old_source = BASE.format(body=BODY_MARKER + (WAIT if after_wait else '') + RETURN) + new_source = old_source + if case.startswith('required-'): + new_source = new_source.replace('ctx, amount)', 'ctx, amount, required)') + elif case.startswith('body-'): + new_source = new_source.replace(RETURN, ' raise TypeError("application-body-control")\n') + elif case == 'optional': + new_source = new_source.replace('ctx, amount)', 'ctx, amount, extra=None)') + else: + raise AssertionError(case) + module.write_text(old_source) + processes = [] + fixed = os.environ.get('EXPECTED_BINDING_REFUSAL') == '1' + + async def worker(which, enabled): + gate, trace = directory / (which + '.gate'), directory / (which + '.jsonl') + if enabled: + gate.touch() + env = {**os.environ, 'TASK_QUEUE': 'q-' + identity, 'WORKER_ID': identity + '-' + which, + 'TRACE': str(trace), 'GATE': str(gate), + 'BODY_TRACE': str(directory / (which + '.entered')), + 'PYTHONPATH': str(directory)} + log = (directory / (which + '.log')).open('w') + process = await asyncio.create_subprocess_exec(sys.executable, '/phase/worker.py', env=env, + stdout=log, stderr=asyncio.subprocess.STDOUT) + processes.append((process, log)) + + async def registered(): + assert process.returncode is None, (case, which, process.returncode) + return any(row['kind'] == 'registered' for row in records(trace)) + + await until(registered, 'registration ' + which) + return process, trace + + async def stop(process): + if process.returncode is None: + process.send_signal(signal.SIGTERM) + code = await asyncio.wait_for(process.wait(), 8) + assert code == 0, (case, 'worker exit', code) + + headers = {'Authorization': 'Bearer general221-fixture', 'X-Namespace': 'default', + 'X-Durable-Workflow-Control-Plane-Version': '2'} + try: + async with ( + httpx.AsyncClient(base_url='http://api:8080', headers=headers, timeout=5) as http, + Client('http://api:8080', token='general221-fixture') as client, + ): + async def get(path): + response = await http.get(path) + response.raise_for_status() + return response.json() + + old, old_trace = await worker('old', after_wait) + handle = await client.start_workflow(workflow_type='general221.entry', workflow_id='original-' + identity, + task_queue='q-' + identity, input=[7], run_timeout_seconds=30, execution_timeout_seconds=60) + route = '/api/workflows/' + handle.workflow_id + history_route = route + '/runs/' + handle.run_id + '/history' + if after_wait: + async def waiting(): + return any(row['kind'] == 'completed_task' for row in records(old_trace)) + await until(waiting, 'original committed wait') + before = await get(route) + prefix = await get(history_route) + (directory / 'original.json').write_text(json.dumps({'case': case, 'before': before, 'prefix': prefix})) + await stop(old) + module.write_text(new_source) + new, new_trace = await worker('new', True) + if after_wait: + await handle.signal('ready', []) + + async def terminal_or_refused(): + state = await get(route) + return state if state['status'] in ('completed', 'failed') or any( + row['kind'] == 'failed_task' for row in records(new_trace) + ) else None + + after = await until(terminal_or_refused, 'terminal or refused observation') + await stop(new) + history = await get(history_route) + assert after['run_id'] == before['run_id'] == handle.run_id + for key in ('run_deadline_at', 'execution_deadline_at'): + assert before[key] == after[key] and after[key], (key, before, after) + assert prefix['events'] == history['events'][:len(prefix['events'])] + refusal, debug, repair, waterline = None, None, None, None + if fixed and case.startswith('required-'): + refusal = after + assert not after['is_terminal'], after + failed = [row for row in records(new_trace) if row['kind'] == 'failed_task'] + assert len(failed) == 1, failed + assert failed[0]['request']['failure_type'] == 'WorkflowEntryArgumentsMismatchError', failed + assert failed[0]['request']['failure_reason'] == 'workflow_entry_arguments_mismatch', failed + assert not (directory / 'new.entered').exists(), 'incompatible entry executed application code' + assert 'workflow_entry_arguments_mismatch' in failed[0]['request']['message'] + assert not any(row['kind'] == 'completed_task' for row in records(new_trace)) + assert not any( + row['event_type'] in ('WorkflowFailed', 'WorkflowCompleted') for row in history['events'] + ) + debug = await get(route + '/runs/' + handle.run_id + '/debug') + assert 'WorkflowEntryArgumentsMismatchError' in json.dumps(debug), debug + assert '"replay_blocked": true' in json.dumps(debug), debug + module.write_text(old_source) + recovery, recovery_trace = await worker('recovery', True) + waterline = await repair_in_waterline(handle.workflow_id, handle.run_id, case, directory) + repair = waterline['repair_response'] + result = await handle.result(timeout=12) + assert result == {'amount': 7, 'typed': [9223372036854775807, None, 'Київ']}, result + await stop(recovery) + after = await get(route) + history = await get(history_route) + assert after['run_id'] == handle.run_id and after['status'] == 'completed' + assert sum(row['event_type'] == 'WorkflowCompleted' for row in history['events']) == 1 + for key in ('run_deadline_at', 'execution_deadline_at'): + assert before[key] == after[key] + assert prefix['events'] == history['events'][:len(prefix['events'])] + elif case == 'optional': + assert (directory / 'new.entered').exists() + assert after['status'] == 'completed', after + assert await handle.result(timeout=2) == {'amount': 7, 'typed': [9223372036854775807, None, 'Київ']} + else: + assert (directory / 'new.entered').exists(), 'business failure must execute the body' + assert after['status'] == 'failed', after + assert any(row['event_type'] == 'WorkflowFailed' for row in history['events']), history + evidence = {'case': case, 'cohort': cohort, 'sdk': importlib.metadata.version('durable-workflow'), + 'before': before, 'after': after, 'prefix': prefix, 'history': history, + 'old_source_sha256': hashlib.sha256(old_source.encode()).hexdigest(), + 'new_source_sha256': hashlib.sha256(new_source.encode()).hexdigest(), + 'new_trace': records(new_trace), 'refusal': refusal, 'debug': debug, 'repair': repair, + 'new_body_entered': (directory / 'new.entered').exists(), + 'waterline': waterline, + 'worker_exits': [p.returncode for p, _ in processes]} + (directory / 'evidence.json').write_text(json.dumps(evidence, ensure_ascii=False, indent=2)) + print(json.dumps({'case': case, 'cohort': cohort, 'run_id': handle.run_id, 'status': after['status'], + 'events': len(history['events']), 'worker_exits': evidence['worker_exits']}), flush=True) + finally: + for process, log in processes: + if process.returncode is None: + process.send_signal(signal.SIGTERM) + try: + await asyncio.wait_for(process.wait(), 8) + except TimeoutError: + process.kill() + await process.wait() + log.close() + + +asyncio.run(asyncio.wait_for(main(), timeout=30)) diff --git a/tests/fixtures/entry_recovery_http/requirements.txt b/tests/fixtures/entry_recovery_http/requirements.txt new file mode 100644 index 0000000..677b01e --- /dev/null +++ b/tests/fixtures/entry_recovery_http/requirements.txt @@ -0,0 +1,2 @@ +durable-workflow==2.8.0 +playwright==1.55.0 diff --git a/tests/fixtures/entry_recovery_http/setup.py b/tests/fixtures/entry_recovery_http/setup.py new file mode 100644 index 0000000..9469cc9 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/setup.py @@ -0,0 +1,37 @@ +"""Install verification and browser readiness before business admission.""" + +import asyncio +import hashlib +import json +import urllib.request +from pathlib import Path + +import httpx +from playwright.async_api import async_playwright + +with urllib.request.urlopen('https://pypi.org/pypi/durable-workflow/2.8.0/json', timeout=20) as response: + release = json.load(response) +wheel = next(file for file in release['urls'] if file['filename'].endswith('.whl')) +with urllib.request.urlopen(wheel['url'], timeout=20) as response: + content = response.read() +assert hashlib.sha256(content).hexdigest() == '908267666c35890fce13ab9c8e62a5c93abac78bdffbb0e56423eb3b3e57a77b' +Path('/phase/sdk.whl').write_bytes(content) + + +async def ready(): + async with httpx.AsyncClient(timeout=5) as client: + response = await client.get('http://api:8080/api/ready') + assert response.status_code == 200, response.text + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(args=['--no-sandbox']) + try: + page = await browser.new_page() + response = await page.goto('http://viewer:8080/waterline', wait_until='domcontentloaded', timeout=10000) + assert response.status == 200 + finally: + await browser.close() + Path('/phase/setup.ready').write_text('ready') + print('API, viewer and actual browser ready; no workflow admitted.', flush=True) + + +asyncio.run(ready()) diff --git a/tests/fixtures/entry_recovery_http/ui.py b/tests/fixtures/entry_recovery_http/ui.py new file mode 100644 index 0000000..5abbbb2 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/ui.py @@ -0,0 +1,49 @@ +"""Real published Waterline interaction. No routes or backend replies are mocked.""" + +import os +from pathlib import Path + +from playwright.async_api import async_playwright + + +async def repair_in_waterline(workflow_id, run_id, case, directory): + mobile = case == 'required-after' + theme = 'light' if mobile else 'dark' + viewport = {'width': 390, 'height': 844} if mobile else {'width': 1440, 'height': 900} + root = os.environ.get('WATERLINE_URL', 'http://viewer:8080/waterline').rstrip('/') + selected_path = f'/api/instances/{workflow_id}/runs/{run_id}/repair' + async with async_playwright() as playwright: + browser = await playwright.chromium.launch(args=['--no-sandbox']) + try: + page = await browser.new_page(viewport=viewport) + page.set_default_timeout(5000) + await page.add_init_script(f"localStorage.setItem('waterline-theme', '{theme}')") + await page.goto( + f'{root}/flows/instances/{workflow_id}/runs/{run_id}', wait_until='domcontentloaded', timeout=8000, + ) + await page.get_by_role('button', name='Inspect full details', exact=True).click() + reason = page.get_by_text('workflow_entry_arguments_mismatch', exact=False).first + await reason.wait_for(state='visible') + await reason.scroll_into_view_if_needed() + assert await page.locator('body').evaluate('(body) => body.scrollWidth <= innerWidth + 1') + await page.screenshot(path=str(Path(directory) / 'waterline-refusal.png'), full_page=True) + await page.get_by_role('button', name='Repair', exact=True).click() + await page.get_by_text('Repair run?', exact=True).wait_for(state='visible') + async with page.expect_response( + lambda response: response.request.method == 'POST' and response.url.endswith(selected_path), + timeout=8000, + ) as captured: + await page.get_by_role('button', name='Repair run', exact=True).click() + response = await captured.value + assert response.status == 200, await response.text() + result = await response.json() + return {'url': page.url, 'viewport': viewport, 'theme': theme, + 'selected_run_repair_url': response.url, 'repair_response': result} + except Exception as error: + if 'page' in locals(): + await page.screenshot(path=str(Path(directory) / 'waterline-failure.png'), full_page=True) + text = await page.locator('body').inner_text() + (Path(directory) / 'waterline-failure.txt').write_text(str(error)[:1000] + '\n' + text[:6000]) + raise + finally: + await browser.close() diff --git a/tests/fixtures/entry_recovery_http/verify_install.py b/tests/fixtures/entry_recovery_http/verify_install.py new file mode 100644 index 0000000..444de16 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/verify_install.py @@ -0,0 +1,43 @@ +import ast +import base64 +import hashlib +import importlib.metadata +import json +from pathlib import Path +from zipfile import ZipFile + +dist = importlib.metadata.distribution('durable-workflow') +assert dist.version == '2.8.0' +wheel = Path('/phase/sdk.whl') +assert hashlib.sha256(wheel.read_bytes()).hexdigest() == ( + '908267666c35890fce13ab9c8e62a5c93abac78bdffbb0e56423eb3b3e57a77b' +) +archive = ZipFile(wheel) +checked = 0 +for entry in dist.files or []: + if not str(entry).startswith('durable_workflow/') or entry.hash is None: + continue + path = dist.locate_file(entry) + assert str(path).startswith('/opt/venv/'), path + expected = entry.hash.value.rstrip('=') + actual = base64.urlsafe_b64encode(hashlib.sha256(path.read_bytes()).digest()).decode().rstrip('=') + assert expected == actual, str(entry) + assert path.read_bytes() == archive.read(str(entry)), str(entry) + checked += 1 +assert checked > 25 +source = Path('/phase/consumer.py').read_text() +tree = ast.parse(source) +constants = {} +for node in tree.body: + if isinstance(node, ast.Assign) and isinstance(node.value, ast.Constant): + for target in node.targets: + if isinstance(target, ast.Name): + constants[target.id] = node.value.value +for wait in ('', constants['WAIT']): + definition = constants['BASE'].format(body=constants['BODY_MARKER'] + wait + constants['RETURN']) + compile(definition, '', 'exec') +archive.close() +print(json.dumps({ + 'installed_version': dist.version, 'production_record_and_wheel_hashes_checked': checked, + 'source_override': False, 'generated_workflows_compile': True, +})) diff --git a/tests/fixtures/entry_recovery_http/worker.py b/tests/fixtures/entry_recovery_http/worker.py new file mode 100644 index 0000000..6f584e3 --- /dev/null +++ b/tests/fixtures/entry_recovery_http/worker.py @@ -0,0 +1,63 @@ +import asyncio +import json +import os +import signal +import time +from pathlib import Path + +from definition import WaitingWorkflow + +from durable_workflow import Client, Worker + +TRACE = Path(os.environ['TRACE']) +STOP = asyncio.Event() + + +def trace(record): + with TRACE.open('a') as stream: + stream.write(json.dumps({'time': time.monotonic(), **record}, ensure_ascii=False) + '\n') + + +class ObservedClient(Client): + async def register_worker(self, **kwargs): + result = await super().register_worker(**kwargs) + trace({'kind': 'registered', 'worker_id': kwargs.get('worker_id')}) + return result + + async def poll_workflow_task(self, **kwargs): + gate = Path(os.environ['GATE']) + while not gate.exists(): + if STOP.is_set(): + return None + await asyncio.sleep(0.02) + task = await super().poll_workflow_task(**kwargs) + trace({'kind': 'poll', 'task': task}) + return task + + async def complete_workflow_task(self, **kwargs): + response = await super().complete_workflow_task(**kwargs) + trace({'kind': 'completed_task', 'request': kwargs, 'response': response}) + return response + + async def fail_workflow_task(self, **kwargs): + response = await super().fail_workflow_task(**kwargs) + trace({'kind': 'failed_task', 'request': kwargs, 'response': response}) + return response + + +async def main(): + async with ObservedClient('http://api:8080', token='general221-worker') as client: + worker = Worker(client, task_queue=os.environ['TASK_QUEUE'], workflows=[WaitingWorkflow], + worker_id=os.environ['WORKER_ID'], poll_timeout=1, heartbeat_interval=1) + loop = asyncio.get_running_loop() + + def shutdown(): + STOP.set() + asyncio.create_task(worker.stop()) + + for signum in (signal.SIGTERM, signal.SIGINT): + loop.add_signal_handler(signum, shutdown) + await worker.run() + + +asyncio.run(main())