diff --git a/CAPABILITIES.md b/CAPABILITIES.md index 513a7683f..6a6a74eaf 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -214,6 +214,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page, a redirect or a count that doesn't match the total is unreadable | On main ([#1538](https://github.com/OpenAdaptAI/OpenAdapt/pull/1538)) | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/llms-full.txt b/llms-full.txt index 490afde94..bb85c7e59 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -611,6 +611,7 @@ improvising or retrying when something is off. | Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | On main ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | +| Read a paged REST or FHIR record list whole (FHIR `next` links and `total`, REST `records_next_path`), or not at all: a page elsewhere, a failed page, a redirect or a count that doesn't match the total is unreadable | On main ([#1538](https://github.com/OpenAdaptAI/OpenAdapt/pull/1538)) | Synthetic (local FHIR and REST servers) | `effects: kind: rest` `records_next_path`, `kind: fhir`; [EFFECT_KIT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/EFFECT_KIT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Record where each step's time goes in every run report | On main ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Real app and synthetic | `timings` in `report.json` | diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 393d94754..81d3ec6f8 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/packages/flow/docs/EFFECT_KIT.md @@ -1035,7 +1035,7 @@ Per-kind required fields: | `ui_readback` | `ui_readback` section: `record_url`, `sign_in` (env-var names for the read-back account), `ready_selector`, `row_selector`, `fields`, `identity` | `url_params`, `empty_selector`, `not_found_selector`, `read_post_paths`, per-field `attribute` / `date_format` / `casefold`, `truncation_markers`, `navigation_timeout_s` | | `desktop_readback` | `desktop_readback` section: `session` (a `backend:`-style config for the read-back session), `search` (`box`, `query`), `ready`, `identity`, `fields` | `credentials` (env-var names), `sign_in`, `search.submit`, `open.results`, `not_found`, per-control `shows` / `region` / `at`, per-field `date_format` / `casefold`, `min_ocr_confidence`, `navigation_timeout_s`, `settle_s` | | `deferred_report` | `deferred_report` section: `report` (`format`, and for csv/text its columns or pattern and `accept`), `key`, `deadline_hours`, `ledger` | `report.reject`, `describe` | -| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `path_params`, `auth`, `records_drop_fields` | +| `rest` | `base_url` | `records_path` (may contain `{placeholder}`s), `records_key`, `records_next_path`, `records_total_path`, `path_params`, `auth`, `records_drop_fields` | | `graphql` | `base_url` + `graphql_query` | `graphql_variables` (`{param: ...}` entity binding), `graphql_records_path`, `graphql_freshness_field` + `graphql_freshness_window_s`, `auth` | | `fhir` | `base_url` | `resource_type`, `search_params`, `search_param_exprs`, `field_paths`, `access_token_env` or `smart_backend`, `verify_tls`, `records_drop_fields` | | `sql` | `sql_query` + (`sqlite_database` or `sql_driver`) | `sql_query_params`, `sql_connect_args`, `sql_password_env`, `records_drop_fields` | @@ -1045,6 +1045,31 @@ Per-kind required fields: | `document-hash` | `root` | `glob` | | `surfaces` | `surfaces`: name -> one plain verifier config (any kind above) | each effect's `probe` names its surface (`surface=`); an effect naming none, or an unknown one, halts | +A paged list is read whole or not at all. `fhir` follows the Bundle's `next` +links. `rest` follows the next page's URL at `records_next_path` (a dotted +path in the body, such as `links.next`; a relative URL resolves against the +page it came from). Without `records_next_path` one response is the whole +read, as before. A next page on another origin, a page that fails, more than +50 pages, or a FHIR `total` other than the number of matches read (included +resources don't count) makes the read unreadable (INDETERMINATE), never a +confirmation or a refutation from the first pages. + +REST has no standard total. With `records_total_path` (a dotted path in the +first page, such as `meta.total`), a read whose record count differs, or a +first page without the total, is unreadable. Without it, a REST read trusts +the page chain. + +A redirect is a failed read: the configured auth headers never follow it to +another host. Why a read was unreadable is in the verdict's reason and the +state's `detail.read_problem`, for example +`next link origin https://localhost:9300 != base_url http://localhost:9520`. + +OpenEMR builds its REST next links from its configured site address +(`site_addr_oath`), not from the address the request used. Read through a +mapped port or another host name, every next link names another origin and +the read is unreadable. Set `base_url` to the site address OpenEMR is +configured with. + `records_drop_fields` (`rest`, `sql`, `fhir`) removes fields from every record before judging. `[id]` compares records by content: a record whose id stayed the same but whose fields changed reads as one record removed and one added, diff --git a/packages/flow/openadapt_flow/deployment.py b/packages/flow/openadapt_flow/deployment.py index 334f2170d..f5b482560 100644 --- a/packages/flow/openadapt_flow/deployment.py +++ b/packages/flow/openadapt_flow/deployment.py @@ -387,6 +387,15 @@ class EffectsConfig(BaseModel): base_url: Optional[str] = None records_path: str = "/api/db" records_key: Optional[str] = "records" + #: ``rest`` only: dotted path in the response body to the next page's URL + #: (``links.next``). Set, the read follows every page on the same origin + #: and reads as unreadable when a page fails or the chain is too long. + #: Unset: one response is the whole read, as before. + records_next_path: Optional[str] = None + #: ``rest`` only: dotted path in the first page's body to the number of + #: records the read should return (``meta.total``). Set, a read whose + #: count differs reads as unreadable. Unset: the read trusts the chain. + records_total_path: Optional[str] = None #: Secret-isolated auth headers for the records read (bearer_env / #: header+value_env / basic_env). Never a credential literal. auth: Optional[AuthRef] = None @@ -612,7 +621,12 @@ def _serialize_compatible(self, handler: Any) -> dict[str, Any]: """Keep configs without a desktop read-back byte-identical.""" data: dict[str, Any] = handler(self) - for added in ("desktop_readback", "deferred_report"): + for added in ( + "desktop_readback", + "deferred_report", + "records_next_path", + "records_total_path", + ): if data.get(added, "") is None: data.pop(added, None) return data @@ -1964,6 +1978,8 @@ def sanitized(value: Any, *, key: str = "") -> Any: "fresh_session", "desktop_readback", "deferred_report", + "records_next_path", + "records_total_path", ): if config.get(added) is None: config.pop(added, None) @@ -2116,6 +2132,8 @@ def _build_effect_verifier_unredacted( timeout_s=cfg.timeout_s, poll_interval_s=cfg.poll_interval_s, drop_fields=tuple(cfg.records_drop_fields), + next_path=cfg.records_next_path, + total_path=cfg.records_total_path, ) if kind == "fhir": diff --git a/packages/flow/openadapt_flow/runtime/effects/fhir.py b/packages/flow/openadapt_flow/runtime/effects/fhir.py index 912e340b6..be28e37bc 100644 --- a/packages/flow/openadapt_flow/runtime/effects/fhir.py +++ b/packages/flow/openadapt_flow/runtime/effects/fhir.py @@ -67,6 +67,35 @@ } +#: A search result longer than this many pages reads as unreadable, never as +#: the first pages. +MAX_PAGES = 50 + + +def _next_link(bundle: dict[str, Any]) -> Optional[str]: + """The Bundle's ``next`` page URL, if any.""" + + links = bundle.get("link") + if not isinstance(links, list): + return None + for link in links: + if isinstance(link, dict) and link.get("relation") == "next": + url = link.get("url") + return url if isinstance(url, str) and url else None + return None + + +def _origin(url: str) -> str: + from urllib.parse import urlsplit + + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}" + + +def _same_origin(url: str, base_url: str) -> bool: + return _origin(url) == _origin(base_url) + + class AccessTokenSource(Protocol): """Supplies bearer tokens that can expire.""" @@ -155,6 +184,8 @@ def __init__( # (``requests``-style) session; flipped only for the httpx fallback # client built in :meth:`_get_session`. self._verify_per_request = True + #: Why the last read was unreadable, for the state and the verdict. + self.read_problem: Optional[str] = None # -- transport ---------------------------------------------------------- @@ -213,61 +244,121 @@ def _search(self) -> Optional[list[dict[str, Any]]]: transport error, non-2xx (incl. auth 401/403), non-Bundle body, or malformed entries. Never raises. """ + self.read_problem = None url = f"{self.base_url}/{self.resource_type}" headers = self._headers() if headers is None: + self.read_problem = "no access token" return None session = self._get_session() # ``verify`` is passed per request only for a ``requests``-style client; # the httpx fallback already bound it at construction (see - # ``_get_session``), and passing it again would raise. - tls_kwargs = {"verify": self.verify_tls} if self._verify_per_request else {} - try: - resp = session.get( - url, - params=self.search_params, - headers=headers, - timeout=self.timeout_s, - **tls_kwargs, - ) - except Exception: # noqa: BLE001 - transport failure is unreadable - return None - if resp.status_code == 401 and self.token_source is not None: - self.token_source.invalidate() - if resp.status_code // 100 != 2: - return None - try: - bundle = resp.json() - except Exception: # noqa: BLE001 - unparseable body is unreadable - return None - if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle": - return None - entries = bundle.get("entry", []) - if not isinstance(entries, list): + # ``_get_session``), and passing it again would raise. A ``requests`` + # session must not follow a redirect with the bearer token. + tls_kwargs: dict[str, Any] = ( + {"verify": self.verify_tls} if self._verify_per_request else {} + ) + if (type(session).__module__ or "").startswith("requests"): + tls_kwargs["allow_redirects"] = False + + def get(target: str, params: Optional[dict[str, str]]) -> Any: + """One page: the parsed Bundle, or None with ``read_problem``.""" + + try: + kwargs = dict(tls_kwargs) + if params is not None: + kwargs["params"] = params + resp = session.get( + target, headers=headers, timeout=self.timeout_s, **kwargs + ) + except Exception as exc: # noqa: BLE001 - transport failure is unreadable + self.read_problem = f"GET {target} failed: {type(exc).__name__}" + return None + if resp.status_code == 401 and self.token_source is not None: + self.token_source.invalidate() + if resp.status_code // 100 != 2: + self.read_problem = f"GET {target} returned HTTP {resp.status_code}" + return None + landed = getattr(resp, "url", None) + landed = str(landed) if landed is not None else "" + if landed and _origin(landed) != _origin(target): + self.read_problem = ( + f"GET {target} was redirected to origin {_origin(landed)}" + ) + return None + try: + bundle = resp.json() + except Exception: # noqa: BLE001 - unparseable body is unreadable + self.read_problem = f"GET {target} returned a body that is not JSON" + return None + if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle": + self.read_problem = f"GET {target} returned no Bundle" + return None + return bundle + + bundle = get(url, self.search_params) + if bundle is None: return None out: list[dict[str, Any]] = [] - for entry in entries: - if not isinstance(entry, dict): + matches = 0 + total: Any = bundle.get("total") + for _page in range(MAX_PAGES): + entries = bundle.get("entry", []) + if not isinstance(entries, list): + self.read_problem = "a Bundle's entry is not a list" return None - resource = entry.get("resource") - if not isinstance(resource, dict): + for entry in entries: + if not isinstance(entry, dict): + self.read_problem = "a Bundle entry is not an object" + return None + resource = entry.get("resource") + if not isinstance(resource, dict): + self.read_problem = "a Bundle entry has no resource" + return None + mode = (entry.get("search") or {}).get("mode") + if mode not in ("include", "outcome"): + matches += 1 + out.append(self._flatten(resource)) + next_url = _next_link(bundle) + if next_url is None: + break + if not _same_origin(next_url, self.base_url): + # A next page elsewhere is not this system of record. + self.read_problem = ( + f"next link origin {_origin(next_url)} != base_url " + f"{_origin(self.base_url)}" + ) return None - out.append(self._flatten(resource)) + bundle = get(next_url, None) + if bundle is None: + return None # a lost page makes the set partial + else: + self.read_problem = f"more than {MAX_PAGES} pages" + return None # never a truncated read + if isinstance(total, int) and not isinstance(total, bool) and total != matches: + # The server counts a different number of matches than it returned. + self.read_problem = ( + f"the Bundle's total is {total}, the read returned {matches} matches" + ) + return None return drop_record_fields(out, self.drop_fields) # -- EffectVerifier protocol -------------------------------------------- def capture_pre_state(self, context: Any = None) -> EffectState: records = self._search() + detail: dict[str, Any] = { + "base_url": self.base_url, + "resource_type": self.resource_type, + "search_params": self.search_params, + } + if records is None and self.read_problem: + detail["read_problem"] = self.read_problem return EffectState( substrate=self.substrate, reachable=records is not None, records=records or [], - detail={ - "base_url": self.base_url, - "resource_type": self.resource_type, - "search_params": self.search_params, - }, + detail=detail, ) def verify( @@ -278,6 +369,10 @@ def verify( while True: current = self._search() last = judge_records(expected, before, current, substrate=self.substrate) + if current is None and self.read_problem: + last = last.model_copy( + update={"reason": f"{last.reason} ({self.read_problem})"} + ) if last.confirmed or time.monotonic() >= deadline: return last time.sleep(self.poll_interval_s) diff --git a/packages/flow/openadapt_flow/runtime/effects/rest.py b/packages/flow/openadapt_flow/runtime/effects/rest.py index d9e9f608d..390d7ce6d 100644 --- a/packages/flow/openadapt_flow/runtime/effects/rest.py +++ b/packages/flow/openadapt_flow/runtime/effects/rest.py @@ -22,6 +22,7 @@ import time from typing import Any, Optional +from urllib.parse import urljoin, urlsplit from openadapt_flow.runtime.effects._common import drop_record_fields, judge_records from openadapt_flow.runtime.effects.adapter import VerifierAdapterBase @@ -32,6 +33,41 @@ ) from openadapt_flow.verification import VerificationTier +#: A paged read longer than this many pages reads as unreadable. +MAX_PAGES = 50 + +_UNREADABLE = object() + + +def _dotted(body: Any, path: str) -> Any: + node = body + for part in path.split("."): + if not isinstance(node, dict): + return None + node = node.get(part) + return node + + +def _origin(url: str) -> str: + parts = urlsplit(url) + return f"{parts.scheme}://{parts.netloc}" + + +def _same_origin(url: str, base_url: str) -> bool: + return _origin(url) == _origin(base_url) + + +def _no_redirects(session: Any) -> dict[str, Any]: + """Keyword that stops ``requests`` following a redirect. + + A redirect could carry the configured auth headers to another host, so a + 3xx is a failed read. ``httpx`` doesn't follow redirects by default; an + injected session keeps its own ``get`` signature. + """ + + module = type(session).__module__ or "" + return {"allow_redirects": False} if module.startswith("requests") else {} + class RestRecordVerifier(VerifierAdapterBase): """Verify effects against a JSON REST system of record. @@ -55,6 +91,17 @@ class RestRecordVerifier(VerifierAdapterBase): records by their whole content, so a change to another record reads as that record removed and a new one added, and the collateral-loss check catches it). Empty -> records as read. + next_path: Dotted path in the body to the next page's URL. Set, every + page on the same origin is read (at most :data:`MAX_PAGES`); a + failed page or a longer chain reads as unreadable. Unset, one + response is the whole read. + total_path: Dotted path in the first page's body to the number of + records the server holds for the read. Set, a read whose record + count differs, or a body without it, reads as unreadable. Unset, + the read trusts the page chain. + + Why the last read was unreadable is in :attr:`read_problem` (and in the + state's ``detail`` and the verdict's reason). """ substrate = "rest" @@ -71,8 +118,13 @@ def __init__( timeout_s: float = 5.0, poll_interval_s: float = 0.2, drop_fields: tuple[str, ...] = (), + next_path: Optional[str] = None, + total_path: Optional[str] = None, ) -> None: self.base_url = base_url.rstrip("/") + self.next_path = next_path + self.total_path = total_path + self.read_problem: Optional[str] = None self.drop_fields = tuple(drop_fields) self.records_path = records_path self.records_key = records_key @@ -107,52 +159,118 @@ def _get_session(self) -> Any: self._session = requests.Session() return self._session - def _fetch_records(self) -> Optional[list[dict[str, Any]]]: - """GET the system-of-record document and extract the records list. + def _get_json(self, url: str) -> Any: + """One GET; the parsed body, or ``_UNREADABLE``. Never raises.""" - Returns ``None`` -- read as unreadable, forcing INDETERMINATE -- on - any transport error, non-2xx status, or shape mismatch. Never raises. - """ - url = f"{self.base_url}{self.records_path}" + session = self._get_session() try: # ``headers`` is passed ONLY when configured, so injected # ``requests``-style sessions with a narrower ``get`` signature # (tests, custom oracles) keep working exactly as before. if self.headers: - resp = self._get_session().get( - url, timeout=self.timeout_s, headers=self.headers + resp = session.get( + url, + timeout=self.timeout_s, + headers=self.headers, + **_no_redirects(session), ) else: - resp = self._get_session().get(url, timeout=self.timeout_s) - except Exception: # noqa: BLE001 - any transport failure is unreadable - return None + resp = session.get(url, timeout=self.timeout_s) + except Exception as exc: # noqa: BLE001 - any transport failure is unreadable + self.read_problem = f"GET {url} failed: {type(exc).__name__}" + return _UNREADABLE if resp.status_code // 100 != 2: - return None + self.read_problem = f"GET {url} returned HTTP {resp.status_code}" + return _UNREADABLE + landed = getattr(resp, "url", None) + if isinstance(landed, str) and landed and not _same_origin(landed, url): + self.read_problem = f"GET {url} was redirected to origin {_origin(landed)}" + return _UNREADABLE try: - body = resp.json() + return resp.json() except Exception: # noqa: BLE001 - unparseable body is unreadable - return None + self.read_problem = f"GET {url} returned a body that is not JSON" + return _UNREADABLE + + def _records_of(self, body: Any) -> Optional[list[Any]]: if self.records_key is None: records = body elif isinstance(body, dict): records = body.get(self.records_key) else: return None - if not isinstance(records, list): + return records if isinstance(records, list) else None + + def _fetch_records(self) -> Optional[list[dict[str, Any]]]: + """GET the system-of-record document and extract the records list. + + Returns ``None`` -- read as unreadable, forcing INDETERMINATE -- on + any transport error, non-2xx status, or shape mismatch, and, with + ``next_path``, on a failed page, a page on another origin or a chain + longer than :data:`MAX_PAGES`. Never raises. + """ + self.read_problem = None + url = f"{self.base_url}{self.records_path}" + collected: list[Any] = [] + total: Any = None + for page in range(MAX_PAGES): + body = self._get_json(url) + if body is _UNREADABLE: + return None + records = self._records_of(body) + if records is None: + self.read_problem = f"GET {url}: no records list in the body" + return None + if page == 0 and self.total_path is not None: + total = _dotted(body, self.total_path) + collected.extend(records) + if self.next_path is None: + break + following = _dotted(body, self.next_path) + if following in (None, ""): + break + if not isinstance(following, str): + self.read_problem = f"the next link at {self.next_path} is not a URL" + return None + following = urljoin(url, following) + if not _same_origin(following, self.base_url): + # OpenEMR, for one, builds next links from its configured site + # address, which can differ from the address the read uses. + self.read_problem = ( + f"next link origin {_origin(following)} != base_url " + f"{_origin(self.base_url)}" + ) + return None + url = following + else: + self.read_problem = f"more than {MAX_PAGES} pages" return None + if self.total_path is not None: + if isinstance(total, bool) or not isinstance(total, int): + self.read_problem = f"no record total at {self.total_path}" + return None + if total != len(collected): + self.read_problem = ( + f"the record total at {self.total_path} is {total}, the read " + f"returned {len(collected)}" + ) + return None return drop_record_fields( - [r for r in records if isinstance(r, dict)], self.drop_fields + [r for r in collected if isinstance(r, dict)], self.drop_fields ) # -- EffectVerifier protocol -------------------------------------------- def capture_pre_state(self, context: Any = None) -> EffectState: records = self._fetch_records() + detail: dict[str, Any] = {"url": f"{self.base_url}{self.records_path}"} + if records is None and self.read_problem: + detail["read_problem"] = self.read_problem return EffectState( substrate=self.substrate, reachable=records is not None, records=records or [], - detail={"url": f"{self.base_url}{self.records_path}"}, + detail=detail, ) def verify( @@ -167,6 +285,10 @@ def verify( while True: current = self._fetch_records() last = judge_records(expected, before, current, substrate=self.substrate) + if current is None and self.read_problem: + last = last.model_copy( + update={"reason": f"{last.reason} ({self.read_problem})"} + ) if last.confirmed or time.monotonic() >= deadline: return last time.sleep(self.poll_interval_s) diff --git a/packages/flow/tests/test_effect_paging.py b/packages/flow/tests/test_effect_paging.py new file mode 100644 index 000000000..8f7740a0c --- /dev/null +++ b/packages/flow/tests/test_effect_paging.py @@ -0,0 +1,412 @@ +"""A paged read is the whole read, or unreadable: never its first page. + +The FHIR verifier follows a Bundle's ``next`` links and checks ``total``; the +REST verifier follows ``records_next_path`` when it is set. A page on another +origin, a failed page, a chain longer than ``MAX_PAGES`` or a ``total`` above +what was read makes the read unreadable, so no check confirms or refutes from +part of the store. Records are made up. +""" + +from __future__ import annotations + +import json +import threading +from collections.abc import Callable +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + +import pytest + +from openadapt_flow.deployment import EffectsConfig, build_effect_verifier +from openadapt_flow.runtime.effects import fhir, rest +from openadapt_flow.runtime.effects.fhir import FhirEffectVerifier +from openadapt_flow.runtime.effects.rest import RestRecordVerifier + +PATIENTS = [{"id": str(i), "fname": f"F{i}", "lname": f"L{i}"} for i in range(1, 6)] + +Page = Callable[[str, int], tuple] +#: Request headers every test server saw, in order. +SEEN: list[dict] = [] + + +@pytest.fixture +def serve(): + servers: list[ThreadingHTTPServer] = [] + + def start(page: Page) -> str: + class Handler(BaseHTTPRequestHandler): + def do_GET(self) -> None: # noqa: N802 - http.server API + parts = urlsplit(self.path) + number = int(parse_qs(parts.query).get("page", ["0"])[0]) + SEEN.append(dict(self.headers)) + status, body, *extra = page(parts.path, number) + data = json.dumps(body).encode() + self.send_response(status) + for name, value in (extra[0] if extra else {}).items(): + self.send_header(name, value) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *_args) -> None: + return + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + servers.append(server) + return f"http://127.0.0.1:{server.server_address[1]}" + + yield start + for server in servers: + server.shutdown() + server.server_close() + + +# -- FHIR ------------------------------------------------------------------------ + + +def _bundle(url: str, number: int, *, per: int = 2, total: object = None) -> dict: + rows = PATIENTS[number * per : number * per + per] + bundle: dict = { + "resourceType": "Bundle", + "type": "searchset", + "entry": [ + { + "resource": { + "resourceType": "Patient", + "id": p["id"], + "name": [{"family": p["lname"], "given": [p["fname"]]}], + } + } + for p in rows + ], + "link": [{"relation": "self", "url": f"{url}/fhir/Patient?page={number}"}], + } + if (number + 1) * per < len(PATIENTS): + bundle["link"].append( + {"relation": "next", "url": f"{url}/fhir/Patient?page={number + 1}"} + ) + if total is not None: + bundle["total"] = total + return bundle + + +def _fhir(url: str) -> FhirEffectVerifier: + return FhirEffectVerifier( + f"{url}/fhir", + resource_type="Patient", + field_paths={"id": "id", "lname": "name.0.family"}, + access_token="test-token-not-a-secret", + timeout_s=2.0, + ) + + +def _read(verifier) -> list | None: + state = verifier.capture_pre_state() + return list(state.records) if state.reachable else None + + +def test_fhir_reads_every_page(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, total=5))) + holder["url"] = url + records = _read(_fhir(url)) + assert records is not None + assert [r["id"] for r in records] == ["1", "2", "3", "4", "5"] + + +def test_fhir_one_page_without_a_next_link_is_the_whole_read(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10))) + holder["url"] = url + assert len(_read(_fhir(url)) or []) == 5 + + +def test_fhir_next_page_on_another_origin_is_unreadable(serve) -> None: + def page(path: str, n: int) -> tuple[int, dict]: + bundle = _bundle("http://203.0.113.9", n) + return 200, bundle + + url = serve(page) + assert _read(_fhir(url)) is None + + +def test_fhir_failed_later_page_is_unreadable(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + if n == 2: + return 503, {"resourceType": "OperationOutcome"} + return 200, _bundle(holder["url"], n) + + holder["url"] = serve(page) + assert _read(_fhir(holder["url"])) is None + + +def test_fhir_total_above_the_read_is_unreadable(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10, total=7))) + holder["url"] = url + assert _read(_fhir(url)) is None + + +def test_fhir_endless_next_links_are_unreadable(serve, monkeypatch) -> None: + monkeypatch.setattr(fhir, "MAX_PAGES", 4) + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + bundle = _bundle(holder["url"], 0) + bundle["link"][-1]["url"] = f"{holder['url']}/fhir/Patient?page={n + 1}" + return 200, bundle + + holder["url"] = serve(page) + assert _read(_fhir(holder["url"])) is None + + +# -- REST ------------------------------------------------------------------------ + + +def _rest_page( + n: int, *, per: int = 2, next_url: Callable[[int], str] | None = None +) -> dict: + rows = PATIENTS[n * per : n * per + per] + more = (n + 1) * per < len(PATIENTS) + nxt = next_url(n + 1) if (more and next_url) else None + return {"data": rows, "links": {"next": nxt}} + + +def _rest(url: str, next_path: str | None = "links.next") -> RestRecordVerifier: + return RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + timeout_s=2.0, + next_path=next_path, + ) + + +def test_rest_follows_relative_and_absolute_next_links(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple[int, dict]: + # Page 1 links absolutely, page 2 relatively. + return 200, _rest_page( + n, + next_url=lambda m: ( + f"{holder['url']}/api/patient?page={m}" + if m == 1 + else f"/api/patient?page={m}" + ), + ) + + holder["url"] = serve(page) + records = _read(_rest(holder["url"])) + assert records is not None and [r["id"] for r in records] == list("12345") + + +def test_rest_without_next_path_reads_one_response(serve) -> None: + holder: dict = {} + holder["url"] = serve( + lambda path, n: ( + 200, + _rest_page(n, next_url=lambda m: f"/api/patient?page={m}"), + ) + ) + records = _read(_rest(holder["url"], next_path=None)) + assert records is not None and len(records) == 2 + + +def test_rest_next_page_on_another_origin_is_unreadable(serve) -> None: + url = serve( + lambda path, n: ( + 200, + _rest_page( + n, next_url=lambda m: f"http://203.0.113.9/api/patient?page={m}" + ), + ) + ) + assert _read(_rest(url)) is None + + +def test_rest_failed_later_page_is_unreadable(serve) -> None: + def page(path: str, n: int) -> tuple[int, dict]: + if n == 1: + return 500, {"error": "unavailable"} + return 200, _rest_page(n, next_url=lambda m: f"/api/patient?page={m}") + + assert _read(_rest(serve(page))) is None + + +def test_rest_a_next_link_that_is_not_text_is_unreadable(serve) -> None: + url = serve(lambda path, n: (200, {"data": PATIENTS[:2], "links": {"next": 7}})) + assert _read(_rest(url)) is None + + +def test_rest_endless_next_links_are_unreadable(serve, monkeypatch) -> None: + monkeypatch.setattr(rest, "MAX_PAGES", 4) + url = serve( + lambda path, n: ( + 200, + {"data": PATIENTS[:1], "links": {"next": f"/api/patient?page={n + 1}"}}, + ) + ) + assert _read(_rest(url)) is None + + +def test_records_next_path_in_a_config(serve) -> None: + holder: dict = {} + holder["url"] = serve( + lambda path, n: ( + 200, + _rest_page(n, next_url=lambda m: f"/api/patient?page={m}"), + ) + ) + section = { + "kind": "rest", + "base_url": holder["url"], + "records_path": "/api/patient", + "records_key": "data", + "records_next_path": "links.next", + } + verifier = build_effect_verifier(EffectsConfig.model_validate(section)) + assert len(_read(verifier) or []) == 5 + # Unset, the config serializes as before, so existing digests hold. + plain = EffectsConfig.model_validate( + {k: v for k, v in section.items() if k != "records_next_path"} + ) + assert "records_next_path" not in plain.model_dump(mode="json") + + +# -- review of #1538 --------------------------------------------------------------- + + +def test_fhir_total_below_the_read_is_unreadable(serve) -> None: + holder: dict = {} + url = serve(lambda path, n: (200, _bundle(holder["url"], n, per=10, total=3))) + holder["url"] = url + verifier = _fhir(url) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"] == ( + "the Bundle's total is 3, the read returned 5 matches" + ) + + +def test_fhir_included_resources_are_not_matches(serve) -> None: + holder: dict = {} + + def page(path: str, n: int) -> tuple: + bundle = _bundle(holder["url"], n, per=10, total=5) + bundle["entry"].append( + { + "resource": {"resourceType": "Practitioner", "id": "dr1"}, + "search": {"mode": "include"}, + } + ) + return 200, bundle + + holder["url"] = serve(page) + assert len(_read(_fhir(holder["url"])) or []) == 6 + + +def test_fhir_next_origin_is_named(serve) -> None: + url = serve(lambda path, n: (200, _bundle("http://203.0.113.9", n))) + state = _fhir(url).capture_pre_state() + assert state.detail["read_problem"] == ( + f"next link origin http://203.0.113.9 != base_url {url}" + ) + + +def test_openemr_next_links_name_the_site_address(serve) -> None: + """OpenEMR builds REST next links from its configured site address. Read + through a mapped port, the link names another origin: unreadable, with + both origins in the reason, never page one alone.""" + + site = "https://localhost:9300" + + def page(path: str, n: int) -> tuple: + return 200, { + "validationErrors": [], + "internalErrors": [], + "data": PATIENTS[n * 2 : n * 2 + 2], + "links": {"next": f"{site}/apis/default/api/patient?page={n + 1}"}, + } + + url = serve(page) + verifier = _rest(url) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"] == ( + f"next link origin {site} != base_url {url}" + ) + + +def _rest_total(url: str, path: str = "meta.total") -> RestRecordVerifier: + return RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + timeout_s=2.0, + next_path="links.next", + total_path=path, + ) + + +@pytest.mark.parametrize("total,ok", [(5, True), (6, False), (4, False), (None, False)]) +def test_rest_total_path(serve, total, ok) -> None: + def page(path: str, n: int) -> tuple: + body = _rest_page(n, next_url=lambda m: f"/api/patient?page={m}") + if total is not None: + body["meta"] = {"total": total} + return 200, body + + verifier = _rest_total(serve(page)) + state = verifier.capture_pre_state() + assert state.reachable is ok + if not ok: + assert "total" in state.detail["read_problem"] + + +def test_a_redirect_never_carries_the_auth_headers(serve) -> None: + """A 3xx is a failed read; the configured headers never reach the other + host.""" + + SEEN.clear() + other = serve(lambda path, n: (200, {"data": PATIENTS, "links": {}})) + url = serve(lambda path, n: (302, {}, {"Location": f"{other}/api/patient"})) + verifier = RestRecordVerifier( + url, + records_path="/api/patient", + records_key="data", + headers={"X-Api-Key": "test-key-not-a-secret"}, + timeout_s=2.0, + ) + state = verifier.capture_pre_state() + assert not state.reachable + assert state.detail["read_problem"].endswith("returned HTTP 302") + assert len(SEEN) == 1 # the other host was never asked + + SEEN.clear() + fhir_url = serve(lambda path, n: (302, {}, {"Location": f"{other}/fhir/Patient"})) + state = _fhir(fhir_url).capture_pre_state() + assert not state.reachable + assert len(SEEN) == 1 + + +def test_the_verdict_names_why_the_read_failed(serve) -> None: + from openadapt_flow.runtime.effects import Effect, EffectKind + + url = serve( + lambda path, n: ( + 200, + _rest_page( + n, next_url=lambda m: f"http://203.0.113.9/api/patient?page={m}" + ), + ) + ) + verifier = _rest(url) + before = verifier.capture_pre_state() + effect = Effect(kind=EffectKind.RECORD_WRITTEN, match={"lname": "L5"}, timeout_s=0) + verdict = verifier.verify(effect, before) + assert "next link origin http://203.0.113.9" in verdict.reason