Improve test coverage and documentation for proxy configuration

qiskit_ibm_runtime/accounts/__init__.py

@@ -14,5 +14,5 @@
 Account management functionality related to the IBM Runtime Services.
 """
 
-from .account import Account, AccountType
+from .account import Account, AccountType, ProxyConfigurationType
 from .management import AccountManager

qiskit_ibm_runtime/accounts/account.py

@@ -13,16 +13,32 @@
 """Account related classes and functions."""
 
 
-from typing import Optional
+from typing import Optional, Dict
 from urllib.parse import urlparse
 
-from typing_extensions import Literal
 from requests.auth import AuthBase
-
+from typing_extensions import Literal, TypedDict
+from ..utils.hgp import from_instance_format
 from ..api.auth import LegacyAuth, CloudAuth
 
 AccountType = Optional[Literal["cloud", "legacy"]]
 
+
+class ProxyConfigurationType(TypedDict, total=False):
+    """Dictionary type for custom proxy configuration.
+
+    All items in the dictionary are optional. When ``urls`` are provided, they must contain a dictionary
+    mapping protocol or protocol and host to the URL of the proxy. Refer to
+    https://docs.python-requests.org/en/latest/api/#requests.Session.proxies for details and examples.
+
+    NTLM user authentication can be enabled by setting ``username_ntlm`` and ``password_ntlm``.
+    """
+
+    urls: Dict[str, str]
+    username_ntlm: str
+    password_ntlm: str
+
+
 LEGACY_API_URL = "https://auth.quantum-computing.ibm.com/api"
 CLOUD_API_URL = "https://us-east.quantum-computing.cloud.ibm.com"
 
@@ -31,15 +47,15 @@ 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}'."
+            f"Invalid `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}'."
+            f"Invalid `token` value. Expected a non-empty string, got '{token}'."
         )
 
 
@@ -48,17 +64,51 @@ def _assert_valid_url(url: str) -> None:
     try:
         urlparse(url)
     except:
-        raise ValueError(f"Inappropriate `url` value. Failed to parse '{url}' as URL.")
+        raise ValueError(f"Invalid `url` value. Failed to parse '{url}' as URL.")
+
+
+def _assert_valid_proxies(config: ProxyConfigurationType) -> None:
+    """Assert that the proxy configuration is valid."""
+
+    if config is None:
+        return
+
+    # verify NTLM authentication configuration
+    ntlm_user = config.get("username_ntlm")
+    ntlm_pass = config.get("password_ntlm")
+    if not any(
+        [
+            isinstance(ntlm_user, str) and isinstance(ntlm_pass, str),
+            ntlm_user is None and ntlm_pass is None,
+        ]
+    ):
+        raise ValueError(
+            f"Invalid proxy configuration for NTLM authentication. None or both of username and "
+            f"password must be provided. Got username_ntlm={ntlm_user}, password_ntlm={ntlm_pass}."
+        )
+
+    # verify proxy configuration
+    urls = config.get("urls")
+    if urls is not None and not isinstance(urls, dict):
+        raise ValueError(
+            f"Invalid proxy configuration. Expected `urls` to contain a dictionary mapping protocol "
+            f"or protocol and host to the URL of the proxy. Got {urls}"
+        )
 
 
 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
+        if not (isinstance(instance, str) and instance.find("crn:") != -1):
+            raise ValueError(f"Invalid `instance` value. Expected a non-empty CRN.")
+    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}"
+                )
 
 
 class Account:
@@ -70,8 +120,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[ProxyConfigurationType] = None,
         verify: Optional[bool] = True,
     ):
         """Account constructor.
@@ -81,7 +130,7 @@ 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)
@@ -94,8 +143,12 @@ def __init__(
         _assert_valid_url(resolved_url)
         self.url = resolved_url
 
+        _assert_valid_instance(auth, instance)
         self.instance = instance
+
+        _assert_valid_proxies(proxies)
         self.proxies = proxies
+
         self.verify = verify
 
     def to_saved_format(self) -> dict:

qiskit_ibm_runtime/accounts/management.py

@@ -15,7 +15,7 @@
 import os
 from typing import Optional, Dict
 
-from .account import Account, AccountType
+from .account import Account, AccountType, ProxyConfigurationType
 from .storage import save_config, read_config, delete_config
 
 _DEFAULT_ACCOUNT_CONFIG_JSON_FILE = os.path.join(
@@ -39,7 +39,7 @@ def save(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = _DEFAULT_ACCOUNT_NAME,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfigurationType] = None,
         verify: Optional[bool] = None,
     ) -> None:
         """Save account on disk."""

qiskit_ibm_runtime/api/client_parameters.py

@@ -15,7 +15,7 @@
 from typing import Dict, Optional, Any, Union
 
 from requests_ntlm import HttpNtlmAuth
-
+from ..accounts import ProxyConfigurationType
 from ..api.auth import LegacyAuth, CloudAuth
 
 TEMPLATE_IBM_HUBS = "{prefix}/Network/{hub}/Groups/{group}/Projects/{project}"
@@ -31,7 +31,7 @@ def __init__(
         token: str,
         url: str = None,
         instance: Optional[str] = None,
-        proxies: Optional[Dict] = None,
+        proxies: Optional[ProxyConfigurationType] = None,
         verify: bool = True,
     ) -> None:
         """ClientParameters constructor.
@@ -64,7 +64,7 @@ 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:

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -24,8 +24,8 @@
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit.providers.providerutils import filter_backends
 
-from qiskit_ibm_runtime import ibm_backend  # pylint: disable=unused-import
-from .accounts import AccountManager, Account, AccountType
+from qiskit_ibm_runtime import ibm_backend
+from .accounts import AccountManager, Account, AccountType, ProxyConfigurationType
 from .accounts.exceptions import AccountsError
 from .api.clients import AuthClient, VersionClient
 from .api.clients.runtime import RuntimeClient
@@ -117,7 +117,7 @@ def __init__(
         url: Optional[str] = None,
         name: Optional[str] = None,
         instance: Optional[str] = None,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfigurationType] = None,
         verify: Optional[bool] = None,
     ) -> None:
         """IBMRuntimeService constructor
@@ -143,7 +143,7 @@ def __init__(
             name: Name of the account to load.
             instance: The service instance to use. For cloud runtime, this is the Cloud Resource
                 Name (CRN). For legacy runtime, this is the hub/group/project in that format.
-            proxies: Proxy configuration for the server.
+            proxies: Proxy configuration.
             verify: Whether to verify the server's TLS certificate.
 
         Returns:
@@ -211,7 +211,7 @@ def _discover_account(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = None,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfigurationType] = None,
         verify: Optional[bool] = None,
     ) -> Account:
         """Discover account."""
@@ -536,7 +536,7 @@ def save_account(
         instance: Optional[str] = None,
         auth: Optional[AccountType] = None,
         name: Optional[str] = None,
-        proxies: Optional[dict] = None,
+        proxies: Optional[ProxyConfigurationType] = None,
         verify: Optional[bool] = None,
     ) -> None:
         """Save the account to disk for future use.
@@ -549,7 +549,7 @@ def save_account(
             instance: The CRN (cloud) or hub/group/project (legacy).
             auth: Authentication type. `cloud` or `legacy`.
             name: Name of the account to save.
-            proxies: Proxy configuration for the server.
+            proxies: Proxy configuration.
             verify: Verify the server's TLS certificate.
         """
 

test/test_account.py

@@ -13,15 +13,15 @@
 """Tests for the account functions."""
 
 import json
-import uuid
 import logging
 import os
+import uuid
+from typing import Any
 from unittest import skipIf
 
-from qiskit_ibm_runtime.accounts.account import CLOUD_API_URL, LEGACY_API_URL
 from qiskit_ibm_runtime.accounts import AccountManager, Account, management
+from qiskit_ibm_runtime.accounts.account import CLOUD_API_URL, LEGACY_API_URL
 from qiskit_ibm_runtime.exceptions import IBMInputValueError
-
 from .ibm_test_case import IBMTestCase
 from .mock.fake_runtime_service import FakeRuntimeService
 from .utils.account import (
@@ -46,6 +46,83 @@
 )
 
 
+class TestAccount(IBMTestCase):
+    """Tests for Account class."""
+
+    dummy_token = "123"
+    dummy_cloud_url = "https://us-east.quantum-computing.cloud.ibm.com"
+    dummy_legacy_url = "https://auth.quantum-computing.ibm.com/api"
+
+    def test_invalid_auth(self):
+        """Test invalid values for auth parameter."""
+
+        with self.assertRaises(ValueError) as err:
+            invalid_auth: Any = "phantom"
+            Account(auth=invalid_auth, token=self.dummy_token, url=self.dummy_cloud_url)
+        self.assertIn("Invalid `auth` value.", str(err.exception))
+
+    def test_invalid_token(self):
+        """Test invalid values for token parameter."""
+
+        invalid_tokens = [1, None, ""]
+        for token in invalid_tokens:
+            with self.subTest(token=token):
+                with self.assertRaises(ValueError) as err:
+                    Account(auth="cloud", token=token, url=self.dummy_cloud_url)
+                self.assertIn("Invalid `token` value.", str(err.exception))
+
+    def test_invalid_url(self):
+        """Test invalid values for url parameter."""
+
+        subtests = [
+            {"auth": "cloud", "url": 123},
+        ]
+        for params in subtests:
+            with self.subTest(params=params):
+                with self.assertRaises(ValueError) as err:
+                    Account(**params, token=self.dummy_token)
+                self.assertIn("Invalid `url` value.", str(err.exception))
+
+    def test_invalid_instance(self):
+        """Test invalid values for instance parameter."""
+
+        subtests = [
+            {"auth": "cloud", "instance": "no-valid-crn"},
+            {"auth": "cloud"},
+            {"auth": "legacy", "instance": "no-hgp-format"},
+        ]
+        for params in subtests:
+            with self.subTest(params=params):
+                with self.assertRaises(ValueError) as err:
+                    Account(**params, token=self.dummy_token, url=self.dummy_cloud_url)
+                self.assertIn("Invalid `instance` value.", str(err.exception))
+
+    def test_invalid_proxy_config(self):
+        """Test invalid values for proxy configuration."""
+
+        subtests = [
+            {
+                "proxies": {"username_ntlm": "user-only"},
+            },
+            {
+                "proxies": {"password_ntlm": "password-only"},
+            },
+            {
+                "proxies": {"urls": ""},
+            },
+        ]
+        for params in subtests:
+            with self.subTest(params=params):
+                with self.assertRaises(ValueError) as err:
+                    Account(
+                        **params,
+                        auth="legacy",
+                        token=self.dummy_token,
+                        url=self.dummy_cloud_url,
+                    )
+                self.assertIn("Invalid proxy configuration", str(err.exception))
+
+
 # NamedTemporaryFiles not supported in Windows
 @skipIf(os.name == "nt", "Test not supported in Windows")
 class TestAccountManager(IBMTestCase):