Function _upsert_managed_inference

e2e/python/test_inference_routing.py:57–105

(
    inference_client: InferenceRouteClient,
    sandbox_client: SandboxClient,
    *,
    provider_name: str,
    provider_type: str,
    credential_key: str,
    base_url_key: str,
    model_id: str,
    base_url: str,
)

def _upsert_managed_inference(
    inference_client: InferenceRouteClient,
    sandbox_client: SandboxClient,
    *,
    provider_name: str,
    provider_type: str,
    credential_key: str,
    base_url_key: str,
    model_id: str,
    base_url: str,
) -> None:
    provider = datamodel_pb2.Provider(
        metadata=datamodel_pb2.ObjectMeta(name=provider_name),
        type=provider_type,
        credentials={credential_key: "mock"},
        config={
            base_url_key: base_url,
        },
    )
    timeout = sandbox_client._timeout

    for _ in range(5):
        try:
            sandbox_client._stub.UpdateProvider(
                openshell_pb2.UpdateProviderRequest(provider=provider),
                timeout=timeout,
            )
            break
        except grpc.RpcError as exc:
            if exc.code() != grpc.StatusCode.NOT_FOUND:
                raise

            try:
                sandbox_client._stub.CreateProvider(
                    openshell_pb2.CreateProviderRequest(provider=provider),
                    timeout=timeout,
                )
                break
            except grpc.RpcError as create_exc:
                if create_exc.code() == grpc.StatusCode.ALREADY_EXISTS:
                    continue
                raise
    else:
        raise RuntimeError("failed to upsert managed e2e provider after retries")

    inference_client.set_cluster(
        provider_name=provider_name,
        model_id=model_id,
    )


def _current_cluster_inference(