diff --git a/solana/utils/helpers.py b/solana/utils/helpers.py index 18d36db6..a443a857 100644 --- a/solana/utils/helpers.py +++ b/solana/utils/helpers.py @@ -1,5 +1,8 @@ """Helper functions.""" +from base64 import b64decode +from base58 import b58decode + def from_uint8_bytes(uint8: bytes) -> int: """Convert from uint8 to python int.""" @@ -9,3 +12,14 @@ def from_uint8_bytes(uint8: bytes) -> int: def to_uint8_bytes(val: int) -> bytes: """Convert an integer to uint8.""" return val.to_bytes(1, byteorder="little") + + +def decode_byte_string(byte_string: str, encoding: str = "base64") -> bytes: + """Decode a encoded string from an RPC Response.""" + b_str = str.encode(byte_string) + if encoding == "base64": + return b64decode(b_str) + if encoding == "base58": + return b58decode(b_str) + + raise NotImplementedError(f"{encoding} decoding not currently supported.") diff --git a/spl/token/core.py b/spl/token/core.py index 17fb04dc..c0ad881e 100644 --- a/spl/token/core.py +++ b/spl/token/core.py @@ -12,10 +12,10 @@ from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Commitment, Confirmed from solana.rpc.types import RPCResponse, TokenAccountOpts, TxOpts +from solana.utils.helpers import decode_byte_string from solana.transaction import Transaction from spl.token._layouts import ACCOUNT_LAYOUT, MINT_LAYOUT, MULTISIG_LAYOUT # type: ignore from spl.token.constants import WRAPPED_SOL_MINT -from tests.integration.utils import decode_byte_string if TYPE_CHECKING: from spl.token.async_client import AsyncToken # noqa: F401 diff --git a/tests/integration/test_async_token_client.py b/tests/integration/test_async_token_client.py index 41dceb95..0bcdee01 100644 --- a/tests/integration/test_async_token_client.py +++ b/tests/integration/test_async_token_client.py @@ -5,10 +5,11 @@ import spl.token._layouts as layouts from solana.publickey import PublicKey from solana.rpc.types import TxOpts +from solana.utils.helpers import decode_byte_string from spl.token.async_client import AsyncToken from spl.token.constants import ASSOCIATED_TOKEN_PROGRAM_ID, TOKEN_PROGRAM_ID -from .utils import AIRDROP_AMOUNT, aconfirm_transaction, assert_valid_response, decode_byte_string +from .utils import AIRDROP_AMOUNT, aconfirm_transaction, assert_valid_response @pytest.mark.integration diff --git a/tests/integration/test_token_client.py b/tests/integration/test_token_client.py index 0fa5e78e..5720812b 100644 --- a/tests/integration/test_token_client.py +++ b/tests/integration/test_token_client.py @@ -5,10 +5,11 @@ import spl.token._layouts as layouts from solana.publickey import PublicKey from solana.rpc.types import TxOpts +from solana.utils.helpers import decode_byte_string from spl.token.client import Token from spl.token.constants import ASSOCIATED_TOKEN_PROGRAM_ID, TOKEN_PROGRAM_ID -from .utils import AIRDROP_AMOUNT, assert_valid_response, confirm_transaction, decode_byte_string +from .utils import AIRDROP_AMOUNT, assert_valid_response, confirm_transaction @pytest.mark.integration diff --git a/tests/integration/utils.py b/tests/integration/utils.py index d7b1aeb5..230ceae2 100644 --- a/tests/integration/utils.py +++ b/tests/integration/utils.py @@ -1,11 +1,8 @@ """Integration test utils.""" import asyncio import time -from base64 import b64decode from typing import Any, Dict -from base58 import b58decode - from solana.rpc.api import Client from solana.rpc.async_api import AsyncClient from solana.rpc.types import RPCResponse @@ -99,14 +96,3 @@ async def aconfirm_transaction(client: AsyncClient, tx_sig: str) -> RPCResponse: if not resp["result"]: raise RuntimeError("could not confirm transaction: ", tx_sig) return resp - - -def decode_byte_string(byte_string: str, encoding: str = "base64") -> bytes: - """Decode a encoded string from an RPC Response.""" - b_str = str.encode(byte_string) - if encoding == "base64": - return b64decode(b_str) - if encoding == "base58": - return b58decode(b_str) - - raise NotImplementedError(f"{encoding} decoding not currently supported.")