test(integration): align stream and A2A contracts and isolate batch authorization

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/integration/caching/test_response_cache_empty_output.py

@@ -426,8 +426,16 @@ def test_streamed_chat_without_choices_is_never_assembled_or_cached(gateway: Gat
     with gateway.scenario() as scenario:
         handle: Final = scripted(scenario, CHAT_STREAM_WITHOUT_CHOICES)
         model: Final = scenario.model(api_base=handle.api_base())
-        empty: Final = _never_replayed(partial(_streamed_chat, gateway, model, text), upstream, handle.scenario_id)
-        assert all(answer.text == "" for answer in empty), empty
+        request: Final = chat_body(model, text, stream=True)
+        for attempt in range(1, 4):
+            refused: Final = gateway.request("POST", "/v1/chat/completions", request)
+            assert refused.status_code == 500, refused.text
+            assert "Stream ended before a finish_reason was received" in refused.text, refused.text
+            assert upstream.calls(handle.scenario_id) == attempt, "the unfinished stream was served from cache"
+        rescript(handle, CHAT_STREAM)
+        answer: Final = _streamed_chat(gateway, model, text)
+        assert answer[1] == "streamed response", answer
+        assert upstream.calls(handle.scenario_id) == 4
 
 
 def _streamed_text(gateway: Gateway, model: str, text: str) -> _Answer:

tests/integration/compatibility/test_a2a_wire_versions.py

@@ -965,9 +965,18 @@ def test_a2a_task_methods_pass_through_verbatim_and_push_callbacks_are_validated
                 {"taskId": "t1", "url": callback},
             ):
                 rejected = call("tasks/pushNotificationConfig/set", params)
-                assert (rejected.status_code, rejected.json()) == (400, {"detail": detail}), (
-                    f"{params}: {rejected.text}"
-                )
+                assert (rejected.status_code, rejected.json()) == (
+                    400,
+                    {
+                        "jsonrpc": "2.0",
+                        "id": marker + "tasks/pushNotificationConfig/set",
+                        "error": {
+                            "code": -32000,
+                            "message": detail,
+                            "data": {"type": "invalid_request_error", "param": None, "code": "400"},
+                        },
+                    },
+                ), f"{params}: {rejected.text}"
         assert tuple(item for item in wire.drain() if item.method == "POST") == ()
 
         public: Final = {"taskId": "t1", "pushNotificationConfig": {"url": "https://1.1.1.1/callback"}}
@@ -1060,7 +1069,15 @@ def test_a2a_v1_task_methods_and_extended_card_lower_to_0_3_upstream_and_answer_
             rejected = call("CreateTaskPushNotificationConfig", params)
             assert (rejected.status_code, rejected.json()) == (
                 400,
-                {"detail": "Push notification URL must use HTTPS"},
+                {
+                    "jsonrpc": "2.0",
+                    "id": marker + "CreateTaskPushNotificationConfig",
+                    "error": {
+                        "code": -32000,
+                        "message": "Push notification URL must use HTTPS",
+                        "data": {"type": "invalid_request_error", "param": None, "code": "400"},
+                    },
+                },
             ), f"{params}: {rejected.text}"
         assert forwarded() == ()
 
@@ -1136,6 +1153,14 @@ def test_a2a_v1_push_config_envelope_sent_to_a_0_3_agent_has_its_callback_valida
         )
         assert (rejected.status_code, rejected.json()) == (
             400,
-            {"detail": "Push notification URL must use HTTPS"},
+            {
+                "jsonrpc": "2.0",
+                "id": marker,
+                "error": {
+                    "code": -32000,
+                    "message": "Push notification URL must use HTTPS",
+                    "data": {"type": "invalid_request_error", "param": None, "code": "400"},
+                },
+            },
         ), rejected.text
         assert tuple(item for item in wire.drain() if item.method == "POST") == ()

tests/integration/sdk/test_anthropic_json_tool_cut_stream_sync_sdk.py

@@ -10,6 +10,7 @@ from integration._support.wire import Reply, Request, wire_server
 from pydantic import JsonValue
 
 import litellm
+from litellm.exceptions import MidStreamFallbackError
 from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
 
 _JSON_TOOL: Final = "json_tool_call"
@@ -141,6 +142,13 @@ def test_sync_stream_cut_before_the_json_tool_block_closes_still_yields_the_whol
             **_credentials(model),
         )
         assert isinstance(response, CustomStreamWrapper), type(response)
-        text: Final = "".join(str(chunk.choices[0].delta.content or "") for chunk in tuple(response))
+        fragments: Final[list[str]] = []
+        if cut:
+            with pytest.raises(MidStreamFallbackError, match="Stream ended before a finish_reason was received"):
+                for chunk in response:
+                    fragments.append(str(chunk.choices[0].delta.content or ""))
+        else:
+            fragments.extend(str(chunk.choices[0].delta.content or "") for chunk in response)
+        text: Final = "".join(fragments)
         assert len(wire.drain()) == 1
     assert json.loads(text) == _payload(marker_hex), text

tests/integration/security/test_files_batches_access.py

@@ -25,7 +25,7 @@ from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from integration._support.client import Gateway, Scenario, eventually, object_value, string_value
 from integration._support.database import read_rows
-from integration._support.process import owned_proxy
+from integration._support.process import OwnedState, owned_proxy
 from integration._support.tls import server_context, write_self_signed_cert
 from integration._support.wire import Reply, Request, Wire, wire_server
 from openai import OpenAI, PermissionDeniedError
@@ -250,15 +250,20 @@ def test_stranger_cannot_read_download_or_delete_another_teams_managed_file(gate
         assert _drained_other_than_model_list_probes(wire) == ()
 
 
-def test_stranger_cannot_retrieve_or_cancel_another_teams_managed_batch(gateway: Gateway) -> None:
+@pytest.mark.timeout(180)
+def test_stranger_cannot_retrieve_or_cancel_another_teams_managed_batch(gateway: Gateway, tmp_path: Path) -> None:
     key: Final = "provider-key-" + uuid.uuid4().hex[:8]
-    with wire_server(_openai_backend()) as wire, gateway.scenario() as scenario:
+    with (
+        owned_proxy(gateway, tmp_path, {"PROXY_BATCH_POLLING_ENABLED": "false"}, state=OwnedState.PRIVATE) as candidate,
+        wire_server(_openai_backend()) as wire,
+        candidate.scenario() as scenario,
+    ):
         model: Final = scenario.model(model="openai/gpt-4o-mini", api_base=wire.url + "/v1", api_key=key)
-        _wait_until_every_worker_serves(gateway, model)
+        _wait_until_every_worker_serves(candidate, model)
         owner: Final = _member(scenario, model)
         stranger: Final = _member(scenario, model)
 
-        uploaded: Final = gateway.request_multipart(
+        uploaded: Final = candidate.request_multipart(
             "/v1/files",
             {"purpose": "batch", "target_model_names": model},
             {"file": (INPUT_FILENAME, _jsonl(model), "application/jsonl")},
@@ -266,7 +271,7 @@ def test_stranger_cannot_retrieve_or_cancel_another_teams_managed_batch(gateway:
         )
         assert uploaded.status_code == 200, uploaded.text
         managed_input: Final = string_value(_json(uploaded)["id"])
-        created: Final = gateway.request(
+        created: Final = candidate.request(
             "POST",
             "/v1/batches",
             {"input_file_id": managed_input, "endpoint": "/v1/chat/completions", "completion_window": "24h"},
@@ -277,13 +282,13 @@ def test_stranger_cannot_retrieve_or_cancel_another_teams_managed_batch(gateway:
         wire.drain()
 
         expected_message: Final = f"User {stranger.user} does not have access to the object {unified_batch}"
-        stranger_get: Final = gateway.request("GET", f"/v1/batches/{unified_batch}", key=stranger.key)
+        stranger_get: Final = candidate.request("GET", f"/v1/batches/{unified_batch}", key=stranger.key)
         assert _error_message(stranger_get, 403) == expected_message, stranger_get.text
-        stranger_cancel: Final = gateway.request("POST", f"/v1/batches/{unified_batch}/cancel", {}, key=stranger.key)
+        stranger_cancel: Final = candidate.request("POST", f"/v1/batches/{unified_batch}/cancel", {}, key=stranger.key)
         assert _error_message(stranger_cancel, 403) == expected_message, stranger_cancel.text
         assert _drained_other_than_model_list_probes(wire) == (), "stranger batch calls must not reach the provider"
 
-        owner_cancel: Final = gateway.request("POST", f"/v1/batches/{unified_batch}/cancel", {}, key=owner.key)
+        owner_cancel: Final = candidate.request("POST", f"/v1/batches/{unified_batch}/cancel", {}, key=owner.key)
         assert owner_cancel.status_code == 200, owner_cancel.text
         assert string_value(_json(owner_cancel)["id"]) == unified_batch, owner_cancel.text
         (cancel_request,) = _drained_other_than_model_list_probes(wire)