Skip to content

refactor: further aligning composer class; initial batch of resource population tests #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Merged
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
4 changes: 4 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,10 @@ untyped_calls_exclude = [
module = ["algosdk", "algosdk.*"]
disallow_untyped_calls = false

[[tool.mypy.overrides]]
module = ["tests.transactions.test_transaction_composer"]
disable_error_code = ["call-overload", "union-attr"]

[tool.semantic_release]
version_toml = "pyproject.toml:tool.poetry.version"
remove_dist = false
Expand Down
141 changes: 88 additions & 53 deletions src/algokit_utils/accounts/account_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,16 @@
from typing import Any

from algosdk import mnemonic
from algosdk.atomic_transaction_composer import AccountTransactionSigner, TransactionSigner
from algosdk.atomic_transaction_composer import LogicSigTransactionSigner, TransactionSigner
from algosdk.mnemonic import to_private_key
from algosdk.transaction import SuggestedParams
from algosdk.transaction import LogicSigAccount, SuggestedParams
from typing_extensions import Self

from algokit_utils.accounts.kmd_account_manager import KmdAccountManager
from algokit_utils.clients.client_manager import ClientManager
from algokit_utils.clients.dispenser_api_client import DispenserAssetName, TestNetDispenserApiClient
from algokit_utils.config import config
from algokit_utils.models.account import DISPENSER_ACCOUNT_NAME, Account
from algokit_utils.models.account import DISPENSER_ACCOUNT_NAME, Account, MultiSigAccount, MultisigMetadata
from algokit_utils.models.amount import AlgoAmount
from algokit_utils.transactions.transaction_composer import (
PaymentParams,
Expand Down Expand Up @@ -52,7 +52,7 @@ def __init__(self, client_manager: ClientManager):
"""
self._client_manager = client_manager
self._kmd_account_manager = KmdAccountManager(client_manager)
self._accounts = dict[str, Account]()
self._signers = dict[str, TransactionSigner]()
self._default_signer: TransactionSigner | None = None

def set_default_signer(self, signer: TransactionSigner) -> Self:
Expand All @@ -73,17 +73,26 @@ def set_signer(self, sender: str, signer: TransactionSigner) -> Self:
:param signer: The signer to sign transactions with for the given sender
:return: The AccountCreator instance for method chaining
"""
if isinstance(signer, AccountTransactionSigner):
self._accounts[sender] = Account(private_key=signer.private_key)
self._signers[sender] = signer
return self

def get_account(self, sender: str) -> Account:
account = self._accounts.get(sender)
account = self._signers.get(sender)
if not account:
raise ValueError(f"No account found for address {sender}")
if not isinstance(account, Account):
raise ValueError(f"Account {sender} is not a regular account")
return account

def get_signer(self, sender: str | Account) -> TransactionSigner:
def get_logic_sig_account(self, sender: str) -> LogicSigAccount:
account = self._signers.get(sender)
if not account:
raise ValueError(f"No account found for address {sender}")
if not isinstance(account, LogicSigAccount):
raise ValueError(f"Account {sender} is not a logic sig account")
return account

def get_signer(self, sender: str | Account | LogicSigAccount) -> TransactionSigner:
"""
Returns the `TransactionSigner` for the given sender address.

Expand All @@ -92,8 +101,7 @@ def get_signer(self, sender: str | Account) -> TransactionSigner:
:param sender: The sender address
:return: The `TransactionSigner` or throws an error if not found
"""
account = self._accounts.get(self._get_address(sender))
signer = account.signer if account else self._default_signer
signer = self._signers.get(self._get_address(sender))
if not signer:
raise ValueError(f"No signer found for address {sender}")
return signer
Expand All @@ -109,29 +117,48 @@ def get_information(self, sender: str | Account) -> dict[str, Any]:
assert isinstance(info, dict)
return info

def from_mnemonic(self, mnemonic: str) -> Account:
private_key = to_private_key(mnemonic)
def _register_account(self, private_key: str) -> Account:
"""Helper method to create and register an account with its signer.

Args:
private_key: The private key for the account

Returns:
The registered Account instance
"""
account = Account(private_key=private_key)
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=private_key))
self._signers[account.address] = account.signer
return account

def _register_logic_sig(self, program: bytes, args: list[bytes] | None = None) -> LogicSigAccount:
logic_sig = LogicSigAccount(program, args)
self._signers[logic_sig.address()] = LogicSigTransactionSigner(logic_sig)
return logic_sig

def _register_multi_sig(
self, version: int, threshold: int, addrs: list[str], signing_accounts: list[Account]
) -> MultiSigAccount:
msig_account = MultiSigAccount(
MultisigMetadata(version=version, threshold=threshold, addresses=addrs),
signing_accounts,
)
self._signers[str(msig_account.address)] = msig_account.signer
return msig_account

def from_mnemonic(self, mnemonic: str) -> Account:
private_key = to_private_key(mnemonic)
return self._register_account(private_key)

def from_environment(self, name: str, fund_with: AlgoAmount | None = None) -> Account:
account_mnemonic = os.getenv(f"{name.upper()}_MNEMONIC")

if account_mnemonic:
private_key = mnemonic.to_private_key(account_mnemonic)
account = Account(private_key=private_key)
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=private_key))
return account
return self._register_account(private_key)

if self._client_manager.is_local_net():
kmd_account = self._kmd_account_manager.get_or_create_wallet_account(name, fund_with)
account = Account(private_key=kmd_account.private_key)
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=kmd_account.private_key))
return account
return self._register_account(kmd_account.private_key)

raise ValueError(f"Missing environment variable {name.upper()}_MNEMONIC when looking for account {name}")

Expand All @@ -142,14 +169,38 @@ def from_kmd(
if not kmd_account:
raise ValueError(f"Unable to find KMD account {name}{' with predicate' if predicate else ''}")

account = Account(private_key=kmd_account.private_key)
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=kmd_account.private_key))
return account
return self._register_account(kmd_account.private_key)

def logic_sig(self, program: bytes, args: list[bytes] | None = None) -> LogicSigAccount:
return self._register_logic_sig(program, args)

def multi_sig(
self, version: int, threshold: int, addrs: list[str], signing_accounts: list[Account]
) -> MultiSigAccount:
return self._register_multi_sig(version, threshold, addrs, signing_accounts)

def random(self) -> Account:
"""
Tracks and returns a new, random Algorand account.

:return: The account
"""
account = Account.new_account()
return self._register_account(account.private_key)

def localnet_dispenser(self) -> Account:
kmd_account = self._kmd_account_manager.get_localnet_dispenser_account()
return self._register_account(kmd_account.private_key)

def dispenser_from_environment(self) -> Account:
name = os.getenv(f"{DISPENSER_ACCOUNT_NAME}_MNEMONIC")
if name:
return self.from_environment(DISPENSER_ACCOUNT_NAME)
return self.localnet_dispenser()

def rekeyed(self, sender: Account | str, account: Account) -> Account:
sender_address = sender.address if isinstance(sender, Account) else sender
self._accounts[sender_address] = account
self._signers[sender_address] = account.signer
return Account(address=sender_address, private_key=account.private_key)

def rekey_account( # noqa: PLR0913
Expand Down Expand Up @@ -223,30 +274,6 @@ def rekey_account( # noqa: PLR0913

return result

def random(self) -> Account:
"""
Tracks and returns a new, random Algorand account.

:return: The account
"""
account = Account.new_account()
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=account.private_key))
return account

def localnet_dispenser(self) -> Account:
kmd_account = self._kmd_account_manager.get_localnet_dispenser_account()
account = Account(private_key=kmd_account.private_key)
self._accounts[account.address] = account
self.set_signer(account.address, AccountTransactionSigner(private_key=kmd_account.private_key))
return account

def dispenser_from_environment(self) -> Account:
name = os.getenv(f"{DISPENSER_ACCOUNT_NAME}_MNEMONIC")
if name:
return self.from_environment(DISPENSER_ACCOUNT_NAME)
return self.localnet_dispenser()

def ensure_funded( # noqa: PLR0913
self,
account_to_fund: str | Account,
Expand Down Expand Up @@ -448,8 +475,16 @@ def ensure_funded_from_testnet_dispenser_api(
amount_funded=AlgoAmount.from_micro_algo(result.amount),
)

def _get_address(self, sender: str | Account) -> str:
return sender.address if isinstance(sender, Account) else sender
def _get_address(self, sender: str | Account | LogicSigAccount) -> str:
match sender:
case Account():
return sender.address
case LogicSigAccount():
return sender.address()
case str():
return sender
case _:
raise ValueError(f"Unknown sender type: {type(sender)}")

def _get_composer(self, get_suggested_params: Callable[[], SuggestedParams] | None = None) -> TransactionComposer:
if get_suggested_params is None:
Expand Down
13 changes: 7 additions & 6 deletions src/algokit_utils/applications/app_deployer.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,12 +268,13 @@ def deploy(self, deployment: AppDeployParams) -> AppDeployResult:
clear_program=clear_program,
)

return AppDeployResult(
**existing_app.__dict__,
operation_performed=OperationPerformed.Nothing,
app_id=existing_app.app_id,
app_address=existing_app.app_address,
)
existing_app_dict = existing_app.__dict__
existing_app_dict["operation_performed"] = OperationPerformed.Nothing
existing_app_dict["app_id"] = existing_app.app_id
existing_app_dict["app_address"] = existing_app.app_address

logger.debug("No detected changes in app, nothing to do.", suppress_log=deployment.suppress_log)
return AppDeployResult(**existing_app_dict)

def _create_app(
self,
Expand Down
12 changes: 10 additions & 2 deletions src/algokit_utils/applications/app_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,14 +54,22 @@ class DataTypeFlag(IntEnum):


class BoxReference(AlgosdkBoxReference):
def __init__(self, app_id: int, name: bytes):
super().__init__(app_index=app_id, name=name)
def __init__(self, app_id: int, name: bytes | str):
super().__init__(app_index=app_id, name=self._b64_decode(name))

def __eq__(self, other: object) -> bool:
if isinstance(other, (BoxReference | AlgosdkBoxReference)):
return self.app_index == other.app_index and self.name == other.name
return False

def _b64_decode(self, value: str | bytes) -> bytes:
if isinstance(value, str):
try:
return base64.b64decode(value)
except Exception:
return value.encode("utf-8")
return value


def _is_valid_token_character(char: str) -> bool:
return char.isalnum() or char == "_"
Expand Down
12 changes: 7 additions & 5 deletions src/algokit_utils/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,11 +127,12 @@ def with_debug(self, func: Callable[[], str | None]) -> None:
def configure(
self,
*,
debug: bool,
debug: bool | None = None,
project_root: Path | None = None,
trace_all: bool = False,
trace_buffer_size_mb: float = 256,
max_search_depth: int = 10,
populate_app_call_resources: bool = False,
) -> None:
"""
Configures various settings for the application.
Expand All @@ -153,16 +154,17 @@ def configure(
None
"""

self._debug = debug

if project_root:
if debug is not None:
self._debug = debug
if project_root is not None:
self._project_root = project_root.resolve(strict=True)
elif debug and ALGOKIT_PROJECT_ROOT:
elif debug is not None and ALGOKIT_PROJECT_ROOT:
self._project_root = Path(ALGOKIT_PROJECT_ROOT).resolve(strict=True)

self._trace_all = trace_all
self._trace_buffer_size_mb = trace_buffer_size_mb
self._max_search_depth = max_search_depth
self._populate_app_call_resources = populate_app_call_resources


config = UpdatableConfig()
Loading
Loading