Improve test coverage and documentation for proxy configuration

qiskit_ibm_runtime/accounts/account.py

@@ -16,9 +16,10 @@
 from typing import Optional
 from urllib.parse import urlparse
 
-from typing_extensions import Literal
 from requests.auth import AuthBase
-
+from typing_extensions import Literal
+from ..proxies import ProxyConfiguration
+from ..utils.hgp import from_instance_format
 from ..api.auth import LegacyAuth, CloudAuth
 
 AccountType = Optional[Literal["cloud", "legacy"]]
@@ -27,40 +28,6 @@
 CLOUD_API_URL = "https://us-east.quantum-computing.cloud.ibm.com"
 
 
-def _assert_valid_auth(auth: AccountType) -> None:
-    """Assert that the auth parameter is valid."""
-    if not (auth in ["cloud", "legacy"]):
-        raise ValueError(
-            f"Inappropriate `auth` value. Expected one of ['cloud', 'legacy'], got '{auth}'."
-        )
-
-
-def _assert_valid_token(token: str) -> None:
-    """Assert that the token is valid."""
-    if not (isinstance(token, str) and len(token) > 0):
-        raise ValueError(
-            f"Inappropriate `token` value. Expected a non-empty string, got '{token}'."
-        )
-
-
-def _assert_valid_url(url: str) -> None:
-    """Assert that the URL is valid."""
-    try:
-        urlparse(url)
-    except:
-        raise ValueError(f"Inappropriate `url` value. Failed to parse '{url}' as URL.")
-
-
-def _assert_valid_instance(auth: AccountType, instance: str) -> None:
-    """Assert that the instance name is valid for the given account type."""
-    if auth == "cloud":
-        if not (isinstance(instance, str) and len(instance) > 0):
-            raise ValueError(
-                f"Inappropriate `instance` value. Expected a non-empty string."
-            )
-    # TODO: add validation for legacy instance when adding test coverage
-
-
 class Account:
     """Class that represents an account."""
 
@@ -70,8 +37,7 @@ def __init__(
         token: str,
         url: Optional[str] = None,
         instance: Optional[str] = None,
-        # TODO: add validation for proxies input format
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = True,
     ):
         """Account constructor.
@@ -81,36 +47,35 @@ def __init__(
             token: Account token to use.
             url: Authentication URL.
             instance: Service instance to use.
-            proxies: Proxies to use.
+            proxies: Proxy configuration.
             verify: Whether to verify server's TLS certificate.
         """
-        _assert_valid_auth(auth)
-        self.auth = auth
+        resolved_url = url or (LEGACY_API_URL if auth == "legacy" else CLOUD_API_URL)
 
-        _assert_valid_token(token)
+        self.auth = auth
         self.token = token
-
-        resolved_url = url or (LEGACY_API_URL if auth == "legacy" else CLOUD_API_URL)
-        _assert_valid_url(resolved_url)
         self.url = resolved_url
-
         self.instance = instance
         self.proxies = proxies
         self.verify = verify
 
     def to_saved_format(self) -> dict:
         """Returns a dictionary that represents how the account is saved on disk."""
-        return {k: v for k, v in self.__dict__.items() if v is not None}
+        result = {k: v for k, v in self.__dict__.items() if v is not None}
+        if self.proxies:
+            result["proxies"] = self.proxies.to_dict()
+        return result
 
     @classmethod
     def from_saved_format(cls, data: dict) -> "Account":
         """Creates an account instance from data saved on disk."""
+        proxies = data.get("proxies")
         return cls(
             auth=data.get("auth"),
             url=data.get("url"),
             token=data.get("token"),
             instance=data.get("instance"),
-            proxies=data.get("proxies"),
+            proxies=ProxyConfiguration(**proxies) if proxies else None,
             verify=data.get("verify", True),
         )
 
@@ -134,3 +99,64 @@ def __eq__(self, other: object) -> bool:
                 self.verify == other.verify,
             ]
         )
+
+    def validate(self) -> "Account":
+        """Validates the account instance.
+
+        Raises:
+            ValueError: if the account is invalid
+        """
+
+        self._assert_valid_auth(self.auth)
+        self._assert_valid_token(self.token)
+        self._assert_valid_url(self.url)
+        self._assert_valid_instance(self.auth, self.instance)
+        self._assert_valid_proxies(self.proxies)
+        return self
+
+    @staticmethod
+    def _assert_valid_auth(auth: AccountType) -> None:
+        """Assert that the auth parameter is valid."""
+        if not (auth in ["cloud", "legacy"]):
+            raise ValueError(
+                f"Invalid `auth` value. Expected one of ['cloud', 'legacy'], got '{auth}'."
+            )
+
+    @staticmethod
+    def _assert_valid_token(token: str) -> None:
+        """Assert that the token is valid."""
+        if not (isinstance(token, str) and len(token) > 0):
+            raise ValueError(
+                f"Invalid `token` value. Expected a non-empty string, got '{token}'."
+            )
+
+    @staticmethod
+    def _assert_valid_url(url: str) -> None:
+        """Assert that the URL is valid."""
+        try:
+            urlparse(url)
+        except:
+            raise ValueError(f"Invalid `url` value. Failed to parse '{url}' as URL.")
+
+    @staticmethod
+    def _assert_valid_proxies(config: ProxyConfiguration) -> None:
+        """Assert that the proxy configuration is valid."""
+        if config is not None:
+            config.validate()
+
+    @staticmethod
+    def _assert_valid_instance(auth: AccountType, instance: str) -> None:
+        """Assert that the instance name is valid for the given account type."""
+        if auth == "cloud":
+            if not (isinstance(instance, str) and len(instance) > 0):
+                raise ValueError(
+                    f"Invalid `instance` value. Expected a non-empty string, got '{instance}'."
+                )
+        if auth == "legacy":
+            if instance is not None:
+                try:
+                    from_instance_format(instance)
+                except:
+                    raise ValueError(
+                        f"Invalid `instance` value. Expected hub/group/project format, got {instance}"
+                    )

qiskit_ibm_runtime/accounts/management.py

@@ -16,6 +16,7 @@
 from typing import Optional, Dict
 
 from .account import Account, AccountType
+from ..proxies import ProxyConfiguration
 from .storage import save_config, read_config, delete_config
 
 _DEFAULT_ACCOUNT_CONFIG_JSON_FILE = os.path.join(
@@ -39,7 +40,7 @@ def save(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = _DEFAULT_ACCOUNT_NAME,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: Optional[bool] = None,
     ) -> None:
         """Save account on disk."""
@@ -55,7 +56,9 @@ def save(
                 auth=auth,
                 proxies=proxies,
                 verify=verify,
-            ).to_saved_format(),
+            )
+            # avoid storing invalid accounts
+            .validate().to_saved_format(),
         )
 
     @staticmethod

qiskit_ibm_runtime/api/client_parameters.py

@@ -14,9 +14,8 @@
 
 from typing import Dict, Optional, Any, Union
 
-from requests_ntlm import HttpNtlmAuth
-
 from ..api.auth import LegacyAuth, CloudAuth
+from ..proxies import ProxyConfiguration
 
 TEMPLATE_IBM_HUBS = "{prefix}/Network/{hub}/Groups/{group}/Projects/{project}"
 """str: Template for creating an IBM Quantum URL with hub/group/project information."""
@@ -31,7 +30,7 @@ def __init__(
         token: str,
         url: str = None,
         instance: Optional[str] = None,
-        proxies: Optional[Dict] = None,
+        proxies: Optional[ProxyConfiguration] = None,
         verify: bool = True,
     ) -> None:
         """ClientParameters constructor.
@@ -64,15 +63,9 @@ def connection_parameters(self) -> Dict[str, Any]:
             expected by ``requests``. The following keys can be present:
             ``proxies``, ``verify``, and ``auth``.
         """
-        request_kwargs = {"verify": self.verify}
+        request_kwargs: Any = {"verify": self.verify}
 
         if self.proxies:
-            if "urls" in self.proxies:
-                request_kwargs["proxies"] = self.proxies["urls"]
-
-            if "username_ntlm" in self.proxies and "password_ntlm" in self.proxies:
-                request_kwargs["auth"] = HttpNtlmAuth(
-                    self.proxies["username_ntlm"], self.proxies["password_ntlm"]
-                )
+            request_kwargs.update(self.proxies.to_request_params())
 
         return request_kwargs

qiskit_ibm_runtime/api/clients/base.py

@@ -26,7 +26,6 @@
 
 from ..client_parameters import ClientParameters
 from ..exceptions import WebsocketError, WebsocketTimeoutError
-from .utils import ws_proxy_params
 
 logger = logging.getLogger(__name__)
 
@@ -68,8 +67,10 @@ def __init__(
             message_queue: Queue used to hold received messages.
         """
         self._websocket_url = websocket_url.rstrip("/")
-        self._proxy_params = ws_proxy_params(
-            client_params=client_params, ws_url=self._websocket_url
+        self._proxy_params = (
+            client_params.proxies.to_ws_params(self._websocket_url)
+            if client_params.proxies
+            else {}
         )
         self._access_token = client_params.token
         self._job_id = job_id