From 8af3de913d69be14e60298727ec31f4c17c4192f Mon Sep 17 00:00:00 2001 From: Bill Lakenan Date: Thu, 6 Aug 2026 19:46:02 -0400 Subject: [PATCH] chore(scripts): add copy_measure_content.py for cross-server measure content transfer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Copies Measure / Library / ValueSet / CodeSystem from one FHIR server to another as batch Bundles of PUT entries, reusing the shape of fhir_client.push_resources (batch not transaction, so the target does not enforce cross-entry referential integrity). Written to load Lenny's connectathon measure content onto an external MCS. Used to move 185 resources to the CMS connectathon server. Safety properties, each of which came from a bug found while building it: - A read-only collision check runs before any write and aborts without --overwrite when the target holds different content at the same IDs. - Only 404/410 counts as "absent". A first version treated any error as "not there", so an expired token made an unreadable server look like an empty one and reported "185 new, 0 collisions" — a green light to overwrite based on 401s. - A preflight does one authenticated search before real work, because /metadata is frequently unsecured and proves nothing about the token. - Verification is by direct read, not search count. HAPI's index refresh is async; a measured 40s lag on the CMS server made a successful write look like a failure. Search count is still reported, labelled as informational. Secrets are never command-line arguments (argv is world-readable via ps and lands in shell history). The access token comes from $TARGET_FHIR_TOKEN, a 0600 cache file, or an OAuth2 client_credentials grant; client credentials come from env vars or the macOS keychain. Expired JWTs are skipped by `exp`, and a 401 mid-run triggers one refresh-and-retry. Tests cover the token-expiry and collision logic. CI's unit job runs `cd backend && pytest tests/`, so it does not collect scripts/tests/*.py — run them manually when touching the script. --- scripts/copy_measure_content.py | 882 +++++++++++++++++++++ scripts/tests/test_copy_measure_content.py | 146 ++++ 2 files changed, 1028 insertions(+) create mode 100755 scripts/copy_measure_content.py create mode 100644 scripts/tests/test_copy_measure_content.py diff --git a/scripts/copy_measure_content.py b/scripts/copy_measure_content.py new file mode 100755 index 00000000..89cb1f53 --- /dev/null +++ b/scripts/copy_measure_content.py @@ -0,0 +1,882 @@ +#!/usr/bin/env python3 +"""Copy measure content (Measure / Library / ValueSet / CodeSystem) from one +FHIR server to another. + +Reads every resource of the selected types from --source, then writes them to +--target as FHIR `batch` Bundles of PUT entries (same shape as +`backend/app/services/fhir_client.py::push_resources`). Batch, not transaction, +so the target does not enforce cross-entry referential integrity. + +Resources are written in dependency order: CodeSystem, ValueSet, Library, +Measure. Libraries missing `url` get `url = "Library/{id}"` backfilled so HAPI +8.x can resolve canonical references (mirrors `_normalize_measure_def`). + +Before writing anything it runs a read-only collision check and refuses to +proceed, without --overwrite, if the target holds different content at any of +the same IDs. A probe that cannot read the target (401/403/5xx) is an error, not +an empty target. + +AUTH + Secrets are never command-line arguments: argv is readable by any process via + `ps` and is recorded in shell history. + + Access token, first usable wins: + 1. $TARGET_FHIR_TOKEN + 2. the cache file (default ~/.lenny-target-token) + 3. minted via the OAuth2 client_credentials grant + A cached token whose JWT `exp` has passed is skipped, and a 401 mid-run + triggers one refresh-and-retry, so a long push cannot die on expiry. + + Client credentials, first found wins: + 1. $TARGET_CLIENT_ID / $TARGET_CLIENT_SECRET + 2. macOS keychain, service `lenny-target-oauth` + + Recommended one-time keychain setup (`-w` with no value prompts, so the + secret never reaches argv or history): + security add-generic-password -s lenny-target-oauth -a client_id -w + security add-generic-password -s lenny-target-oauth -a client_secret -w + +Usage: + python3 scripts/copy_measure_content.py --target https://example.org/fhir --check-only + python3 scripts/copy_measure_content.py --target https://example.org/fhir + python3 scripts/copy_measure_content.py --target https://example.org/fhir --refresh-token +""" + +from __future__ import annotations + +import argparse +import base64 +import json +import os +import pathlib +import subprocess +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any + +# Dependency order: referenced content first, Measures last. +DEFAULT_TYPES = ["CodeSystem", "ValueSet", "Library", "Measure"] +DEFAULT_SOURCE = "http://localhost:8080/fhir" +PAGE_SIZE = 200 +FHIR_JSON = "application/fhir+json" +DEFAULT_TOKEN_FILE = "~/.lenny-target-token" +DEFAULT_KEYCHAIN_SERVICE = "lenny-target-oauth" + + +def _default_token_url(target: str) -> str: + """Guess the token endpoint as the target's origin + /token.""" + parsed = urllib.parse.urlparse(target) + return f"{parsed.scheme}://{parsed.netloc}/token" + + +class TransferError(Exception): + pass + + +class AuthError(TransferError): + pass + + +def _keychain_get(service: str, account: str) -> str | None: + """Read one secret from the macOS login keychain. None if absent or not macOS.""" + if sys.platform != "darwin": + return None + try: + result = subprocess.run( + ["security", "find-generic-password", "-s", service, "-a", account, "-w"], + capture_output=True, + text=True, + timeout=15, + ) + except (OSError, subprocess.SubprocessError): + return None + return result.stdout.strip() or None if result.returncode == 0 else None + + +def _jwt_expiry(token: str) -> int | None: + """Read the `exp` claim from a JWT without verifying it. + + Signature verification is the server's job. We only need to know whether it is + worth sending. Returns None for opaque (non-JWT) tokens. + """ + parts = token.split(".") + if len(parts) != 3: + return None + try: + payload = parts[1] + "=" * (-len(parts[1]) % 4) + claims = json.loads(base64.urlsafe_b64decode(payload)) + except (ValueError, json.JSONDecodeError): + return None + exp = claims.get("exp") + return exp if isinstance(exp, int) else None + + +class TokenSource: + """Supplies a bearer token for the target, minting a fresh one when needed. + + Resolution order for an existing token: TARGET_FHIR_TOKEN env var, then the + cache file. Either is used only if it is not already expired. When there is no + usable token — or the server rejects one mid-run — the client-credentials grant + mints a new one and rewrites the cache with 0600 permissions. + + Client credentials are looked up from the environment first, then the macOS + keychain. They are never accepted as command-line arguments: argv is world- + readable via `ps` and lands in shell history. + """ + + SKEW_SECONDS = 120 # refresh early rather than race the clock mid-batch + + def __init__( + self, + token_url: str | None, + cache_path: pathlib.Path | None, + keychain_service: str, + token_env: str, + id_env: str, + secret_env: str, + ) -> None: + self.token_url = token_url + self.cache_path = cache_path + self.keychain_service = keychain_service + self.token_env = token_env + self.id_env = id_env + self.secret_env = secret_env + self._token: str | None = None + self.last_source = "none" + self.refresh_count = 0 + + def client_credentials(self) -> tuple[str, str] | None: + client_id = os.environ.get(self.id_env, "").strip() or _keychain_get(self.keychain_service, "client_id") + client_secret = os.environ.get(self.secret_env, "").strip() or _keychain_get( + self.keychain_service, "client_secret" + ) + if client_id and client_secret: + return client_id, client_secret + return None + + @property + def can_refresh(self) -> bool: + return bool(self.token_url) and self.client_credentials() is not None + + def _cached_token(self) -> tuple[str, str] | None: + """Return (token, where_it_came_from) for the first non-expired candidate.""" + candidates: list[tuple[str, str]] = [] + env_token = os.environ.get(self.token_env, "").strip() + if env_token: + candidates.append((env_token, f"${self.token_env}")) + if self.cache_path and self.cache_path.exists(): + try: + file_token = self.cache_path.read_text().strip() + except OSError: + file_token = "" + if file_token: + candidates.append((file_token, str(self.cache_path))) + + for token, origin in candidates: + exp = _jwt_expiry(token) + if exp is None: + return token, f"{origin} (opaque token, expiry unknown)" + if exp - time.time() > self.SKEW_SECONDS: + mins = (exp - time.time()) / 60 + return token, f"{origin} (valid {mins:.0f} more min)" + return None + + def mint(self) -> str: + """Exchange client credentials for a fresh access token.""" + if not self.token_url: + raise AuthError("no --token-url configured, cannot mint a token") + creds = self.client_credentials() + if not creds: + raise AuthError( + f"no client credentials found. Set ${self.id_env} and ${self.secret_env}, " + f"or add them to the macOS keychain under service '{self.keychain_service}'." + ) + client_id, client_secret = creds + basic = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() + status, payload, raw = _request( + self.token_url, + method="POST", + body=b"grant_type=client_credentials", + headers={ + "Authorization": f"Basic {basic}", + "Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + }, + ) + if status >= 400 or not isinstance(payload, dict): + raise AuthError(f"token endpoint returned HTTP {status}: {raw[:300]}") + token = payload.get("access_token") + if not isinstance(token, str) or not token: + raise AuthError(f"token endpoint response had no access_token: {raw[:300]}") + + self._token = token + self.refresh_count += 1 + self._write_cache(token) + exp = _jwt_expiry(token) + lifetime = payload.get("expires_in") + if exp is not None: + detail = f"expires in {(exp - time.time()) / 60:.0f} min" + elif isinstance(lifetime, int): + detail = f"expires in {lifetime / 60:.0f} min" + else: + detail = "expiry unknown" + print(f" minted a new access token ({detail})") + return token + + def _write_cache(self, token: str) -> None: + if not self.cache_path: + return + try: + # Create with 0600 from the start — never briefly world-readable. + fd = os.open(self.cache_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as handle: + handle.write(token + "\n") + os.chmod(self.cache_path, 0o600) + except OSError as exc: + print(f" warning: could not write token cache {self.cache_path}: {exc}") + + def token(self) -> str: + if self._token: + return self._token + cached = self._cached_token() + if cached: + self._token, self.last_source = cached[0], cached[1] + print(f" using token from {self.last_source}") + return self._token + if not self.can_refresh: + raise AuthError( + "no valid token available and cannot mint one.\n" + f" Provide a token via ${self.token_env} or {self.cache_path}, or supply\n" + f" client credentials (${self.id_env}/${self.secret_env} or macOS keychain\n" + f" service '{self.keychain_service}') together with --token-url." + ) + self.last_source = "client_credentials grant" + return self.mint() + + def headers(self) -> dict[str, str]: + """Auth header, or none at all. + + An unauthenticated server (local HAPI) needs no header, so a missing token + is not fatal here — we send the request bare and let the target answer. A + real 401 is then reported by preflight with an actionable message, rather + than this layer guessing that auth was required. + """ + try: + return {"Authorization": f"Bearer {self.token()}"} + except AuthError: + return {} + + def invalidate(self) -> bool: + """Drop the current token and mint a replacement. False if that is impossible.""" + self._token = None + if not self.can_refresh: + return False + try: + self.mint() + except AuthError as exc: + print(f" token refresh failed: {exc}") + return False + return True + + +def _request_target( + url: str, + auth: TokenSource | dict[str, str], + method: str = "GET", + body: bytes | None = None, + extra_headers: dict[str, str] | None = None, +) -> tuple[int, dict[str, Any] | None, str]: + """Request against the target, refreshing the token once on a 401. + + A long push can straddle the token's lifetime, so one silent retry after a + refresh keeps a multi-minute run from dying halfway through. + """ + headers = {"Accept": FHIR_JSON, **(extra_headers or {})} + if isinstance(auth, TokenSource): + headers.update(auth.headers()) + else: + headers.update(auth) + + status, payload, raw = _request(url, method=method, body=body, headers=headers) + if status == 401 and isinstance(auth, TokenSource) and auth.can_refresh: + print(" target returned 401 — refreshing token and retrying once") + if auth.invalidate(): + headers.update(auth.headers()) + status, payload, raw = _request(url, method=method, body=body, headers=headers) + return status, payload, raw + + +def _request( + url: str, + method: str = "GET", + body: bytes | None = None, + headers: dict[str, str] | None = None, + timeout: int = 180, +) -> tuple[int, dict[str, Any] | None, str]: + """Issue one HTTP request. Returns (status, parsed_json_or_None, raw_text).""" + req = urllib.request.Request(url, data=body, method=method) + for key, value in (headers or {}).items(): + req.add_header(key, value) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8", errors="replace") + status = resp.status + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", errors="replace") + status = exc.code + except urllib.error.URLError as exc: + raise TransferError(f"Cannot reach {url}: {exc.reason}") from exc + try: + return status, json.loads(raw), raw + except json.JSONDecodeError: + return status, None, raw + + +def fetch_all(source: str, resource_type: str, auth: dict[str, str]) -> list[dict[str, Any]]: + """Page through every resource of one type on the source server.""" + resources: list[dict[str, Any]] = [] + url = f"{source.rstrip('/')}/{resource_type}?_count={PAGE_SIZE}" + seen_pages = 0 + while url: + status, payload, raw = _request(url, headers={"Accept": FHIR_JSON, **auth}) + if status >= 400 or payload is None: + raise TransferError(f"GET {url} -> HTTP {status}: {raw[:400]}") + for entry in payload.get("entry", []) or []: + resource = entry.get("resource") + if isinstance(resource, dict) and resource.get("id"): + resources.append(resource) + seen_pages += 1 + url = next( + (link.get("url") for link in payload.get("link", []) or [] if link.get("relation") == "next"), + None, + ) + if seen_pages > 500: # runaway-paging guard + raise TransferError(f"{resource_type}: exceeded 500 pages, aborting") + return resources + + +def normalize(resource: dict[str, Any]) -> dict[str, Any]: + """Backfill Library.url so HAPI can resolve canonical Measure.library refs.""" + if resource.get("resourceType") == "Library" and not resource.get("url") and resource.get("id"): + return {**resource, "url": f"Library/{resource['id']}"} + return resource + + +def _outcome_text(entry: dict[str, Any]) -> str: + outcome = (entry.get("response") or {}).get("outcome") + if not isinstance(outcome, dict): + return "" + issues = outcome.get("issue", []) or [] + parts = [(issue.get("diagnostics") or (issue.get("details") or {}).get("text") or "").strip() for issue in issues] + return "; ".join(p for p in parts if p)[:300] + + +def push_chunk( + target: str, + chunk: list[dict[str, Any]], + auth: TokenSource | dict[str, str], +) -> tuple[list[str], list[tuple[str, str, str]]]: + """POST one batch Bundle. Returns (succeeded_refs, [(ref, status, detail)]).""" + bundle = { + "resourceType": "Bundle", + "type": "batch", + "entry": [ + { + "resource": normalize(r), + "request": {"method": "PUT", "url": f"{r['resourceType']}/{r['id']}"}, + } + for r in chunk + ], + } + refs = [f"{r['resourceType']}/{r['id']}" for r in chunk] + status, payload, raw = _request_target( + target.rstrip("/"), + auth, + method="POST", + body=json.dumps(bundle).encode("utf-8"), + extra_headers={"Content-Type": FHIR_JSON}, + ) + # Whole-bundle rejection: every entry in this chunk failed. + if status >= 400 or payload is None: + return [], [(ref, str(status), raw[:300]) for ref in refs] + + succeeded: list[str] = [] + failed: list[tuple[str, str, str]] = [] + entries = payload.get("entry", []) or [] + for i, ref in enumerate(refs): + entry = entries[i] if i < len(entries) else {} + entry_status = (entry.get("response") or {}).get("status", "") + if entry_status.startswith("2"): + succeeded.append(ref) + else: + failed.append((ref, entry_status or "no-response", _outcome_text(entry))) + return succeeded, failed + + +def _identity(resource: dict[str, Any]) -> tuple[str, str]: + """The (url, version) pair that decides whether two resources are the same content.""" + return (resource.get("url") or "", resource.get("version") or "") + + +def probe_existing( + target: str, + resource_type: str, + ids: list[str], + auth: TokenSource | dict[str, str], + chunk_size: int = 20, +) -> dict[str, dict[str, Any]]: + """Read-only: return {id: existing_resource} for the target IDs that already exist. + + Tries a chunked `_id=a,b,c` search first (few round trips), and falls back to + one GET per resource when the server rejects multi-value `_id`. Writes nothing + either way — GET only. + + Only 404/410 counts as "this resource is absent". Any other error (401, 403, + 5xx) raises, because a probe that cannot see the target must never be reported + as "nothing there" — that would turn an auth failure into a silent green light + to overwrite. + """ + found: dict[str, dict[str, Any]] = {} + base = target.rstrip("/") + elements = "id,url,version,name,title,status" + + for i in range(0, len(ids), chunk_size): + chunk = ids[i : i + chunk_size] + joined = ",".join(urllib.parse.quote(rid, safe="") for rid in chunk) + url = f"{base}/{resource_type}?_id={joined}&_elements={elements}&_count={len(chunk)}" + status, payload, raw = _request_target(url, auth) + + if status in (401, 403) or status >= 500: + raise TransferError(f"probe of {resource_type} failed: HTTP {status}: {raw[:300]}") + + if status >= 400 or payload is None or payload.get("resourceType") != "Bundle": + # Server does not honor multi-value _id — fall back to single reads. + for rid in chunk: + one_status, one_payload, one_raw = _request_target( + f"{base}/{resource_type}/{urllib.parse.quote(rid, safe='')}", auth + ) + if one_status == 200 and isinstance(one_payload, dict): + found[rid] = one_payload + elif one_status in (404, 410): + continue # genuinely absent + else: + raise TransferError(f"probe of {resource_type}/{rid} failed: HTTP {one_status}: {one_raw[:300]}") + continue + + for entry in payload.get("entry", []) or []: + resource = entry.get("resource") + if isinstance(resource, dict) and resource.get("id"): + found[resource["id"]] = resource + return found + + +def preflight_target(target: str, resource_type: str, auth: TokenSource | dict[str, str]) -> None: + """Confirm the target answers an authenticated search before we do real work. + + `/metadata` is often unsecured, so it proves nothing about the token. Search on + a real resource type is what the collision check and verify steps depend on. + """ + url = f"{target.rstrip('/')}/{resource_type}?_summary=count" + status, _, raw = _request_target(url, auth) + if status in (401, 403): + diagnostics = "" + try: + issues = json.loads(raw).get("issue", []) or [] + diagnostics = "; ".join(i.get("diagnostics", "") for i in issues if i.get("diagnostics")) + except (json.JSONDecodeError, AttributeError): + diagnostics = raw[:200] + hint = "" + if isinstance(auth, TokenSource) and not auth.can_refresh: + hint = ( + "\n No client credentials are configured, so a fresh token cannot be minted." + "\n Add them to the macOS keychain:" + f"\n security add-generic-password -s {auth.keychain_service} -a client_id -w" + f"\n security add-generic-password -s {auth.keychain_service} -a client_secret -w" + "\n (the -w with no value prompts, keeping the secret out of argv and history)" + ) + raise TransferError( + f"target rejected the credentials: HTTP {status}" + (f" — {diagnostics}" if diagnostics else "") + hint + ) + if status >= 400: + raise TransferError(f"target search failed: GET {url} -> HTTP {status}: {raw[:300]}") + + +def collision_check( + target: str, + inventory: dict[str, list[dict[str, Any]]], + types: list[str], + auth: TokenSource | dict[str, str], +) -> tuple[int, int]: + """Report which source resources already exist on the target. Read-only. + + Returns (identical_count, differing_count). "Identical" means the target's + resource carries the same canonical url and version as the source's — an + overwrite there is a no-op in content terms. "Differing" is where an + overwrite would actually replace something. + """ + print("collision check (read-only, nothing is written):") + identical_total = 0 + differing_total = 0 + differing_detail: list[str] = [] + + for resource_type in types: + resources = inventory[resource_type] + if not resources: + continue + by_id = {r["id"]: r for r in resources} + # A probe failure propagates: an unreadable target is not an empty target. + existing = probe_existing(target, resource_type, list(by_id), auth) + + identical = 0 + differing = 0 + for rid, target_resource in existing.items(): + source_resource = by_id.get(rid) + if source_resource is None: + continue + if _identity(source_resource) == _identity(target_resource): + identical += 1 + else: + differing += 1 + src_url, src_ver = _identity(source_resource) + tgt_url, tgt_ver = _identity(target_resource) + differing_detail.append( + f" {resource_type}/{rid}\n" + f" target: url={tgt_url or '(none)'} version={tgt_ver or '(none)'}\n" + f" source: url={src_url or '(none)'} version={src_ver or '(none)'}" + ) + + identical_total += identical + differing_total += differing + new_count = len(resources) - len(existing) + print( + f" {resource_type:<12} {new_count:>4} new, " + f"{identical:>4} already identical, {differing:>4} would be replaced" + ) + + if differing_detail: + print(f"\n{differing_total} resource(s) on the target would be REPLACED with different content:") + for line in differing_detail[:25]: + print(line) + if len(differing_detail) > 25: + print(f" … and {len(differing_detail) - 25} more") + + print() + return identical_total, differing_total + + +def search_count( + target: str, + resource_type: str, + auth: TokenSource | dict[str, str], + want: int = 0, + settle_seconds: int = 0, +) -> int | None: + """Search-based count, optionally polled until it reaches `want`. + + HAPI refreshes its Lucene index asynchronously unless + `synchronization.strategy=sync` is set, so a resource that was just written + successfully can be invisible to search for tens of seconds. Measured against + this target: 40s for a single CodeSystem. Polling avoids reporting a healthy + write as a failure. See the HAPI async-indexing section in CLAUDE.md. + """ + url = f"{target.rstrip('/')}/{resource_type}?_summary=count" + deadline = time.monotonic() + max(0, settle_seconds) + total: int | None = None + while True: + try: + status, payload, _ = _request_target(url, auth) + except TransferError: + return None + if status >= 400 or payload is None: + return None + candidate = payload.get("total") + total = candidate if isinstance(candidate, int) else None + if total is None or total >= want or time.monotonic() >= deadline: + return total + time.sleep(5) + + +def verify_written( + target: str, + resource_type: str, + ids: list[str], + auth: TokenSource | dict[str, str], +) -> tuple[int, list[str]]: + """Direct-read each id. Returns (ok_count, missing_ids). + + `GET {Type}/{id}` bypasses the search index entirely, so this is the + authoritative check that content actually landed — the triage rule from + CLAUDE.md: if the direct read shows the data and search does not, the index + is stale and the write was fine. + """ + base = target.rstrip("/") + ok = 0 + missing: list[str] = [] + for rid in ids: + status, _, _ = _request_target(f"{base}/{resource_type}/{urllib.parse.quote(rid, safe='')}", auth) + if status == 200: + ok += 1 + else: + missing.append(f"{rid} (HTTP {status})") + return ok, missing + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--target", required=True, help="Target FHIR base URL (e.g. https://example.org/fhir)") + parser.add_argument("--source", default=DEFAULT_SOURCE, help=f"Source FHIR base URL (default: {DEFAULT_SOURCE})") + parser.add_argument( + "--types", + default=",".join(DEFAULT_TYPES), + help="Comma-separated resource types, in write order", + ) + parser.add_argument("--chunk-size", type=int, default=25, help="Max entries per batch Bundle (default: 25)") + parser.add_argument( + "--limit", + type=int, + default=0, + help="Copy at most N resources of each type. For smoke-testing the write path " + "against a real target before committing to the full set. 0 = no limit.", + ) + parser.add_argument( + "--verify-sample", + type=int, + default=5, + help="Direct-read this many resources per type after writing (default: 5)", + ) + parser.add_argument( + "--verify-all", + action="store_true", + help="Direct-read every written resource instead of a sample. Slower, exhaustive.", + ) + parser.add_argument( + "--settle-seconds", + type=int, + default=60, + help="How long to let the target's search index catch up before reporting its " + "count (default: 60). Does not affect pass/fail — direct reads decide that.", + ) + parser.add_argument("--dry-run", action="store_true", help="Read from source and report, write nothing") + parser.add_argument( + "--check-only", + action="store_true", + help="Run the read-only collision check against the target and exit. Writes nothing.", + ) + parser.add_argument( + "--overwrite", + action="store_true", + help="Required to proceed when the collision check finds target resources whose " + "content differs from the source. Without it, the run aborts before writing.", + ) + parser.add_argument( + "--source-token-env", + default="SOURCE_FHIR_TOKEN", + help="Env var holding the source bearer token, if it needs one", + ) + parser.add_argument("--token-env", default="TARGET_FHIR_TOKEN", help="Env var holding the target bearer token") + parser.add_argument( + "--token-url", + default=None, + help="OAuth2 token endpoint for the client_credentials grant. Defaults to the " + "target's origin + /token. Use --no-token-url to disable auto-refresh.", + ) + parser.add_argument( + "--no-token-url", + action="store_true", + help="Never mint tokens; use only the supplied token and fail when it expires.", + ) + parser.add_argument( + "--token-file", + default=DEFAULT_TOKEN_FILE, + help=f"Cache file for the access token, written 0600 (default: {DEFAULT_TOKEN_FILE})", + ) + parser.add_argument( + "--keychain-service", + default=DEFAULT_KEYCHAIN_SERVICE, + help=f"macOS keychain service holding client_id/client_secret (default: {DEFAULT_KEYCHAIN_SERVICE})", + ) + parser.add_argument( + "--refresh-token", + action="store_true", + help="Mint a fresh access token, cache it, and exit. Touches no FHIR content.", + ) + args = parser.parse_args() + + types = [t.strip() for t in args.types.split(",") if t.strip()] + + token_url = None + if not args.no_token_url: + token_url = args.token_url or _default_token_url(args.target) + cache_path = pathlib.Path(args.token_file).expanduser() if args.token_file else None + + target_auth = TokenSource( + token_url=token_url, + cache_path=cache_path, + keychain_service=args.keychain_service, + token_env=args.token_env, + id_env="TARGET_CLIENT_ID", + secret_env="TARGET_CLIENT_SECRET", + ) + + source_token = os.environ.get(args.source_token_env, "").strip() + source_auth = {"Authorization": f"Bearer {source_token}"} if source_token else {} + + print(f"source: {args.source}") + print(f"target: {args.target}{' (DRY RUN — nothing will be written)' if args.dry_run else ''}") + print(f"token: endpoint {token_url or '(auto-refresh disabled)'}") + print( + f" credentials {'available' if target_auth.client_credentials() else 'NOT found'}" + f" (env TARGET_CLIENT_ID/SECRET or keychain '{args.keychain_service}')" + ) + + if args.refresh_token: + try: + target_auth.mint() + except TransferError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + print(f"Token cached at {cache_path}. Nothing else was done.") + return 0 + + # Resolve a token now so its state is visible up front. Not having one is not + # fatal — unauthenticated targets exist — so preflight makes the real call. + try: + target_auth.token() + except AuthError as exc: + print(f" no token yet: {exc}") + print() + + # Prove the credentials work before reading 185 resources or touching anything. + try: + preflight_target(args.target, types[0] if types else "Measure", target_auth) + except TransferError as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 2 + print(" target preflight: search OK\n") + + # Read everything up front so a source-side failure aborts before any write. + inventory: dict[str, list[dict[str, Any]]] = {} + for resource_type in types: + try: + found = fetch_all(args.source, resource_type, source_auth) + except TransferError as exc: + print(f"ERROR reading {resource_type}: {exc}", file=sys.stderr) + return 1 + if args.limit > 0 and len(found) > args.limit: + print(f" read {len(found):>4} {resource_type} -> limited to {args.limit}") + found = found[: args.limit] + else: + print(f" read {len(found):>4} {resource_type}") + inventory[resource_type] = found + total_read = sum(len(v) for v in inventory.values()) + print(f" {'-' * 24}\n read {total_read:>4} resources total\n") + + if args.dry_run and not args.check_only: + for resource_type in types: + ids = [r["id"] for r in inventory[resource_type]] + preview = ", ".join(ids[:5]) + (f", … (+{len(ids) - 5})" if len(ids) > 5 else "") + print(f"{resource_type}: {preview}") + print() + + # Always look before writing. This is GET-only against the target. + try: + _, differing = collision_check(args.target, inventory, types, target_auth) + except TransferError as exc: + print(f"ERROR during collision check: {exc}", file=sys.stderr) + print("Nothing was written. An unreadable target is not an empty target.", file=sys.stderr) + return 2 + + if args.check_only: + print("Check-only run complete. Nothing written.") + return 0 + if args.dry_run: + print("Dry run complete. Nothing written.") + return 0 + if differing and not args.overwrite: + print( + f"ABORTED: {differing} target resource(s) hold different content at these IDs.\n" + f"Nothing was written. Review the list above, then re-run with --overwrite to replace them.", + file=sys.stderr, + ) + return 3 + + all_failed: list[tuple[str, str, str]] = [] + total_ok = 0 + started = time.monotonic() + + for resource_type in types: + resources = inventory[resource_type] + if not resources: + continue + chunks = [resources[i : i + args.chunk_size] for i in range(0, len(resources), args.chunk_size)] + ok_count = 0 + for idx, chunk in enumerate(chunks, 1): + try: + ok, failed = push_chunk(args.target, chunk, target_auth) + except TransferError as exc: + print( + f"ERROR writing {resource_type} chunk {idx}/{len(chunks)}: {exc}", + file=sys.stderr, + ) + return 1 + ok_count += len(ok) + all_failed.extend(failed) + print( + f" {resource_type} chunk {idx}/{len(chunks)}: {len(ok)} ok, {len(failed)} failed", + flush=True, + ) + total_ok += ok_count + print(f" → {resource_type}: {ok_count}/{len(resources)} written\n") + + elapsed = time.monotonic() - started + print(f"{'=' * 46}") + print(f"wrote {total_ok}/{total_read} resources in {elapsed:.1f}s") + + if all_failed: + print(f"\n{len(all_failed)} FAILED:") + for ref, status, detail in all_failed[:40]: + print(f" {ref:<50} HTTP {status} {detail}") + if len(all_failed) > 40: + print(f" … and {len(all_failed) - 40} more") + + # Direct reads decide pass/fail. The search count is informational only: it + # lags the write by design and is not evidence of anything when it is low. + print("\nverifying on target (direct reads are authoritative):") + missing_total: list[str] = [] + for resource_type in types: + resources = inventory[resource_type] + if not resources: + continue + ids = [r["id"] for r in resources] + sample = ids if args.verify_all else ids[: min(len(ids), args.verify_sample)] + ok, missing = verify_written(args.target, resource_type, sample, target_auth) + missing_total.extend(f"{resource_type}/{m}" for m in missing) + + counted = search_count( + args.target, + resource_type, + target_auth, + want=len(resources), + settle_seconds=args.settle_seconds, + ) + if counted is None: + count_note = "search count unavailable" + elif counted >= len(resources): + count_note = f"search count {counted}" + else: + count_note = f"search count {counted} (index still catching up, not a failure)" + print(f" {resource_type:<12} direct reads {ok}/{len(sample)} OK, {count_note}") + + if missing_total: + print(f"\n{len(missing_total)} resource(s) could NOT be read back:") + for ref in missing_total[:20]: + print(f" {ref}") + + return 1 if (all_failed or missing_total) else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_copy_measure_content.py b/scripts/tests/test_copy_measure_content.py new file mode 100644 index 00000000..96d3af27 --- /dev/null +++ b/scripts/tests/test_copy_measure_content.py @@ -0,0 +1,146 @@ +"""Unit tests for scripts/copy_measure_content.py — pure logic, no network. + +Run: + python3 -m pytest scripts/tests/test_copy_measure_content.py -v + +NOTE: CI's unit job runs `cd backend && pytest tests/`, so it does not collect +this file. Run it manually when touching the script. + +The expiry cases here are regression guards. During development the collision +check reported "185 new, 0 collisions" against a server that was 401ing every +probe — an expired token made an unreadable target look like an empty one, which +is a green light to overwrite. `probe_existing` now treats only 404/410 as +absent, and `_cached_token` refuses to hand back an already-expired token. +""" + +from __future__ import annotations + +import base64 +import importlib.util +import json +import pathlib +import time + +import pytest + +_SCRIPT = pathlib.Path(__file__).resolve().parents[1] / "copy_measure_content.py" + + +def _load_module(): + spec = importlib.util.spec_from_file_location("copy_measure_content", _SCRIPT) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +cmc = _load_module() + + +def _jwt(exp: int) -> str: + """Build an unsigned JWT carrying only an `exp` claim.""" + header = base64.urlsafe_b64encode(b'{"alg":"none"}').decode().rstrip("=") + payload = base64.urlsafe_b64encode(json.dumps({"exp": exp}).encode()).decode().rstrip("=") + return f"{header}.{payload}.signature" + + +class TestJwtExpiry: + def test_reads_exp_claim(self): + exp = int(time.time()) + 3600 + assert cmc._jwt_expiry(_jwt(exp)) == exp + + def test_reads_past_exp(self): + exp = int(time.time()) - 3600 + assert cmc._jwt_expiry(_jwt(exp)) == exp + + @pytest.mark.parametrize("token", ["not-a-jwt", "a.b.c", "", "one.two"]) + def test_non_jwt_returns_none(self, token): + """Opaque tokens have no readable expiry — None, not a crash.""" + assert cmc._jwt_expiry(token) is None + + +class TestCachedToken: + """Token selection must never hand back something the server will reject.""" + + def _source(self, monkeypatch, token: str) -> cmc.TokenSource: + monkeypatch.setenv("TEST_TOKEN_ENV", token) + return cmc.TokenSource( + token_url=None, + cache_path=None, + keychain_service="unused", + token_env="TEST_TOKEN_ENV", + id_env="TEST_ID_ENV", + secret_env="TEST_SECRET_ENV", + ) + + def test_expired_token_is_rejected(self, monkeypatch): + src = self._source(monkeypatch, _jwt(int(time.time()) - 60)) + assert src._cached_token() is None + + def test_live_token_is_accepted(self, monkeypatch): + src = self._source(monkeypatch, _jwt(int(time.time()) + 7200)) + assert src._cached_token() is not None + + def test_token_inside_skew_window_is_rejected(self, monkeypatch): + """Refuse a token about to expire — a long push would die mid-run.""" + src = self._source(monkeypatch, _jwt(int(time.time()) + 30)) + assert src._cached_token() is None + + def test_opaque_token_is_accepted(self, monkeypatch): + """Expiry unknowable — send it and let the server decide.""" + src = self._source(monkeypatch, "opaque-token-value") + assert src._cached_token() is not None + + def test_cannot_refresh_without_credentials(self, monkeypatch): + monkeypatch.delenv("TEST_ID_ENV", raising=False) + monkeypatch.delenv("TEST_SECRET_ENV", raising=False) + monkeypatch.setattr(cmc, "_keychain_get", lambda service, account: None) + src = self._source(monkeypatch, "tok") + src.token_url = "https://example.org/token" + assert src.can_refresh is False + + def test_can_refresh_with_env_credentials(self, monkeypatch): + monkeypatch.setenv("TEST_ID_ENV", "client-abc") + monkeypatch.setenv("TEST_SECRET_ENV", "secret-xyz") + src = self._source(monkeypatch, "tok") + src.token_url = "https://example.org/token" + assert src.can_refresh is True + assert src.client_credentials() == ("client-abc", "secret-xyz") + + +class TestIdentity: + """(url, version) decides whether an overwrite would replace real content.""" + + def test_same_url_and_version_match(self): + a = {"url": "http://x/vs", "version": "1.0"} + b = {"url": "http://x/vs", "version": "1.0"} + assert cmc._identity(a) == cmc._identity(b) + + def test_differing_version_does_not_match(self): + a = {"url": "http://x/vs", "version": "1.0"} + b = {"url": "http://x/vs", "version": "2.0"} + assert cmc._identity(a) != cmc._identity(b) + + def test_missing_fields_normalize_to_empty(self): + assert cmc._identity({}) == ("", "") + + +class TestNormalize: + def test_backfills_library_url(self): + out = cmc.normalize({"resourceType": "Library", "id": "L1"}) + assert out["url"] == "Library/L1" + + def test_leaves_existing_library_url(self): + out = cmc.normalize({"resourceType": "Library", "id": "L1", "url": "http://real/url"}) + assert out["url"] == "http://real/url" + + def test_ignores_other_resource_types(self): + out = cmc.normalize({"resourceType": "ValueSet", "id": "V1"}) + assert "url" not in out + + +class TestDefaultTokenUrl: + def test_derives_origin_plus_token(self): + assert cmc._default_token_url("https://example.org/fhir/") == "https://example.org/token" + + def test_ignores_deep_paths(self): + assert cmc._default_token_url("https://example.org/a/b/fhir") == "https://example.org/token"