#!/usr/bin/env python3 """Original synthetic FHIR R4 reference-location lab. Python 3.11+, stdlib. Read-only, bounded collection Bundles and Patient subjects. No network, real health records, authorization, identity verification or complete FHIR validator. Never treats absence from this fixture as absence from an authoritative server. """ import argparse from copy import deepcopy import json import platform import re from urllib.parse import urlsplit A = "https://synthetic-a.example/fhir/" B = "https://synthetic-b.example/fhir/" OWNER = A + "Observation/obs-1" URN_OWNER = "urn:uuid:00000000-0000-4000-8000-000000000001" URN_PATIENT = "urn:uuid:11111111-1111-4111-8111-111111111111" ID = r"[A-Za-z0-9.-]{1,64}" CASES = ("relative-two-bases", "urn", "version-specific", "versionless-versions", "contained", "logical-only", "urn-owner-relative", "missing", "duplicate-identity") def patient(label, version="v1", key="42"): return {"resourceType": "Patient", "id": key, "meta": {"versionId": version}, "identifier": [{"system": "urn:lockedin:synthetic-patient", "value": label}]} def scenario(name): owner_url = OWNER owner = {"resourceType": "Observation", "id": "obs-1", "status": "final", "code": {"text": "Synthetic reference exercise only"}, "subject": {"reference": "Patient/42"}} targets = [{"fullUrl": B + "Patient/42", "resource": patient("SYNTHETIC-B")}, {"fullUrl": A + "Patient/42", "resource": patient("SYNTHETIC-A")}] if name == "urn": owner["subject"] = {"reference": URN_PATIENT} targets = [{"fullUrl": URN_PATIENT, "resource": patient("SYNTHETIC-URN")}] elif name in ("version-specific", "versionless-versions", "duplicate-identity"): owner["subject"] = {"reference": "Patient/42/_history/v2" if name != "versionless-versions" else "Patient/42"} targets = [{"fullUrl": A + "Patient/42", "resource": patient("SYNTHETIC-A-V1", "v1")}, {"fullUrl": A + "Patient/42", "resource": patient("SYNTHETIC-A-V2", "v2")}] if name == "duplicate-identity": targets.append({"fullUrl": A + "Patient/42", "resource": patient("SYNTHETIC-CONFLICT", "v2")}) elif name == "contained": owner["subject"] = {"reference": "#p1"} contained = patient("SYNTHETIC-CONTAINED", key="p1") contained.pop("meta") owner["contained"] = [contained] targets = [{"fullUrl": A + "Patient/p1", "resource": patient("SYNTHETIC-TOP-LEVEL", key="p1")}] elif name == "logical-only": owner["subject"] = {"identifier": {"system": "urn:lockedin:synthetic-chart", "value": "42"}} elif name == "urn-owner-relative": owner_url = URN_OWNER elif name == "missing": owner["subject"] = {"reference": "Patient/99"} return {"resourceType": "Bundle", "type": "collection", "entry": [{"fullUrl": owner_url, "resource": owner}, *targets]}, owner_url def resolve(bundle, owner_url): trace = [] def finish(reason, target=None, locator=None, entry_index=None): return {"state": "located_in_fixture" if target is not None else "unresolved", "reason": reason, "locator": locator, "entry_index": entry_index, "target": deepcopy(target), "trace": trace} if not isinstance(bundle, dict) or bundle.get("resourceType") != "Bundle" or bundle.get("type") != "collection": return finish("unsupported_fixture_bundle") entries = bundle.get("entry") if not isinstance(entries, list) or len(entries) > 16 or any(not isinstance(e, dict) or not isinstance(e.get("resource"), dict) or not isinstance(e["resource"].get("meta", {}), dict) or not isinstance(e.get("fullUrl"), str) or len(e["fullUrl"]) > 512 for e in entries): return finish("invalid_or_unbounded_entries") owners = [e for e in entries if e["fullUrl"] == owner_url] if len(owners) != 1 or owners[0]["resource"].get("resourceType") != "Observation": return finish("owner_identity_unresolved") owner = owners[0]["resource"] ref = owner.get("subject") if not isinstance(ref, dict) or ref.get("type", "Patient") != "Patient": return finish("unsupported_patient_reference") raw = ref.get("reference") if raw is None: return finish("identifier_lookup_required" if "identifier" in ref else "literal_reference_required") if not isinstance(raw, str) or len(raw) > 512 or not raw: return finish("invalid_literal_reference") trace.append({"step": "literal", "value": raw}) if re.fullmatch("#" + ID, raw): contained = owner.get("contained", []) if not isinstance(contained, list) or len(contained) > 16 or any(not isinstance(r, dict) for r in contained): return finish("invalid_contained_fixture") candidates = [r for r in contained if r.get("id") == raw[1:]] if len(candidates) != 1: return finish("contained_identity_unresolved") if candidates[0].get("resourceType") != "Patient": return finish("target_type_mismatch") return finish("contained_match", candidates[0], "owner.contained/" + raw[1:]) version = None history = re.fullmatch(r"(.+)/_history/(" + ID + ")", raw) if history: raw, version = history.groups() elif "/_history/" in raw: return finish("unsupported_history_reference") if re.fullmatch("Patient/" + ID, raw): try: context = urlsplit(owner_url) except ValueError: return finish("relative_base_unavailable") suffix = "/Observation/" + str(owner.get("id", "")) if context.scheme not in ("http", "https") or not context.netloc or context.query or context.fragment or not context.path.endswith(suffix): return finish("relative_base_unavailable") raw = owner_url[:-len(suffix)] + "/" + raw elif raw.startswith(("http://", "https://")): try: url = urlsplit(raw) except ValueError: return finish("unsupported_absolute_reference") if not url.netloc or url.query or url.fragment or url.username or url.password or not re.search(r"/Patient/" + ID + "$", url.path): return finish("unsupported_absolute_reference") elif not re.fullmatch(r"urn:uuid:[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", raw) or version: return finish("unsupported_literal_reference") trace.append({"step": "bundle_locator", "fullUrl": raw, "requested_version": version}) candidates = [(i, e) for i, e in enumerate(entries) if e["fullUrl"] == raw] if not candidates: return finish("not_in_bundle", locator=raw) if version is not None: candidates = [(i, e) for i, e in candidates if e["resource"].get("meta", {}).get("versionId") == version] if not candidates: return finish("version_not_in_bundle", locator=raw) if len(candidates) != 1: return finish("ambiguous_bundle_identity", locator=raw) index, entry = candidates[0] if entry["resource"].get("resourceType") != "Patient": return finish("target_type_mismatch", locator=raw) if raw.startswith(("http://", "https://")) and raw.rsplit("/", 1)[-1] != entry["resource"].get("id"): return finish("target_identity_mismatch", locator=raw) return finish("fullUrl_and_version_match" if version else "fullUrl_match", entry["resource"], raw, index) def naive(bundle, owner_url): # Deliberately broken: discards base, version, contained scope and identifier meaning. owner = next(e["resource"] for e in bundle["entry"] if e["fullUrl"] == owner_url) ref = owner["subject"] raw = ref.get("reference", ref.get("identifier", {}).get("value", "")) key = raw.split("/_history/")[0].rsplit("/", 1)[-1].lstrip("#") return next((deepcopy(e["resource"]) for e in bundle["entry"] if e["resource"].get("resourceType") == "Patient" and e["resource"].get("id") == key), None) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--case", choices=CASES) parser.add_argument("--reverse-entries", action="store_true") args = parser.parse_args() reports = [] for name in CASES: if args.case and args.case != name: continue bundle, owner = scenario(name) if args.reverse_entries: bundle["entry"].reverse() reports.append({"case": name, "owner_fullUrl": owner, "bundle": bundle, "guarded": resolve(bundle, owner), "broken_id_only_target": naive(bundle, owner)}) print(json.dumps({"python_version": platform.python_version(), "fhir_contract": "R4 collection Bundle subset; Patient subject only", "reverse_entries": args.reverse_entries, "cases": reports}, indent=2)) if __name__ == "__main__": main()