From 62e98efe446cad81ab72b0579f1a39392aa7d9c9 Mon Sep 17 00:00:00 2001 From: Baran Sekin Date: Wed, 16 Sep 2026 11:37:49 +0300 Subject: [PATCH] test: add instance query and manual index regression scenarios --- TEST-SCENARIOS.md | 3 + .../attribute-index-preparation/README.md | 69 ++++++ .../test_auto_indexes.py | 142 +++++++++++++ .../test_manual_indexes.py | 97 +++++++++ api-tests/instance-queries/README.md | 159 ++++++++++++++ .../instance-queries/pagination_checks.py | 115 ++++++++++ .../instance-queries/test_instance_queries.py | 201 ++++++++++++++++++ 7 files changed, 786 insertions(+) create mode 100644 api-tests/attribute-index-preparation/README.md create mode 100644 api-tests/attribute-index-preparation/test_auto_indexes.py create mode 100644 api-tests/attribute-index-preparation/test_manual_indexes.py create mode 100644 api-tests/instance-queries/README.md create mode 100644 api-tests/instance-queries/pagination_checks.py create mode 100644 api-tests/instance-queries/test_instance_queries.py diff --git a/TEST-SCENARIOS.md b/TEST-SCENARIOS.md index 17ec586..7b64a44 100644 --- a/TEST-SCENARIOS.md +++ b/TEST-SCENARIOS.md @@ -15,6 +15,9 @@ geçerliliğini yitiren senaryo **silinmez**, `deprecated` işaretlenip sebebi y | Senaryo | Test Edilen vNext Feature Seti | Neden Eklendi | Integration Test | Python Test | Durum | |---|---|---|---|---|---| +| **instance-queries** | Instance data persistence → parameterized JSONB filtering; GraphQL/legacy value limits; HTTP SQL pagination, requested ordering, ID tie-breaks, hydration and HATEOAS next links | vnext #934 / #933, PR #987 (2026-09-15/16) | Locally built runtime + isolated PostgreSQL/OpenObserve | [test_instance_queries.py](api-tests/instance-queries/test_instance_queries.py), [pagination_checks.py](api-tests/instance-queries/pagination_checks.py), [README](api-tests/instance-queries/README.md) | Value-limit **25/25**; pagination **611/611**, 192/192 traces on runtime `549722fe`. Baseline four legacy-newline failures and envelope-format limits remain documented; no index benchmark claimed | +| **attribute-index-preparation** | Publish → no automatic jobs/DDL → offline CLI SQL → explicit DBA execution; replay; Master update | Manual DBA-owned index maintenance | Real locally built Orchestration + isolated PostgreSQL | [test_manual_indexes.py](api-tests/attribute-index-preparation/test_manual_indexes.py) | PASS: 4 checks (2026-09-14) | +| **attribute-index-preparation (historical)** | Former directly:true automatic job experiment | Superseded by manual DBA decision | Historical evidence only | [README](api-tests/attribute-index-preparation/README.md) | DEPRECATED: previous 6 checks / 13 jobs remain recorded; automatic behavior removed 2026-09-14 | | **error-boundary-lab** | Error boundary çözümlemesi: `CompiledBoundaryChain` Task→State→Global seviye baskınlığı · seviye içi sıra (`EffectivePriority` ASC → specificity DESC; default wildcard'ın 999'a düşmesi) · aksiyonlar abort/retry/rollback/notify/ignore/log · `TaskExecutionEngine` retry döngüsü (`1 + maxRetries`) ve tükenince `ResolveExcluding(Retry)` fallback'i · `BoundaryOutcomeHandler` → fault vs `RequestNextTransition` · `FinalizeTransitionStep`'in boundary transition'ı bitince incident'ı resolve etmesi · `InstanceIncidents` tablosu + denormalize `HasActiveIncident` · state function `incident` bloğu (link tabanlı, ResponseShapeVersion v9, ETag materyali) · `GET .../instances/{id}/incidents/active` (404 = açık arıza yok) · `GET .../instances/{id}/incidents` sayfalama · `metadata.incident` (state bloğuyla aynı şekil) · `POST .../retry` (400 `Instance:100027`, yeniden fault, veriyle kurtarma) · `queryRoles` kapısının incident geçmişine de uygulanması | vnext `feature/incident-table` (issue #865) incident'ları jsonb'den kendi tablosuna taşıdı ve iki yeni client yüzeyi ekledi; ayrıca error boundary bu repoda hiç senaryo olarak yoktu ve `POST .../retry` hiçbir testte çağrılmıyordu (2026-09-06) | `Tests/ErrorBoundaryLab` (6 sınıf, 31 test) | — (bilinçli: davranış/çözümleme senaryosu, eşzamanlılık iddiası yok) | ✅ **Aktif — 31/31 yeşil** (lokal runtime `feature/incident-table`, art arda iki koşu, ~1 dk 46 sn; ilk koşu 2026-09-06, üç davranışsal kusurun düzeltilmesinden sonra 2026-09-07'de yeniden) | | **event-driven-lab** | Olay tetiklemeli start + `triggerType: 3` transition · Dapr Subscription route'u · CloudEvent açma · `IEventMapping` ile iş anahtarı korelasyonu · eşleşmeyen olayın ack'lenmesi | `POST .../instances/events` ucunun **hiç testi yoktu** (council `2026-09-11-route-trace-coverage`, adı konmuş boşluk); ayrıca `Event.Intake` span'i bu ucu ölçüyor ve testsiz route'a span gönderilmez (2026-09-13) | `Tests/EventDrivenLab` (1 sınıf) | — | ✅ Aktif | | **chain-busy** | Accept-time subflow chain reserve **ve başarısız forward'da geri alınması (E31)** · Busy-as-mutex · `$self` shared transition vs `updateData` lifecycle sınırı · start `initial → initial` semantiği · cancel propagasyonu (in-process ↕ distributed) · scheduled transition re-arm | `updateData`-only self-target profil sınırını pinlemek — `target: $self` "hook'ları atla" demek değil (2026-08-17). **E31 (2026-09-12):** post-commit forward istemci hatasıyla düşünce rezervasyon salınmıyordu; üretimde 30 günde 51 mahsur instance (council `2026-09-08-parent-notification-mechanism`, P0) | `Tests/ChainBusy` (5 sınıf) | `api-tests/chain-busy/chain-busy-behaviour-test.py`, `chain-busy-accept-test.py` | ✅ Aktif | diff --git a/api-tests/attribute-index-preparation/README.md b/api-tests/attribute-index-preparation/README.md new file mode 100644 index 0000000..ebc15c2 --- /dev/null +++ b/api-tests/attribute-index-preparation/README.md @@ -0,0 +1,69 @@ +# Manual attribute index preparation (current contract) + +Publication creates no index jobs or generated columns. `wf indexes generate` writes an offline SQL batch; +the DBA executes it in a maintenance window. The former automatic scenario below is historical. + +Run `test_manual_indexes.py` against the locally built worktree runtime and an isolated test domain: + +```sh +VNEXT_BASE_URL=http://localhost: \ +VNEXT_TEST_DOMAIN= \ +VNEXT_TEST_PG_CONTAINER= \ +VNEXT_TEST_DATABASE= \ +VNEXT_ALLOW_ISOLATED_TEST_DDL=true \ +python3 api-tests/attribute-index-preparation/test_manual_indexes.py +``` + +`NODE_BINARY` and `VNEXT_CLI_ROOT` may override the local Node executable and sibling CLI checkout. +The harness uses a unique flow, executes generated SQL only on that fixture, and leaves it for inspection. +It never modifies pre-existing workflow schemas. The CLI itself never contacts the database/API. + +Verified 2026-09-14: **4 checks passed**, local Orchestration built from the dirty worktree, PostgreSQL 18.6, +`autoindexdirect` isolated domain on localhost:4301. Covered: no automatic work on first publication; +CLI generation leaves DB unchanged; explicit SQL creates ready catalog and replay preserves OIDs; +subsequent indexed Master update creates no jobs and leaves catalog unchanged. + +## Historical automatic experiment (superseded) + +# Automatic attribute index preparation + +Run against an isolated **locally built** runtime with automatic preparation enabled. The selected `directly: true` path works +with the existing background-job poller disabled. The script publishes uniquely named components through HTTP. +PostgreSQL is inspected with read-only SQL; one read lock is held to prove the no-op path does +not request ACCESS EXCLUSIVE. No direct optimization DDL is performed by the test. + +```sh +VNEXT_BASE_URL=http://localhost:4301 \ +VNEXT_TEST_DOMAIN=autoindexdirect \ +VNEXT_TEST_PG_CONTAINER=auto-index-pg \ +VNEXT_TEST_DATABASE=vNext_autoindex_direct \ +python3 api-tests/attribute-index-preparation/test_auto_indexes.py +``` + +Expected checks: unindexed schema; Master package update affects latest but not pinned flow; +trigram/B-tree generation; no-op while a conflicting read lock is held; conflict/exception rollback; +workflow reference update. Job status and retry assertions require a dedicated clean job database. + +## 2026-09-11 result: blocked, not passed + +Worktree runtime on `localhost:4301`, PostgreSQL 18.6, Dapr 1.16.19, isolated Redis/Scheduler. +Some initial jobs completed and created projections. Immediate Dapr callbacks also arrived +before Aether's arming confirmation committed; dispatcher acknowledged them as `not found in +an active state`, then their rows became Scheduled and stayed there. The scenario timed out +at its first job-completion gate. Later checks are **not verified**. + +Aether 1.0.39 SDK changes require separate authorization under the runtime repository policy. +No sleep-based timing workaround or direct DB status repair is included in this test. Re-run +all checks after the SDK race is fixed. Aether source was not modified during this investigation. + +## 2026-09-11 direct-path follow-up: passed + +User selected `directly: true` without modifying Aether. Automatic preparation was enabled; +`BackgroundJob:WithHostedService=false` was deliberately retained during verification. A fresh +isolated database (`vNext_autoindex_direct`, domain `autoindexdirect`) preserved the earlier +failed-run evidence. All **6 checks passed**, and **13 jobs completed with zero retries**. +The 409 conflict and 500 extra-data cast failure are intentional rollback probes, not unexpected +failures. Tables are small/empty fixtures; this run is not a populated 10M-row migration benchmark. + +This closes the direct-path scenario. Commit-before-schedule crashes and Pending/Retrying +recovery remain SDK limitations; the earlier failure above has not been erased or fixed. diff --git a/api-tests/attribute-index-preparation/test_auto_indexes.py b/api-tests/attribute-index-preparation/test_auto_indexes.py new file mode 100644 index 0000000..c7a9eb7 --- /dev/null +++ b/api-tests/attribute-index-preparation/test_auto_indexes.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Real-host regression: publication -> durable Aether job -> Dapr -> generated indexes. + +Requires an isolated local worktree runtime. SQL is used for assertions and a held +read lock only; all definitions/data are published through the HTTP API. +""" +import json +import os +import subprocess +import time +import urllib.error +import urllib.request +import uuid + +BASE = os.environ['VNEXT_BASE_URL'].rstrip('/') +DOMAIN = os.environ['VNEXT_TEST_DOMAIN'] +PG = os.environ['VNEXT_TEST_PG_CONTAINER'] +DATABASE = os.environ['VNEXT_TEST_DATABASE'] +PREFIX = 'auto-' + uuid.uuid4().hex[:10] +MASTER = PREFIX + '-master' +LATEST = PREFIX + '-latest' +PINNED = PREFIX + '-pinned' + + +def sql(query): + return subprocess.check_output(['docker', 'exec', PG, 'psql', '-U', 'postgres', '-d', DATABASE, + '-At', '-v', 'ON_ERROR_STOP=1', '-c', query], text=True).strip() + + +def publish(kind, key, version, attributes, data=None, allow_failure=False): + body = dict(domain=DOMAIN, flow=kind, key=key, version=version, flowVersion='1.0.0', tags=[], attributes=attributes) + if data is not None: + body['data'] = data + req = urllib.request.Request(BASE + '/api/v1/definitions/publish', data=json.dumps(body).encode(), + headers={'Content-Type': 'application/json'}) + try: + with urllib.request.urlopen(req, timeout=60) as response: + return response.status + except urllib.error.HTTPError as error: + if error.code not in (400, 409) and not allow_failure: + raise AssertionError(error.read().decode()) from error + print('HTTP', error.code, error.read().decode(), flush=True) + return error.code + + +def master(indexed, description=''): + return {'type': 'workflow', 'schema': {'type': 'object', 'description': description, 'properties': { + 'amount': {'type': 'number', 'x-indexed': indexed, 'x-sortable': True, 'x-filterOperators': ['gte']}, + 'name': {'type': 'string', 'x-indexed': indexed, 'x-filterOperators': ['contains']}}}} + + +def workflow(version): + return {'type': 'F', 'labels': [{'language': 'en-US', 'label': 'Auto index test'}], + 'schema': {'domain': DOMAIN, 'flow': 'sys-schemas', 'key': MASTER, 'version': version}, + 'startTransition': {'key': 'start', 'target': 'ready', 'versionStrategy': 'Minor', + 'labels': [{'language': 'en-US', 'label': 'Start'}]}, + 'states': [{'labels': [{'language': 'en-US', 'label': 'Ready'}], 'key': 'ready', 'stateType': 1, 'subType': 0, 'versionStrategy': 'Minor', + 'onEntries': [], 'onExits': [], 'transitions': []}]} + + +def jobs_finished(): + rows = json.loads(sql('SELECT coalesce(json_agg(t), \'[]\'::json) FROM ' + '(SELECT "Status", "RetryCount", "LastError" FROM sys_queues."BackgroundJobs" ' + 'WHERE "HandlerName" = \'attribute-index.prepare\') t')) + assert not any(row['Status'] == 3 for row in rows), rows + return rows and all(row['Status'] == 2 for row in rows) + + +def wait(predicate, timeout=45): + end = time.monotonic() + timeout + while time.monotonic() < end: + if predicate(): + return + time.sleep(0.25) + raise AssertionError('Timed out waiting for automatic preparation') + + +def projections(flow): + return int(sql("SELECT count(*) FROM information_schema.columns WHERE table_schema='" + + flow.replace('-', '_') + "' AND table_name='InstancesData' AND is_generated='ALWAYS'")) + + +def main(): + v1 = '1.0.0-pkg.1.0.0' + v2 = '1.0.0-pkg.1.0.1' + assert publish('sys-schemas', MASTER, v1, master(False)) == 200 + seed = [{'key': 'seed', 'version': '1.0.0', 'attributes': {'amount': 12, 'name': 'İstanbul'}}] + assert publish('sys-flows', LATEST, v1, workflow('latest')) == 200 + assert publish('sys-flows', PINNED, v1, workflow(v1)) == 200 + wait(jobs_finished) + assert projections(LATEST) == projections(PINNED) == 0 + print('PASS unindexed publication produces no generated columns') + + assert publish('sys-schemas', MASTER, v2, master(True)) == 200 + wait(jobs_finished) + assert projections(LATEST) == 3 + assert projections(PINNED) == 0 + physical = LATEST.replace('-', '_') + assert int(sql(f'SELECT count(*) FROM "{physical}"."AttributeIndexCatalog" WHERE "Ready"')) == 3 + assert int(sql(f"SELECT count(*) FROM pg_indexes WHERE schemaname='{physical}' AND indexname LIKE '%_trgm'")) == 1 + print('PASS Master package update prepares latest reference; full revision stays pinned') + + # Keep a read lock while publishing an equivalent physical definition. Taking ACCESS EXCLUSIVE + # on this path would time out; a real no-op finishes without retries while the lock remains held. + lock = subprocess.Popen(['docker', 'exec', '-i', PG, 'psql', '-U', 'postgres', '-d', DATABASE, + '-At', '-v', 'ON_ERROR_STOP=1'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, text=True) + try: + lock.stdin.write(f'BEGIN; LOCK TABLE "{physical}"."InstancesData" IN ACCESS SHARE MODE; SELECT \'locked\';\n') + lock.stdin.flush() + while lock.stdout.readline().strip() != 'locked': + assert lock.poll() is None + assert publish('sys-schemas', MASTER, '1.0.0-pkg.1.0.2', master(True, 'description only')) == 200 + wait(jobs_finished, timeout=15) + assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs" WHERE "HandlerName" = \'attribute-index.prepare\' AND "RetryCount" > 0') == '0' + print('PASS equivalent publication completes while a conflicting read lock is held (no DDL/retry)') + finally: + lock.stdin.write('ROLLBACK;\n'); lock.stdin.close(); lock.wait(timeout=10) + + before = sql('SELECT count(*) FROM sys_queues."BackgroundJobs"') + assert publish('sys-flows', LATEST, v1, workflow('latest')) == 409 + assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs"') == before + print('PASS conflicting publication creates no job') + + # The extra-data cast rejects a non-component flow AFTER the definition and seed have been + # written. Exercise transaction rollback through the real HTTP middleware and two DbContexts. + failed_flow = PREFIX + '-rollback' + assert publish('sys-flows', failed_flow, v1, workflow('latest'), seed, allow_failure=True) == 500 + assert sql(f'SELECT count(*) FROM sys_flows."Instances" WHERE "Key" = \'{failed_flow}\'') == '0' + assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs"') == before + print('PASS exception after definition writes rolls back publication and leaves no job') + + # New workflow revision moving a pinned reference must also prepare without another Master publish. + assert publish('sys-flows', PINNED, v2, workflow('latest')) == 200 + wait(jobs_finished) + assert projections(PINNED) == 3 + print('PASS workflow reference update triggers automatic preparation') + print(json.dumps({'result': 'passed', 'checks': 6, 'base_url': BASE, 'domain': DOMAIN, 'prefix': PREFIX})) + + +if __name__ == '__main__': + raise SystemExit('Historical automatic experiment was removed. Run test_manual_indexes.py for the current contract.') diff --git a/api-tests/attribute-index-preparation/test_manual_indexes.py b/api-tests/attribute-index-preparation/test_manual_indexes.py new file mode 100644 index 0000000..d1f31df --- /dev/null +++ b/api-tests/attribute-index-preparation/test_manual_indexes.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +"""Publish -> no automatic index work -> offline CLI SQL -> DBA-style execution in an isolated fixture. + +Only run against an explicitly authorized isolated local test domain. Generated SQL +is executed by this test harness, never by the CLI/runtime. Existing flows are untouched. +""" +import json +import os +from pathlib import Path +import subprocess +import tempfile +import urllib.error +import urllib.request +import uuid + +BASE = os.environ['VNEXT_BASE_URL'].rstrip('/') +DOMAIN = os.environ['VNEXT_TEST_DOMAIN'] +PG = os.environ['VNEXT_TEST_PG_CONTAINER'] +DATABASE = os.environ['VNEXT_TEST_DATABASE'] +CLI = Path(os.environ.get('VNEXT_CLI_ROOT', Path(__file__).resolve().parents[3] / 'vnext-workflow-cli')) +NODE = os.environ.get('NODE_BINARY', 'node') +PREFIX = 'manual-' + uuid.uuid4().hex[:10] +MASTER, FLOW = PREFIX + '-master', PREFIX + '-flow' + + +def sql(query): + return subprocess.check_output(['docker', 'exec', '-i', PG, 'psql', '-U', 'postgres', '-d', DATABASE, + '-At', '-v', 'ON_ERROR_STOP=1'], input=query, text=True).strip() + + +def publish(body): + request = urllib.request.Request(BASE + '/api/v1/definitions/publish', data=json.dumps(body).encode(), + headers={'Content-Type': 'application/json'}) + try: + with urllib.request.urlopen(request, timeout=60) as response: + assert response.status == 200 + except urllib.error.HTTPError as error: + raise AssertionError(error.read().decode()) from error + + +def component(kind, key, attributes, version): + return {'domain': DOMAIN, 'flow': kind, 'flowVersion': '1.0.0', 'key': key, 'version': version, + 'tags': ['manual-index-test'], 'attributes': attributes} + + +def main(): + if os.environ.get('VNEXT_ALLOW_ISOLATED_TEST_DDL') != 'true': + raise SystemExit('Requires VNEXT_ALLOW_ISOLATED_TEST_DDL=true and an isolated local test domain.') + before = sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';") + version = '1.0.0-pkg.1.0.0' + master = component('sys-schemas', MASTER, {'type': 'workflow', 'schema': {'type': 'object', 'properties': { + 'amount': {'type': 'number', 'x-indexed': True, 'x-filterOperators': ['gte']}, + 'name': {'type': 'string', 'x-indexed': True, 'x-filterOperators': ['contains']}}}}, version) + flow = component('sys-flows', FLOW, { + 'type': 'F', 'labels': [{'language': 'en-US', 'label': 'Manual index test'}], + 'schema': {'domain': DOMAIN, 'flow': 'sys-schemas', 'key': MASTER, 'version': 'latest'}, + 'startTransition': {'key': 'start', 'target': 'ready', 'versionStrategy': 'Minor', + 'labels': [{'language': 'en-US', 'label': 'Start'}]}, + 'states': [{'labels': [{'language': 'en-US', 'label': 'Ready'}], 'key': 'ready', 'stateType': 1, + 'subType': 0, 'versionStrategy': 'Minor', 'onEntries': [], 'onExits': [], 'transitions': []}] + }, version) + publish(master) + publish(flow) + schema = FLOW.replace('-', '_') + assert sql(f"SELECT count(*) FROM information_schema.columns WHERE table_schema='{schema}' AND table_name='InstancesData' AND is_generated='ALWAYS';") == '0' + assert sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';") == before + print('PASS indexed Master/workflow publication creates no index jobs or projections', flush=True) + with tempfile.TemporaryDirectory(prefix='vnext-manual-indexes-') as folder: + root = Path(folder) + (root/'components'/'Schemas').mkdir(parents=True) + (root/'components'/'Workflows').mkdir(parents=True) + (root/'vnext.config.json').write_text(json.dumps({'domain': DOMAIN, 'paths': { + 'componentsRoot': 'components', 'workflows': 'Workflows', 'schemas': 'Schemas'}})) + (root/'components'/'Schemas'/'master.json').write_text(json.dumps(master)) + (root/'components'/'Workflows'/'flow.json').write_text(json.dumps(flow)) + subprocess.run([NODE, str(CLI/'bin'/'workflow.js'), 'indexes', 'generate', '--flow', FLOW, + '--output', 'sql'], cwd=root, check=True) + generated = next((root/'sql').glob('*/*.sql')).read_text() + assert sql(f"SELECT count(*) FROM information_schema.columns WHERE table_schema='{schema}' AND table_name='InstancesData' AND is_generated='ALWAYS';") == '0' + print('PASS CLI only writes SQL; database is unchanged', flush=True) + sql(generated) # Explicit DBA-style execution by test harness, confined to its unique flow. + assert sql(f'SELECT count(*) FROM "{schema}"."AttributeIndexCatalog" WHERE "Ready";') == '3' + oids = sql(f"SELECT indexrelid FROM pg_index WHERE indrelid='\"{schema}\".\"InstancesData\"'::regclass ORDER BY indexrelid;") + sql(generated) + assert sql(f"SELECT indexrelid FROM pg_index WHERE indrelid='\"{schema}\".\"InstancesData\"'::regclass ORDER BY indexrelid;") == oids + print('PASS DBA execution prepares catalog, replay preserves indexes', flush=True) + master['version'] = '1.0.0-pkg.1.0.1' + master['attributes']['schema']['properties']['another'] = {'type': 'integer', 'x-indexed': True} + publish(master) + assert sql(f'SELECT count(*) FROM "{schema}"."AttributeIndexCatalog" WHERE "Ready";') == '3' + assert sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';") == before + print('PASS subsequent Master publication leaves index maintenance to DBA', flush=True) + print(json.dumps({'result': 'passed', 'checks': 4, 'base_url': BASE, 'domain': DOMAIN, 'prefix': PREFIX})) + + +if __name__ == '__main__': + main() diff --git a/api-tests/instance-queries/README.md b/api-tests/instance-queries/README.md new file mode 100644 index 0000000..37acbe3 --- /dev/null +++ b/api-tests/instance-queries/README.md @@ -0,0 +1,159 @@ +# Instance query regression (#933 / #934) + +## Purpose + +Verify that caller-controlled filter values remain literal JSON string values through +HTTP parsing, `AttributeConditionBuilder`, parameterized PostgreSQL JSONB containment, +and instance listing. Added 2026-09-15 for [vnext issue #934](https://github.com/burgan-tech/vnext/issues/934). + +The script publishes a uniquely named task-free workflow, starts twelve instances through +the API (`start → ready/Active`), queries them, and checks the latest PostgreSQL rows. +All writes use the API. SQL is read-only. The critical path is instance-list filter +compilation and repository execution; no external HTTP task or MockLab is needed. + +## Run + +Use Python 3, Docker CLI, and a dedicated runtime built from the vnext working tree. +Run DbMigrator and publish that domain's system flows first. Do not target a released +runtime image or another team's domain. The script leaves its unique workflow/data for +evidence inspection and requires explicit connection parameters. + +```bash +python3 api-tests/instance-queries/test_instance_queries.py \ + --base-url http://localhost:6201 --domain core \ + --pg-container issue934-postgres --database Issue934 \ + --output /tmp/issue934-results.json +``` + +## Assertions and interpretation + +- Seven payloads × two paths (`name`, `profile.name`) × two operators (`eq`, `ne`) + × two formats (GraphQL JSON, legacy): 56 exact-result-set assertions. +- Payloads cover quotes, backslashes, invalid JSON escapes, newline/tab, extra properties, + duplicate keys, and SQL-looking text. Decoy records detect changed filter semantics. +- Five invalid-input/total-size assertions require HTTP 400-class rejection. +- Two PostgreSQL assertions require exact unchanged values/properties and all twelve + instances Active. +- Eighteen value-limit assertions cover 1000 accepted / 1001 rejected, per-element + membership/range limits, combined list lengths above 1000, logical/nested conditions, + and aggregation envelopes. Total: 81 assertions. Any failure returns exit code 1. +- Every HTTP request records a trace ID; confirm the run in OpenObserve and PostgreSQL. + +For the value-limit change alone, append `--value-limits-only` to run 25 checks. This +retains the malformed-input, total-size, and PostgreSQL assertions while omitting the +pre-existing injection/newline compatibility matrix. + +## Baseline before wiring ValidateValue: 2026-09-15 + +Runtime `3000c6071c869f1b28c745231c04ee527a57fe7d`, branch +`feature/instance-data-performance`, isolated API `http://localhost:6201`: + +- **59/63 assertions passed; four failed.** All four failures are legacy-format queries + containing a newline: HTTP 400 `filter.unrecognizedFormat`. `LegacyFilterPattern` uses + `.` without `RegexOptions.Singleline`. GraphQL JSON supports the same values correctly. + The failures remain visible; the test does not silently accept a rejection as a successful match. +- No JSON injection reproduced. The other 52 payload/operator/path/format combinations + returned exactly the expected IDs; all stored values/properties were unchanged. +- A 1001-character value was accepted and matched in both formats. The independent + 5000-character total-filter limit rejected oversized inputs in both formats. +- OpenObserve: all 75 request trace IDs found, 2687 spans, 108 logs, no filter database + error spans or Error-level logs. Successful GET server spans (54): average 2.509 ms, + p95 4.402 ms, maximum 10.320 ms. These are small-fixture observations, not load benchmarks. +- One setup-only `42P01` span probes a not-yet-created migration-history table during + new workflow publication; publication succeeds and all later filter SQL succeeds. + +This is a targeted review of #934, not a platform-wide security assessment. Aggregation, +authorization bypass, other operators, and load/resource-exhaustion behavior are outside this test. + +## ValidateValue implementation: 2026-09-15 + +On the same branch plus the value-limit working-tree changes, `--value-limits-only` +passed **25/25** checks. Both formats accept 1000-character values and reject 1001 with +HTTP 400, including oversized `in`/`nin`/`between` elements. Lists whose individual +values fit still execute when their combined length exceeds 1000. Nested/logical +conditions and a grouping envelope cannot bypass the limit. All twelve persisted +records remain unchanged and Active; the limit is on filter operands, not stored data. + +OpenObserve confirmed all **36/36 request traces**, 1136 spans and 79 logs. All twelve +value-length failures logged `filter.valueTooLong`. The seventeen rejected requests +(also including existing malformed/total-size checks) generated **zero PostgreSQL +spans**. Successful GET spans (six requests): average 23.661 ms, p95/max 115.939 ms; +this small cold-process sample is not a performance comparison. No filter DB errors +or Error-level logs occurred. The original four legacy-newline compatibility cases +are unchanged and still remain in the default full matrix. + +## HTTP pagination, ordering and tie-breaks (PR #987) + +Added 2026-09-16 for [vnext PR #987](https://github.com/burgan-tech/vnext/pull/987) +and issue #933. `--pagination-only` extends the same task-free HTTP fixture instead of +publishing a second workflow design. The state sketch remains `start → ready/Active`; +no transition, timer, external task or concurrent write runs while pages are read. +The critical path is HTTP query binding → ordered SQL identity page → page hydration +→ HATEOAS links. The companion `pagination_checks.py` uses the authored values and +API-returned IDs as its ordering oracle, not the runtime's first-page order. + +```bash +python3 api-tests/instance-queries/test_instance_queries.py \ + --pagination-only --base-url http://localhost:6201 --domain core \ + --pg-container issue934-postgres --database Issue934 \ + --output /tmp/instance-pagination-results.json +``` + +Use the same locally built runtime/system-flow prerequisites as above. This mode and +`--value-limits-only` are mutually exclusive. It adds a `rank` attribute to the twelve +API-created rows: three groups of four equal values, so three-row pages cut through ties. +All writes still go through HTTP; PostgreSQL access remains read-only. + +### Pass criteria + +- **Eight orderings × three wire forms:** status ASC/DESC (all rows tie), explicit ID + DESC, key ASC/DESC, attribute rank ASC/DESC, and multi-field status/rank ordering. + Wire forms are bare GraphQL, legacy, and the existing request-envelope format. +- Exact ordered IDs on every page, including the exact last page and the following + empty page; no duplicates or missing rows across a traversal. +- Equal primary sort values resolve by **ID ASC**, including descending primary sorts. + The explicit ID DESC case must retain the caller's direction. +- Repeated second-page reads are identical with no intervening writes. +- Page sizes 1, 4, 5, 12 and 20 cover singleton, exact multiples, a partial final page, + an exact fit and a page larger than the result set. A no-match filter yields an empty + first page. +- The public `hasNext` signal is **`links.next`**, a relative URL string or an empty + string, not a `hasNext` response property. Its presence and page/pageSize parameters + must agree with the expected remaining rows. +- Following the returned next URLs verbatim must preserve filter and ordering across + three pages of an eight-row subset, independently for all three wire forms. +- Hydrated attributes equal the original API inputs. All twelve latest PostgreSQL + records retain exact values and Active status. + +The mode currently runs **611 assertions**. Any false assertion or unexpected HTTP +response fails the command. Results include every request trace ID, expected/actual +page IDs, next URLs and persisted rows for independent inspection. + +### Verified result and limits + +On runtime `549722fe8f7fe62c6b1bea0c94a9087fdcc327d1`, branch +`feature/instance-data-performance`, .NET SDK 10.0.200, isolated API `:6201` and +PostgreSQL database `Issue934`: **611/611 passed**. All five host projects were rebuilt; +Orchestration/Execution/Inbox/Outbox ran as local binaries and DbMigrator completed. + +OpenObserve confirmed **192/192 HTTP request traces**, **4709 spans** and **218 logs**. +All 192 requests returned HTTP 200. The 179 list GETs produced 179 identity-selection +queries with SQL `ORDER BY`/`LIMIT`/`OFFSET` and 149 selected-ID hydration queries +(empty pages need no hydration). No list DB error spans or Error-level logs occurred. +A setup-only `42P01` span checked migration history before the unique workflow schema +was created; publication then succeeded. Successful GET server spans: average +**2.555 ms**, p95 **4.159 ms**, maximum **10.334 ms**. These are a small correctness +fixture's timings, not a JSON-versus-index performance comparison. + +The current parser recognizes request envelopes through a `groupBy`/`aggregations` +key. The list-only envelope in this test explicitly uses **`groupBy: null`**. Two +separate exploratory requests established the existing limitations: an envelope with +only `filter` + `orderBy` returns 400 (`filter.unknownOperator`), and `groupBy: {}` +returns 400 (`groupBy` requires at least one field). They were confirmed in runtime +logs/traces and are not presented as supported pagination shapes or fixed by this test. + +Scope: default `IdentityPaging=true` / `LatestJoin=true`, JSON attribute expressions, +no prepared attribute indexes, no concurrent modifications between page reads. This +pins stable ordering for an unchanged data set; offset pagination does not promise a +snapshot across concurrent writes. It does not benchmark indexes or cover authorization, +aggregation-result pagination, or routing kill-switch configurations. diff --git a/api-tests/instance-queries/pagination_checks.py b/api-tests/instance-queries/pagination_checks.py new file mode 100644 index 0000000..c4fcd57 --- /dev/null +++ b/api-tests/instance-queries/pagination_checks.py @@ -0,0 +1,115 @@ +"""HTTP list pagination regression for #933 / PR #987, sharing the #934 API fixture.""" +import json +import urllib.parse + + +def run_pagination_checks(request, check, domain, flow, rows, results): + """Assert ordered IDs and next links on an immutable, API-created 12-row fixture. + + The oracle uses authored values and returned IDs, never the runtime's first-page order. + Every tie is broken by ascending canonical UUID text (PostgreSQL UUID byte order). + """ + route = f'/api/v1/{domain}/workflows/{flow}/instances' + by_id = {row['id']: row for row in rows} + ids = sorted(by_id) + results['pagination'] = [] + + def next_href(body): + # HATEOAS is the public hasNext contract; do not invent a hasNext JSON property. + link = body['links'].get('next') + assert isinstance(link, str), link + return link or None + + def fetch(parameters): + status, body = request('GET', route + '?' + urllib.parse.urlencode(parameters)) + assert status == 200, (status, body) + assert isinstance(body.get('items'), list) and isinstance(body.get('links'), dict), body + return body + + def assert_page(name, body, expected, page, size, total): + actual = [item['id'] for item in body['items']] + expected_next = page * size < total + href = next_href(body) + detail = {'page': page, 'page_size': size, 'expected_ids': expected, 'actual_ids': actual, + 'expected_has_next': expected_next, 'next': href} + check(name + '/ordered-items', actual == expected, detail) + check(name + '/next-link', bool(href) == expected_next, detail) + check(name + '/hydrated-values', all( + item['id'] in by_id and item['attributes'] == {k: v for k, v in by_id[item['id']].items() if k != 'id'} + for item in body['items']), detail) + if href: + query = urllib.parse.parse_qs(urllib.parse.urlsplit(href).query) + check(name + '/next-page-number', query.get('page') == [str(page + 1)] and + query.get('pageSize') == [str(size)], query) + results['pagination'].append({'name': name, **detail}) + return actual, href + + # Status is identical for all 12 records: every page boundary lies inside a tie. + cases = [ + ('status-asc', {'field': 'status', 'direction': 'asc'}, ids), + ('status-desc', {'field': 'status', 'direction': 'desc'}, ids), + ('id-desc', {'field': 'id', 'direction': 'desc'}, list(reversed(ids))), + ('key-asc', {'field': 'key', 'direction': 'asc'}, sorted(ids, key=lambda i: by_id[i]['testCase'])), + ('key-desc', {'field': 'key', 'direction': 'desc'}, sorted(ids, key=lambda i: by_id[i]['testCase'], reverse=True)), + ('attribute-asc', {'field': 'attributes.rank', 'direction': 'asc'}, sorted(ids, key=lambda i: by_id[i]['rank'])), + ('attribute-desc', {'field': 'attributes.rank', 'direction': 'desc'}, sorted(ids, key=lambda i: -by_id[i]['rank'])), + ('multi-field', {'fields': [{'field': 'status', 'direction': 'desc'}, + {'field': 'attributes.rank', 'direction': 'desc'}]}, + sorted(ids, key=lambda i: -by_id[i]['rank'])), + ] + for name, sort, expected_ids in cases: + # The 3-row pages split each 4-row attribute tie and cross into the next group. + for mode in ['graphql', 'legacy', 'envelope']: + # Existing envelope detection requires groupBy/aggregations. Null grouping + # selects the envelope path without changing an instance list into group summaries. + if mode == 'envelope': + parameters = {'filter': json.dumps({'filter': {'status': {'eq': 'A'}}, 'groupBy': None, 'orderBy': sort})} + else: + parameters = {'filter': json.dumps({'status': {'eq': 'A'}}) if mode == 'graphql' else 'status=eq:A', + 'sort': json.dumps(sort)} + seen = [] + for page in range(1, 6): # exact final page plus one empty page + body = fetch({**parameters, 'page': page, 'pageSize': 3}) + actual, _ = assert_page(f'{mode}/{name}/page-{page}', body, + expected_ids[(page - 1) * 3:page * 3], page, 3, len(expected_ids)) + seen.extend(actual) + if page == 2: + repeated = fetch({**parameters, 'page': page, 'pageSize': 3}) + check(f'{mode}/{name}/repeat-page-stable', + [row['id'] for row in repeated['items']] == actual) + check(f'{mode}/{name}/complete-no-duplicates', seen == expected_ids and len(set(seen)) == len(seen)) + + # Boundary sizes: singleton, exact multiple, partial last page, exact fit and oversized page. + sort = json.dumps({'field': 'status', 'direction': 'desc'}) + for size in [1, 4, 5, 12, 20]: + for page in range(1, (len(ids) + size - 1) // size + 2): + body = fetch({'sort': sort, 'page': page, 'pageSize': size}) + assert_page(f'boundaries/{size}/page-{page}', body, ids[(page - 1) * size:page * size], page, size, len(ids)) + + # Empty match on the first page must not advertise a next page. + body = fetch({'filter': json.dumps({'key': {'eq': 'does-not-exist'}}), 'page': 1, 'pageSize': 3}) + assert_page('empty-match', body, [], 1, 3, 0) + + # Filter to eight rows so accidentally dropping the filter on a next link is observable. + expected = sorted([i for i in ids if by_id[i]['rank'] >= 2], key=lambda i: -by_id[i]['rank']) + for mode in ['graphql', 'legacy', 'envelope']: + ordering = {'field': 'attributes.rank', 'direction': 'desc'} + expression = {'attributes': {'rank': {'ge': 2}}} + parameters = ({'filter': json.dumps({'filter': expression, 'groupBy': None, 'orderBy': ordering})} if mode == 'envelope' + else {'filter': json.dumps(expression) if mode == 'graphql' else 'rank=ge:2', + 'sort': json.dumps(ordering)}) + path = route + '?' + urllib.parse.urlencode({**parameters, 'page': 1, 'pageSize': 3}) + seen = [] + for page in range(1, 5): + status, body = request('GET', path) + assert status == 200, (status, body) + actual, href = assert_page(f'follow-next/{mode}/page-{page}', body, + expected[(page - 1) * 3:page * 3], page, 3, len(expected)) + seen.extend(actual) + if not href: + break + parsed = urllib.parse.urlsplit(href) + assert not parsed.scheme and not parsed.netloc, 'Expected relative same-runtime pagination link' + path = parsed.path + ('?' + parsed.query if parsed.query else '') + check(f'follow-next/{mode}/complete-filtered-set', seen == expected and len(set(seen)) == len(seen), + {'expected_ids': expected, 'actual_ids': seen}) diff --git a/api-tests/instance-queries/test_instance_queries.py b/api-tests/instance-queries/test_instance_queries.py new file mode 100644 index 0000000..dc38f5a --- /dev/null +++ b/api-tests/instance-queries/test_instance_queries.py @@ -0,0 +1,201 @@ +#!/usr/bin/env python3 +"""Instance query regression (#933 / #934): filtering, value limits and pagination. + +Publish/start/list through a locally built runtime and assert persisted data. + +Uses only Python's standard library and docker exec for read-only PostgreSQL checks. +Run only against a dedicated, authorized local test environment. +""" +import argparse +import json +from pathlib import Path +import subprocess +import time +import urllib.error +import urllib.parse +import urllib.request +import uuid + +from pagination_checks import run_pagination_checks + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--base-url', required=True) + parser.add_argument('--domain', default='core') + parser.add_argument('--pg-container', required=True) + parser.add_argument('--database', required=True) + parser.add_argument('--output', required=True) + modes = parser.add_mutually_exclusive_group() + modes.add_argument('--pagination-only', action='store_true', help='Run HTTP pagination/order/tie-break regression for PR #987.') + modes.add_argument('--value-limits-only', action='store_true', help='Run the #934 value-limit regression without the existing newline compatibility cases.') + args = parser.parse_args() + flow = 'json934-' + uuid.uuid4().hex[:10] + results = {'flow': flow, 'base_url': args.base_url, 'started_us': time.time_ns() // 1000, + 'requests': [], 'checks': [], 'findings': []} + output = Path(args.output) + output.parent.mkdir(parents=True, exist_ok=True) + + def request(method, path, body=None): + trace = uuid.uuid4().hex + headers = {'content-type': 'application/json', 'x-vnext-payload-mode': 'standard', + 'user_reference': '11111111-1111-1111-1111-111111111111', + 'x-request-id': str(uuid.uuid4()), 'x-device-id': 'issue934-e2e', + 'traceparent': f'00-{trace}-{uuid.uuid4().hex[:16]}-01'} + req = urllib.request.Request(args.base_url.rstrip('/') + path, + data=None if body is None else json.dumps(body).encode(), + headers=headers, method=method) + start = time.perf_counter() + try: + with urllib.request.urlopen(req, timeout=60) as response: + status, raw = response.status, response.read().decode() + except urllib.error.HTTPError as error: + status, raw = error.code, error.read().decode() + results['requests'].append({'method': method, 'path': path, 'status': status, + 'trace_id': trace, 'elapsed_ms': (time.perf_counter() - start) * 1000}) + return status, json.loads(raw) if raw else None + + def check(name, condition, detail=None): + results['checks'].append({'name': name, 'passed': bool(condition), 'detail': detail}) + print(('PASS ' if condition else 'FAIL ') + name, flush=True) + + def sql(query): + return subprocess.check_output( + ['docker', 'exec', '-i', args.pg_container, 'psql', '-U', 'postgres', '-d', args.database, + '-At', '-v', 'ON_ERROR_STOP=1'], input=query, text=True).strip() + + def query(filter_text): + path = f'/api/v1/{args.domain}/workflows/{flow}/instances?' + return request('GET', path + urllib.parse.urlencode({'filter': filter_text, 'pageSize': 100})) + + def row_ids(body): + # PagedResultDto exposes items; fail on unexpected shapes instead of hiding rows. + rows = body['items'] + assert isinstance(rows, list), body + return {row['id'] for row in rows} + + try: + definition = {'domain': args.domain, 'flow': 'sys-flows', 'key': flow, + 'version': '1.0.0', 'flowVersion': '1.0.0', 'tags': ['issue934-e2e'], + 'attributes': {'type': 'F', 'labels': [{'language': 'en-US', 'label': 'JSON filter regression'}], + 'startTransition': {'key': 'start', 'target': 'ready', 'versionStrategy': 'Minor', + 'labels': [{'language': 'en-US', 'label': 'Start'}]}, + 'states': [{'key': 'ready', 'stateType': 1, 'subType': 0, + 'versionStrategy': 'Minor', 'onEntries': [], 'onExits': [], + 'transitions': [], 'labels': [{'language': 'en-US', 'label': 'Ready'}]}]}} + status, body = request('POST', '/api/v1/definitions/publish', definition) + assert status == 200, (status, body) + payloads = [ + ('quote', 'a"b'), + ('backslash', 'C:\\new\\folder'), + ('invalid-escape', 'a\\qb'), + ('controls', 'a\nb\tc'), + ('extra-property', 'x","admin":true,"tail":"y'), + ('duplicate-key', 'x","name":"victim'), + ('sql-looking', "x' OR TRUE --"), + ] + rows = [] + for label, value in payloads + [('decoy', 'x'), ('victim', 'victim'), ('plain', 'plain'), + ('limit-value', 'z' * 1000), ('long-value', 'z' * 1001)]: + attributes = {'name': value, 'profile': {'name': value}, 'admin': label == 'decoy', + 'tail': 'y' if label == 'decoy' else 'original', 'testCase': label} + if args.pagination_only: + attributes['rank'] = len(rows) // 4 + 1 + status, body = request('POST', f'/api/v1/{args.domain}/workflows/{flow}/instances/start?sync=true', + {'key': label, 'attributes': attributes}) + assert status == 200 and body.get('id'), (status, body) + rows.append({'id': body['id'], **attributes}) + results['instances'] = rows + all_ids = {row['id'] for row in rows} + if args.pagination_only: + run_pagination_checks(request, check, args.domain, flow, rows, results) + else: + for label, value in ([] if args.value_limits_only else payloads): + exact = {row['id'] for row in rows if row['name'] == value} + for field in ['name', 'profile.name']: + for op in ['eq', 'ne']: + for fmt in ['graphql', 'legacy']: + filter_text = (json.dumps({'attributes': {field: {op: value}}}) if fmt == 'graphql' + else f'{field}={op}:{value}') + status, body = query(filter_text) + actual = row_ids(body) if status == 200 else set() + expected = exact if op == 'eq' else all_ids - exact + check(f'{fmt}/{field}/{op}/{label}', status == 200 and actual == expected, + {'status': status, 'expected_ids': sorted(expected), 'actual_ids': sorted(actual), + 'error': body if status != 200 else None}) + + # Instance data may exceed the filter-value limit; reject only the authored query operand. + for fmt in ['graphql', 'legacy']: + long_value = 'z' * 1001 + text = (json.dumps({'attributes': {'name': {'eq': long_value}}}) if fmt == 'graphql' + else 'name=eq:' + long_value) + status, body = query(text) + check(f'{fmt}/1001-character-value-rejected', status == 400 and '1000' in json.dumps(body), + {'status': status, 'body': body}) + + limit_value = 'z' * 1000 + text = (json.dumps({'attributes': {'name': {'eq': limit_value}}}) if fmt == 'graphql' + else 'name=eq:' + limit_value) + status, body = query(text) + expected = {row['id'] for row in rows if row['name'] == limit_value} + check(f'{fmt}/1000-character-value-accepted', status == 200 and row_ids(body) == expected, + {'status': status}) + + for op in ['in', 'nin', 'between']: + text = (json.dumps({'attributes': {'name': {op: ['plain', long_value]}}}) if fmt == 'graphql' + else f'name={op}:plain,{long_value}') + status, body = query(text) + check(f'{fmt}/{op}/oversized-element-rejected', status == 400 and '1000' in json.dumps(body), + {'status': status, 'body': body}) + if op == 'between': + continue # Attribute ranges require numeric/date operands, not arbitrary strings. + text = (json.dumps({'attributes': {'name': {op: ['plain', limit_value]}}}) if fmt == 'graphql' + else f'name={op}:plain,{limit_value}') + status, body = query(text) + included = {row['id'] for row in rows if row['name'] in ['plain', limit_value]} + expected = included if op == 'in' else all_ids - included + check(f'{fmt}/{op}/combined-length-over-1000-accepted', status == 200 and row_ids(body) == expected, + {'status': status}) + + leaf = {'attributes': {'name': {'eq': 'z' * 1001}}} + for label, expression in [ + ('and', {'and': [leaf]}), ('not', {'not': leaf}), + ('nested', {'attributes': {'profile': {'name': {'eq': 'z' * 1001}}}}), + ('envelope', {'filter': leaf, 'groupBy': {'fields': ['testCase'], 'aggregations': {'count': True}}}), + ]: + status, body = query(json.dumps(expression)) + check(f'graphql/{label}/oversized-value-rejected', status == 400 and '1000' in json.dumps(body), + {'status': status, 'body': body}) + + for fmt in ['graphql', 'legacy']: + oversized = 'z' * 5001 + text = (json.dumps({'attributes': {'name': {'eq': oversized}}}) if fmt == 'graphql' + else 'name=eq:' + oversized) + status, body = query(text) + check(f'{fmt}/total-filter-size-limit', status == 400, {'status': status, 'body': body}) + + for text in ['{"attributes":{"name":{"eq":"plain"}}', '{invalid-json}', 'name=unsupported:plain']: + status, body = query(text) + check('invalid-filter/' + text[:32], 400 <= status < 500, {'status': status, 'body': body}) + + schema = flow.replace('-', '_') # Generated by this script: only ASCII letters, digits and underscore. + persisted = json.loads(sql(f'SELECT coalesce(json_agg(t), \'[]\'::json) FROM ' + f'(SELECT i."Id" AS id, i."Status" AS status, d."Data" AS data ' + f'FROM "{schema}"."Instances" i JOIN "{schema}"."InstancesData" d ' + f'ON d."InstanceId"=i."Id" WHERE d."IsLatest"=true) t;')) + results['persisted'] = persisted + by_id = {row['id']: row for row in persisted} + check('postgres/exact-values-and-property-boundaries', set(by_id) == all_ids and all( + by_id[row['id']]['data'] == {k: v for k, v in row.items() if k != 'id'} for row in rows)) + check('postgres/no-faulted-instances', all(row['status'] == 'A' for row in persisted)) + finally: + results['ended_us'] = time.time_ns() // 1000 + output.write_text(json.dumps(results, indent=2, ensure_ascii=False)) + failures = [c for c in results['checks'] if not c['passed']] + print(json.dumps({'flow': flow, 'checks': len(results['checks']), 'failed': len(failures), + 'output': str(output)}), flush=True) + return bool(failures) + + +if __name__ == '__main__': + raise SystemExit(main())