prepare public interface and test cases for Cloud auth

qiskit_ibm_runtime/ibm_runtime_service.py

@@ -13,27 +13,36 @@
 """Qiskit runtime service."""
 
 import copy
-import logging
-from collections import OrderedDict
-from typing import Dict, Callable, Optional, Union, List, Any, Type, Tuple
 import json
+import logging
 import os
 import re
 import traceback
 import warnings
+from collections import OrderedDict
+from typing import Dict, Callable, Optional, Union, List, Any, Type, Tuple
 
 from qiskit.circuit import QuantumCircuit
 from qiskit.providers.backend import BackendV1 as Backend
 from qiskit.providers.exceptions import QiskitBackendNotFoundError
 from qiskit.providers.providerutils import filter_backends
 from qiskit.transpiler import Layout
+from typing_extensions import Literal
 
 from qiskit_ibm_runtime import runtime_job, ibm_backend  # pylint: disable=unused-import
-
-from .runtime_job import RuntimeJob
-from .runtime_program import RuntimeProgram, ParameterNamespace
-from .utils import RuntimeDecoder, to_base64_string, to_python_identifier
-from .utils.backend import convert_reservation_data
+from .api.clients import AuthClient, VersionClient
+from .api.clients.runtime import RuntimeClient
+from .api.exceptions import RequestsApiError
+from .apiconstants import QISKIT_IBM_RUNTIME_API_URL
+from .backendreservation import BackendReservation
+from .credentials import Credentials, HubGroupProjectID, discover_credentials
+from .credentials.configrc import (
+    remove_credentials,
+    read_credentials_from_qiskitrc,
+    store_credentials,
+)
+from .credentials.exceptions import HubGroupProjectIDInvalidStateError
+from .exceptions import IBMNotAuthorizedError, IBMInputValueError, IBMProviderError
 from .exceptions import (
     QiskitRuntimeError,
     RuntimeDuplicateProgramError,
@@ -46,27 +55,20 @@
     IBMProviderCredentialsInvalidUrl,
     IBMProviderValueError,
 )
-from .program.result_decoder import ResultDecoder
-from .api.clients import AuthClient, VersionClient
-from .api.clients.runtime import RuntimeClient
-from .apiconstants import QISKIT_IBM_RUNTIME_API_URL
-from .api.exceptions import RequestsApiError
-from .backendreservation import BackendReservation
 from .hub_group_project import HubGroupProject  # pylint: disable=cyclic-import
-from .exceptions import IBMNotAuthorizedError, IBMInputValueError, IBMProviderError
-from .credentials import Credentials, HubGroupProjectID, discover_credentials
-from .credentials.configrc import (
-    remove_credentials,
-    read_credentials_from_qiskitrc,
-    store_credentials,
-)
-from .credentials.exceptions import HubGroupProjectIDInvalidStateError
+from .program.result_decoder import ResultDecoder
 from .runner_result import RunnerResult
+from .runtime_job import RuntimeJob
+from .runtime_program import RuntimeProgram, ParameterNamespace
+from .utils import RuntimeDecoder, to_base64_string, to_python_identifier
+from .utils.backend import convert_reservation_data
 
 logger = logging.getLogger(__name__)
 
 SERVICE_NAME = "runtime"
 
+AuthTypes = Literal["cloud", "legacy"]
+
 
 class IBMRuntimeService:
     """Class for interacting with the Qiskit Runtime service.
@@ -127,13 +129,19 @@ class IBMRuntimeService:
     """
 
     def __init__(
-        self, token: Optional[str] = None, url: Optional[str] = None, **kwargs: Any
+        self,
+        auth: Optional[AuthTypes] = None,
+        token: Optional[str] = None,
+        locator: Optional[str] = None,
+        **kwargs: Any,
     ) -> None:
         """IBMRuntimeService constructor
 
         Args:
-            token: IBM Quantum token.
-            url: URL for the IBM Quantum authentication server.
+            auth: Authentication type. `cloud` or `legacy`. If not specified, the saved default is used.
+                If there is no default value, and both accounts were saved on disk, the cloud type is used.
+            token: Token used for authentication. If not specified, the saved token is used.
+            locator: The authentication url, if `auth=legacy`. Otherwise the CRN.
             **kwargs: Additional settings for the connection:
 
                 * proxies (dict): proxy configuration.
@@ -152,8 +160,11 @@ def __init__(
         """
         # pylint: disable=unused-argument,unsubscriptable-object
         super().__init__()
+        if auth == "cloud":
+            raise NotImplementedError
+
         account_credentials, account_preferences = self._resolve_credentials(
-            token=token, url=url, **kwargs
+            token=token, url=locator, **kwargs
         )
         self._initialize_hgps(
             credentials=account_credentials, preferences=account_preferences

qiskit_ibm_runtime/jupyter/dashboard/dashboard.py

@@ -179,7 +179,7 @@ def ibm_quantum_dashboard(self, line="", cell=None) -> None:
         """A Jupyter magic function to enable the dashboard."""
         # pylint: disable=unused-argument
         try:
-            service = IBMRuntimeService()
+            service = IBMRuntimeService(auth="legacy")
         except Exception:
             raise QiskitError("Could not load IBM Quantum account from the local file.")
         _IBM_DASHBOARD.stop_dashboard()

requirements.txt

@@ -5,3 +5,4 @@ numpy>=1.13
 urllib3>=1.21.1
 python-dateutil>=2.8.0
 websocket-client>=1.0.1
+typing-extensions>=4.0.0

setup.py

@@ -25,18 +25,19 @@
     "numpy>=1.13",
     "urllib3>=1.21.1",
     "python-dateutil>=2.8.0",
-    "websocket-client>=1.0.1"
+    "websocket-client>=1.0.1",
+    "typing-extensions>=4.0.0",  # remove when support for Python 3.7 is dropped (use "from typing import" instead)
 ]
 
 # Handle version.
-VERSION_PATH = os.path.join(os.path.dirname(__file__),
-                            "qiskit_ibm_runtime", "VERSION.txt")
+VERSION_PATH = os.path.join(
+    os.path.dirname(__file__), "qiskit_ibm_runtime", "VERSION.txt"
+)
 with open(VERSION_PATH, "r") as version_file:
     VERSION = version_file.read().strip()
 
 # Read long description from README.
-README_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)),
-                           'README.md')
+README_PATH = os.path.join(os.path.abspath(os.path.dirname(__file__)), "README.md")
 with open(README_PATH) as readme_file:
     README = readme_file.read()
 
@@ -45,9 +46,9 @@
     name="qiskit-ibm-runtime",
     version=VERSION,
     description="Qiskit IBM Runtime service for accessing the quantum devices and "
-                "simulators at IBM",
+    "simulators at IBM",
     long_description=README,
-    long_description_content_type='text/markdown',
+    long_description_content_type="text/markdown",
     url="https://github.com/Qiskit/qiskit-ibm-runtime",
     author="Qiskit Development Team",
     author_email="[email protected]",
@@ -68,16 +69,24 @@
         "Topic :: Scientific/Engineering",
     ],
     keywords="qiskit sdk quantum api runtime ibm",
-    packages=setuptools.find_packages(exclude=['test*']),
+    packages=setuptools.find_packages(exclude=["test*"]),
     install_requires=REQUIREMENTS,
     include_package_data=True,
     python_requires=">=3.6",
     zip_safe=False,
-    extras_require={'visualization': ['matplotlib>=2.1', 'ipywidgets>=7.3.0',
-                                      "seaborn>=0.9.0", "plotly>=4.4",
-                                      "ipyvuetify>=1.1", "pyperclip>=1.7",
-                                      "ipython>=5.0.0", "traitlets!=5.0.5",
-                                      "ipyvue>=1.4.1"]},
+    extras_require={
+        "visualization": [
+            "matplotlib>=2.1",
+            "ipywidgets>=7.3.0",
+            "seaborn>=0.9.0",
+            "plotly>=4.4",
+            "ipyvuetify>=1.1",
+            "pyperclip>=1.7",
+            "ipython>=5.0.0",
+            "traitlets!=5.0.5",
+            "ipyvue>=1.4.1",
+        ]
+    },
     project_urls={
         "Bug Tracker": "https://github.com/Qiskit/qiskit-ibm-runtime/issues",
         "Documentation": "https://qiskit.org/documentation/",

test/decorators.py

@@ -88,7 +88,7 @@ def requires_providers(func):
     def _wrapper(*args, **kwargs):
         qe_token = kwargs.pop("qe_token")
         qe_url = kwargs.pop("qe_url")
-        service = IBMRuntimeService(qe_token, qe_url)
+        service = IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
         # Get open access hgp
         open_hgp = _get_open_hgp(service)
         if not open_hgp:
@@ -140,7 +140,7 @@ def requires_provider(func):
     def _wrapper(*args, **kwargs):
         token = kwargs.pop("qe_token")
         url = kwargs.pop("qe_url")
-        service = IBMRuntimeService(token, url)
+        service = IBMRuntimeService(auth="legacy", token=token, locator=url)
         hub, group, project = _get_custom_hgp()
         kwargs.update(
             {"service": service, "hub": hub, "group": group, "project": project}
@@ -168,7 +168,7 @@ def requires_private_provider(func):
     def _wrapper(*args, **kwargs):
         token = kwargs.pop("qe_token")
         url = kwargs.pop("qe_url")
-        service = IBMRuntimeService(token, url)
+        service = IBMRuntimeService(auth="legacy", token=token, locator=url)
         hub, group, project = _get_private_hgp()
         kwargs.update(
             {"service": service, "hub": hub, "group": group, "project": project}
@@ -250,7 +250,7 @@ def _wrapper(obj, *args, **kwargs):
 
 def _get_backend(qe_token, qe_url, backend_name):
     """Get the specified backend."""
-    service = IBMRuntimeService(qe_token, qe_url)
+    service = IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
     _backend = None
     hub, group, project = _get_custom_hgp()
     if backend_name:

test/ibm/runtime/test_runtime.py

@@ -129,7 +129,7 @@ def setUp(self):
         """Initial test setup."""
         super().setUp()
         with mock_ibm_provider():
-            self.service = IBMRuntimeService("abc")
+            self.service = IBMRuntimeService(auth="legacy", token="abc")
         self.service._programs = {}
         self.service._default_hgp = mock.MagicMock(spec=HubGroupProject)
         self.service._default_hgp.credentials = Credentials(

test/ibm/test_ibm_provider.py

@@ -53,7 +53,7 @@ class TestIBMProviderEnableAccount(IBMTestCase):
     def test_provider_init_token(self, qe_token, qe_url):
         """Test initializing IBMRuntimeService with only API token."""
         # pylint: disable=unused-argument
-        service = IBMRuntimeService(token=qe_token)
+        service = IBMRuntimeService(auth="legacy", token=qe_token)
         self.assertIsInstance(service, IBMRuntimeService)
         self.assertEqual(service._default_hgp.credentials.token, qe_token)
 
@@ -67,7 +67,9 @@ def test_pass_unreachable_proxy(self, qe_token, qe_url):
             }
         }
         with self.assertRaises(RequestsApiError) as context_manager:
-            IBMRuntimeService(qe_token, qe_url, proxies=proxies)
+            IBMRuntimeService(
+                auth="legacy", token=qe_token, locator=qe_url, proxies=proxies
+            )
         self.assertIn("ProxyError", str(context_manager.exception))
 
     def test_provider_init_non_auth_url(self):
@@ -76,7 +78,7 @@ def test_provider_init_non_auth_url(self):
         qe_url = API_URL
 
         with self.assertRaises(IBMProviderCredentialsInvalidUrl) as context_manager:
-            IBMRuntimeService(token=qe_token, url=qe_url)
+            IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
 
         self.assertIn("authentication URL", str(context_manager.exception))
 
@@ -86,7 +88,7 @@ def test_provider_init_non_auth_url_with_hub(self):
         qe_url = API_URL + "/Hubs/X/Groups/Y/Projects/Z"
 
         with self.assertRaises(IBMProviderCredentialsInvalidUrl) as context_manager:
-            IBMRuntimeService(token=qe_token, url=qe_url)
+            IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
 
         self.assertIn("authentication URL", str(context_manager.exception))
 
@@ -95,7 +97,7 @@ def test_provider_init_no_credentials(self):
         with custom_qiskitrc(), self.assertRaises(
             IBMProviderCredentialsNotFound
         ) as context_manager, no_envs(CREDENTIAL_ENV_VARS):
-            IBMRuntimeService()
+            IBMRuntimeService(auth="legacy")
 
         self.assertIn(
             "No IBM Quantum credentials found.", str(context_manager.exception)
@@ -112,7 +114,7 @@ def test_discover_backend_failed(self, qe_token, qe_url):
             with self.assertLogs(
                 hub_group_project.logger, level="WARNING"
             ) as context_manager:
-                IBMRuntimeService(qe_token, qe_url)
+                IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
         self.assertIn("bad_backend", str(context_manager.output))
 
 
@@ -144,7 +146,7 @@ def test_provider_init_saved_account(self, qe_token, qe_url):
 
         with custom_qiskitrc(), no_envs(CREDENTIAL_ENV_VARS):
             IBMRuntimeService.save_account(qe_token, url=qe_url)
-            service = IBMRuntimeService()
+            service = IBMRuntimeService(auth="legacy")
 
         self.assertIsInstance(service, IBMRuntimeService)
         self.assertEqual(service._default_hgp.credentials.token, qe_token)
@@ -223,7 +225,7 @@ def test_load_account_saved_provider(self, qe_token, qe_url):
                 group=non_default_hgp.credentials.group,
                 project=non_default_hgp.credentials.project,
             )
-            saved_provider = IBMRuntimeService()
+            saved_provider = IBMRuntimeService(auth="legacy")
             if saved_provider._default_hgp != non_default_hgp:
                 # Prevent tokens from being logged.
                 saved_provider._default_hgp.credentials.token = None
@@ -268,7 +270,7 @@ def test_load_saved_account_invalid_hgp(self, qe_token, qe_url):
                     group=hgp_id.group,
                     project=hgp_id.project,
                 )
-                IBMRuntimeService()
+                IBMRuntimeService(auth="legacy")
 
             self.assertIn(
                 "No hub/group/project matches the specified criteria",
@@ -296,13 +298,13 @@ def test_load_saved_account_invalid_hgp_format(self):
                         _file.write("default_provider = {}".format(invalid_hgp))
                     # Ensure an error is raised if the stored provider is in an invalid format.
                     with self.assertRaises(IBMProviderError) as context_manager:
-                        IBMRuntimeService()
+                        IBMRuntimeService(auth="legacy")
                     self.assertIn(error_message, str(context_manager.exception))
 
     @requires_qe_access
     def test_active_account(self, qe_token, qe_url):
         """Test get active account"""
-        service = IBMRuntimeService(qe_token, qe_url)
+        service = IBMRuntimeService(auth="legacy", token=qe_token, url=qe_url)
         active_account = service.active_account()
         self.assertIsNotNone(active_account)
         self.assertEqual(active_account["token"], qe_token)
@@ -330,7 +332,7 @@ class TestIBMProviderHubGroupProject(IBMTestCase):
     @requires_qe_access
     def _initialize_provider(self, qe_token=None, qe_url=None):
         """Initialize and return provider."""
-        return IBMRuntimeService(qe_token, qe_url)
+        return IBMRuntimeService(auth="legacy", token=qe_token, locator=qe_url)
 
     def setUp(self):
         """Initial test setup."""

test/ibm/test_proxies.py

@@ -56,7 +56,12 @@ def tearDown(self):
     @requires_qe_access
     def test_proxies_ibm_account(self, qe_token, qe_url):
         """Should reach the proxy using account.enable."""
-        service = IBMRuntimeService(qe_token, qe_url, proxies={"urls": VALID_PROXIES})
+        service = IBMRuntimeService(
+            auth="legacy",
+            token=qe_token,
+            locator=qe_url,
+            proxies={"urls": VALID_PROXIES},
+        )
 
         self.proxy_process.terminate()  # kill to be able of reading the output
 

test/ibm/test_registration.py

@@ -64,7 +64,7 @@ def test_load_account_no_credentials(self) -> None:
 
         with custom_qiskitrc(), no_envs(CREDENTIAL_ENV_VARS):
             with self.assertRaises(IBMProviderError) as context_manager:
-                IBMRuntimeService()
+                IBMRuntimeService(auth="legacy")
 
         self.assertIn(
             "No IBM Quantum credentials found", str(context_manager.exception)
@@ -101,7 +101,7 @@ def test_store_credentials_overwrite(self) -> None:
             with no_envs(CREDENTIAL_ENV_VARS), mock_ibm_provider():
                 # Attempt overwriting.
                 store_credentials(credentials2, overwrite=True)
-                service = IBMRuntimeService()
+                service = IBMRuntimeService(auth="legacy")
 
         # Ensure that the credentials are the overwritten ones.
         # pylint: disable=no-member

test/utils.py

@@ -146,7 +146,9 @@ def get_hgp(qe_token: str, qe_url: str, default: bool = True) -> HubGroupProject
     Returns:
         A HubGroupProject, as specified by `default`.
     """
-    service = IBMRuntimeService(qe_token, url=qe_url)  # Default hub/group/project.
+    service = IBMRuntimeService(
+        auth="legacy", token=qe_token, locator=qe_url
+    )  # Default hub/group/project.
     open_hgp = service._get_hgp()  # Open access hgp
     hgp_to_return = open_hgp
     if not default: