Handle api_key alias in flags token assertions

src/posthog_test_harness/actions.py

@@ -784,13 +784,19 @@ async def execute(self, params: Dict[str, Any], ctx: "TestContext") -> Any:
         # Support dot notation for nested fields (e.g., "person_properties.$device_id")
         parts = field.split(".")
         current = body
-        for part in parts:
+        for index, part in enumerate(parts):
             if not isinstance(current, dict):
                 raise AssertionError(
                     f"Cannot traverse into non-dict at '{part}' in field path '{field}'. "
                     f"Value is: {current!r}"
                 )
             if part not in current:
+                # The /flags endpoint accepts api_key as an alias for token.
+                # Keep the contract's default assertion on token while allowing
+                # SDKs that send api_key to satisfy the same check.
+                if index == 0 and part == "token" and "api_key" in current:
+                    current = current["api_key"]
+                    continue
                 raise AssertionError(
                     f"Field '{part}' not found in /flags request body at path '{field}'. "
                     f"Available keys: {list(current.keys())}"

tests/test_feature_flag_support.py

@@ -8,6 +8,7 @@
     AssertActionResultAction,
     AssertEventCountWithNameAction,
     AssertEventPropertyInNamedEventAction,
+    AssertFlagsRequestFieldAction,
     AssertFlagsRequestQueryParamAction,
     GetFeatureFlagAction,
 )
@@ -179,10 +180,31 @@ def test_feature_flag_contract_asserts_top_level_distinct_id_on_flags_request()
     )
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("auth_field", ["token", "api_key"])
+async def test_assert_flags_request_field_accepts_token_aliases(auth_field: str) -> None:
+    ctx = _ctx(
+        _FakeMockServer(
+            [
+                _make_recorded(
+                    "/flags/",
+                    body_decompressed=json.dumps({auth_field: "phc_test_key"}),
+                )
+            ]
+        )
+    )
+
+    await AssertFlagsRequestFieldAction().execute(
+        {"field": "token", "expected": "phc_test_key"},
+        ctx,
+    )
+
+
 def _make_recorded(
     path: str,
     query_params: dict | None = None,
     parsed_events: list | None = None,
+    body_decompressed: str | None = None,
 ) -> RecordedRequest:
     return RecordedRequest(
         timestamp_ms=0,
@@ -191,7 +213,7 @@ def _make_recorded(
         headers={},
         query_params=query_params or {},
         body_raw=b"",
-        body_decompressed=None,
+        body_decompressed=body_decompressed,
         parsed_events=parsed_events,
         response_status=200,
         response_headers={},