Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion qiskit_ibm_runtime/accounts/account.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ def __init__(
self,
auth: AccountType,
token: str,
url: Optional[str],
url: Optional[str] = None,
instance: Optional[str] = None,
# TODO: add validation for proxies input format
proxies: Optional[dict] = None,
Expand Down Expand Up @@ -102,3 +102,17 @@ def from_saved_format(cls, data: dict) -> "Account":
proxies=data.get("proxies"),
verify=data.get("verify"),
)

def __eq__(self, other: object) -> bool:
if not isinstance(other, Account):
return False
return all(
[
self.auth == other.auth,
self.token == other.token,
self.url == other.url,
self.instance == other.instance,
self.proxies == other.proxies,
self.verify == other.verify,
]
)
68 changes: 39 additions & 29 deletions qiskit_ibm_runtime/accounts/management.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,59 +13,69 @@
"""Account management related classes and functions."""

import os
from typing import Optional, Union
from typing import Optional

from .account import Account, AccountType
from .storage import save_config, read_config, delete_config

_DEFAULT_ACCOUNG_CONFIG_JSON_FILE = os.path.join(
os.path.expanduser("~"), ".qiskit", "qiskit-ibm.json"
)
_DEFAULT_ACCOUNT_NAME = "default"
_DEFAULT_ACCOUNT_NAME_LEGACY = "default-legacy"
_DEFAULT_ACCOUNT_NAME_CLOUD = "default-cloud"
_DEFAULT_ACCOUNT_TYPE: AccountType = "cloud"


class AccountManager:
"""Class that bundles account management related functionality."""

@staticmethod
def save(
token: Optional[str] = None,
url: Optional[str] = None,
instance: Optional[str] = None,
auth: Optional[AccountType] = None,
name: Optional[str] = _DEFAULT_ACCOUNT_NAME,
proxies: Optional[dict] = None,
verify: Optional[bool] = None,
) -> None:
def save(account: Account, name: Optional[str] = None) -> None:
"""Save account on disk."""

config_key = name or _get_default_account_name(account.auth)
return save_config(
filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE,
name=name,
config=Account(
token=token,
url=url,
instance=instance,
auth=auth,
proxies=proxies,
verify=verify,
).to_saved_format(),
name=config_key,
config=account.to_saved_format(),
)

@staticmethod
def list() -> Union[dict, None]:
"""List all accounts saved on disk."""

return read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE)
def list(
default: Optional[bool] = None, auth: Optional[str] = None
) -> dict[str, Account]:
"""List all accounts saved on disk by name."""
return dict(
filter(
lambda input: input[1],
map(
lambda kv: (kv[0], Account.from_saved_format(kv[1])),
read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE).items(),
),
),
)

@staticmethod
def get(name: Optional[str] = _DEFAULT_ACCOUNT_NAME) -> Account:
def get(
name: Optional[str] = None, auth: Optional[str] = _DEFAULT_ACCOUNT_TYPE
) -> Account:
"""Read account from disk."""
config_key = name or _get_default_account_name(auth)
return Account.from_saved_format(
read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE, name=name)
read_config(filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE, name=config_key)
)

@staticmethod
def delete(name: Optional[str] = _DEFAULT_ACCOUNT_NAME) -> bool:
def delete(
name: Optional[str] = None, auth: Optional[str] = _DEFAULT_ACCOUNT_TYPE
) -> bool:
"""Delete account from disk."""
return delete_config(name=name, filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE)
config_key = name or _get_default_account_name(auth)
return delete_config(
name=config_key, filename=_DEFAULT_ACCOUNG_CONFIG_JSON_FILE
)


def _get_default_account_name(auth: AccountType):
return (
_DEFAULT_ACCOUNT_NAME_CLOUD if auth == "cloud" else _DEFAULT_ACCOUNT_NAME_LEGACY
)
6 changes: 6 additions & 0 deletions qiskit_ibm_runtime/accounts/storage.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,13 @@

"""Utility functions related to storing account configuration on disk."""

import logging
import json
import os
from typing import Optional, Union

logger = logging.getLogger(__name__)


def save_config(
filename: str,
Expand All @@ -24,6 +27,7 @@ def save_config(
) -> None:
"""Save configuration data in a JSON file under the given name."""

logger.debug(f"Save configuration data for '{name}' in '{filename}'")
_ensure_file_exists(filename)

with open(filename, mode="r") as json_in:
Expand All @@ -40,6 +44,7 @@ def read_config(
) -> Union[dict, None]:
"""Save configuration data from a JSON file."""

logger.debug(f"Read configuration data for '{name}' from '{filename}'")
_ensure_file_exists(filename)

with open(filename) as json_file:
Expand Down Expand Up @@ -73,6 +78,7 @@ def delete_config(

def _ensure_file_exists(filename: str, initial_content: str = "{}") -> None:
if not os.path.isfile(filename):
logger.debug(f"Create empty configuration file at {filename}")

# create parent directories
os.makedirs(os.path.dirname(filename), exist_ok=True)
Expand Down
127 changes: 127 additions & 0 deletions test/ibm/test_accounts.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,127 @@
import json
from contextlib import ContextDecorator
from tempfile import NamedTemporaryFile
from typing import Optional, List
from unittest import TestCase

from qiskit_ibm_runtime.accounts import AccountManager, Account
from qiskit_ibm_runtime.accounts import management
from ..ibm_test_case import IBMTestCase

_TEST_LEGACY_ACCOUNT = Account(
"legacy",
"token-x",
"https://auth.quantum-computing.ibm.com/api",
"ibm-q/open/main",
)

_TEST_CLOUD_ACCOUNT = Account(
auth="cloud",
token="token-y",
url="https://cloud.ibm.com",
instance="crn:v1:bluemix:public:quantum-computing:us-east:a/...::",
)


class temporary_account_config_file(ContextDecorator):
"""Context manager that uses a temporary account configuration file for test purposes."""

def __init__(self, contents: Optional[str] = "{}"):
# Create a temporary file with provided contents.
self.tmp_file = NamedTemporaryFile(mode="w")
self.tmp_file.write(contents)
self.tmp_file.flush()
self.account_config_json_backup = management._DEFAULT_ACCOUNG_CONFIG_JSON_FILE

def __enter__(self):
# Temporarily modify the default location of the configuration file.
management._DEFAULT_ACCOUNG_CONFIG_JSON_FILE = self.tmp_file.name
return self

def __exit__(self, *exc):
# Delete the temporary file and restore the default location.
self.tmp_file.close()
management._DEFAULT_ACCOUNG_CONFIG_JSON_FILE = self.account_config_json_backup


class TestAccount(TestCase):
def test_init(self):
inputs = [("auth")]
a = ""


class TestAccountManager(IBMTestCase, TestCase):
@temporary_account_config_file()
def test_save_get(self):

# Each tuple contains the
# - account to save
# - the name passed to AccountManager.save
# - the name passed to AccountManager.get
sub_tests = [
# verify accounts can be saved and retrieved via custom names
(_TEST_LEGACY_ACCOUNT, "acct-1", "acct-1"),
(_TEST_CLOUD_ACCOUNT, "acct-2", "acct-2"),
# verify default account name handling for cloud accounts
(_TEST_CLOUD_ACCOUNT, None, management._DEFAULT_ACCOUNT_NAME_CLOUD),
(_TEST_CLOUD_ACCOUNT, None, None),
# verify default account name handling for legacy accounts
(_TEST_LEGACY_ACCOUNT, None, management._DEFAULT_ACCOUNT_NAME_LEGACY),
# verify account override
(_TEST_LEGACY_ACCOUNT, "acct", "acct"),
(_TEST_CLOUD_ACCOUNT, "acct", "acct"),
]
for account, name_save, name_get in sub_tests:
with self.subTest(
f"for account type '{account.auth}' "
f"using `save(name={name_save})` and `get(name={name_get})`"
):
AccountManager.save(account=account, name=name_save)
self.assertEqual(account, AccountManager.get(name=name_get))

@temporary_account_config_file(
contents=json.dumps(
{
"cloud": _TEST_CLOUD_ACCOUNT.to_saved_format(),
"legacy": _TEST_LEGACY_ACCOUNT.to_saved_format(),
}
)
)
def test_list(self):
with temporary_account_config_file(
contents=json.dumps(
{
"key1": _TEST_CLOUD_ACCOUNT.to_saved_format(),
"key2": _TEST_LEGACY_ACCOUNT.to_saved_format(),
}
)
), self.subTest("non-empty list of accounts and filtering"):
accounts = AccountManager.list()
self.assertEqual(len(accounts), 2)
self.assertEqual(accounts["key1"], _TEST_CLOUD_ACCOUNT)
self.assertTrue(accounts["key2"], _TEST_LEGACY_ACCOUNT)

with temporary_account_config_file(), self.subTest("empty list of accounts"):
self.assertEqual(len(AccountManager.list()), 0)

@temporary_account_config_file(
contents=json.dumps(
{
"key1": _TEST_CLOUD_ACCOUNT.to_saved_format(),
management._DEFAULT_ACCOUNT_NAME_LEGACY: _TEST_LEGACY_ACCOUNT.to_saved_format(),
management._DEFAULT_ACCOUNT_NAME_CLOUD: _TEST_CLOUD_ACCOUNT.to_saved_format(),
}
)
)
def test_delete(self):
with self.subTest("delete named account"):
self.assertTrue(AccountManager.delete(name="key1"))
self.assertFalse(AccountManager.delete(name="key1"))

with self.subTest("delete default legacy account"):
self.assertTrue(AccountManager.delete(auth="legacy"))

with self.subTest("delete default cloud account"):
self.assertTrue(AccountManager.delete())

self.assertEquals(len(AccountManager.list()), 0)