Propagate proxies and ssl_verification args to cloud services

qiskit_ibm_runtime/accounts/account.py

@@ -282,6 +282,23 @@ def get_auth_handler(self) -> AuthBase:
             verify=self.verify,
         )
 
+    def _get_proxies_kwargs(self) -> dict:
+        proxies_kwargs = {}
+        if self.proxies is not None:
+            proxies_kwargs = self.proxies.to_request_params()
+        return proxies_kwargs
+
+    def get_iam_authentificator(self) -> IAMAuthenticator:
+        """Return the configured IAM Authentification service"""
+        iam_url = get_iam_api_url(self.url)
+        proxies_kwargs = self._get_proxies_kwargs()
+        return IAMAuthenticator(
+            apikey=self.token,
+            url=iam_url,
+            disable_ssl_verification=not self.verify,
+            **proxies_kwargs,
+        )
+
     def resolve_crn(self) -> None:
         """Resolves the corresponding unique Cloud Resource Name (CRN) for the given non-unique service
         instance name and updates the ``instance`` attribute accordingly.
@@ -314,14 +331,14 @@ def resolve_crn(self) -> None:
 
     def list_instances(self) -> list[dict[str, Any]]:
         """Retrieve all crns with the IBM Cloud Global Search API."""
-        iam_url = get_iam_api_url(self.url)
-        authenticator = IAMAuthenticator(self.token, url=iam_url)
+        authenticator = self.get_iam_authentificator()
         client = GlobalSearchV2(authenticator=authenticator)
         catalog = GlobalCatalogV1(authenticator=authenticator)
         client.set_service_url(get_global_search_api_url(self.url))
         catalog.set_service_url(get_global_catalog_api_url(self.url))
         search_cursor = None
         all_crns = []
+        proxies_kwargs = self._get_proxies_kwargs()
         while True:
             try:
                 result = client.search(
@@ -335,6 +352,8 @@ def list_instances(self) -> list[dict[str, Any]]:
                     ],
                     search_cursor=search_cursor,
                     limit=100,
+                    verify=self.verify,
+                    **proxies_kwargs,
                 ).get_result()
             except:  # noqa: E722 bare-except
                 raise InvalidAccountError(
@@ -349,7 +368,9 @@ def list_instances(self) -> list[dict[str, Any]]:
                 if allocations:
                     try:
                         catalog_result = catalog.get_catalog_entry(
-                            id=item.get("service_plan_unique_id")
+                            id=item.get("service_plan_unique_id"),
+                            verify=self.verify,
+                            **proxies_kwargs,
                         ).get_result()
                         plan_name = (
                             catalog_result.get("overview_ui", {})

release-notes/unreleased/2592.bug.rst

@@ -0,0 +1,4 @@
+The ``proxies`` and ``ssl_verification`` arguments for ``QiskitRuntimeService``
+are nowpropagated to the underlying HTTP requests to ```GlobalSearchV2``,
+``GlobalCatalogV1`` and ``IAMAuthenticator`` services, allowing to instantiate
+``QiskitRuntimeService`` correctly when using a proxy.

test/decorators.py

@@ -19,6 +19,7 @@
 from unittest import SkipTest
 
 from qiskit_ibm_runtime import QiskitRuntimeService
+from qiskit_ibm_runtime.accounts import ChannelType
 
 from .unit.mock.fake_runtime_service import FakeRuntimeService
 
@@ -143,7 +144,7 @@ class IntegrationTestDependencies:
     instance: str | None
     qpu: str
     token: str
-    channel: str
+    channel: ChannelType
     url: str
 
 

test/integration/test_proxies.py

@@ -14,10 +14,13 @@
 
 import subprocess
 import urllib
+from time import sleep
+import socket
 
 
-from qiskit_ibm_runtime.proxies import ProxyConfiguration
-from qiskit_ibm_runtime.api.clients.runtime import RuntimeClient
+from qiskit_ibm_runtime import QiskitRuntimeService
+from qiskit_ibm_runtime.accounts.exceptions import InvalidAccountError
+
 
 from ..ibm_test_case import IBMTestCase
 from ..decorators import IntegrationTestDependencies, integration_test_setup
@@ -38,6 +41,17 @@ def setUp(self):
         # launch a mock server.
         command = ["pproxy", "-v", "-l", "http://{}:{}".format(ADDRESS, PORT)]
         self.proxy_process = subprocess.Popen(command, stdout=subprocess.PIPE)
+        """Time for the proxy to start"""
+        sleep(2)
+        """Block all traffic not routed to the proxy"""
+        self._original_connect = socket.socket.connect
+
+        def blocking_connect(sock, address):
+            if address != (ADDRESS, PORT):
+                raise RuntimeError(f"Blocked network access to {address}")
+            return self._original_connect(sock, address)
+
+        socket.socket.connect = blocking_connect
 
     def tearDown(self):
         """Test cleanup."""
@@ -50,19 +64,39 @@ def tearDown(self):
 
             # wait for the process to terminate
             self.proxy_process.wait()
+        socket.socket.connect = self._original_connect
 
-    @integration_test_setup(supported_channel=["ibm_cloud"])
-    def test_proxies_cloud_runtime_client(self, dependencies: IntegrationTestDependencies) -> None:
-        """Should reach the proxy using RuntimeClient."""
-        params = dependencies.service._client_params
-        params.proxies = ProxyConfiguration(urls=VALID_PROXIES)
-        client = RuntimeClient(params)
-        client.jobs_get(limit=1)
-        api_line = pproxy_desired_access_log_line(params.url)
+    @integration_test_setup(supported_channel=["ibm_quantum_platform"], init_service=False)
+    def test_proxies_qiskit_runtime_service(
+        self, dependencies: IntegrationTestDependencies
+    ) -> None:
+        """Should reach the proxy using QiskitRuntimeService."""
+        # Allow some time for `pproxy` to receive requests.
+        service = QiskitRuntimeService(
+            instance=dependencies.instance,
+            token=dependencies.token,
+            channel=dependencies.channel,
+            verify=False,
+            proxies={"urls": VALID_PROXIES},
+        )
+        service.jobs(limit=1)
+
+        api_line = pproxy_desired_access_log_line(dependencies.url)
         self.proxy_process.terminate()  # kill to be able of reading the output
         proxy_output = self.proxy_process.stdout.read().decode("utf-8")
         self.assertIn(api_line, proxy_output)
 
+    @integration_test_setup(supported_channel=["ibm_quantum_platform"], init_service=False)
+    def test_no_proxy_raises_exception(self, dependencies: IntegrationTestDependencies) -> None:
+        """Should raise an exception when no proxy is specified."""
+        with self.assertRaises(InvalidAccountError):
+            service = QiskitRuntimeService(
+                instance=dependencies.instance,
+                token=dependencies.token,
+                channel=dependencies.channel,
+            )
+            service.jobs(limit=1)
+
 
 def pproxy_desired_access_log_line(url):
     """Return a desired pproxy log entry given a url."""