move decode_byte_string out of tests

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.")

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

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

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

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.")