Skip to content

Commit

Permalink
refactor: wip adding txn_group_for method
Browse files Browse the repository at this point in the history
  • Loading branch information
aorumbayev committed Aug 12, 2024
1 parent 2cda537 commit 0cb6f32
Show file tree
Hide file tree
Showing 4 changed files with 190 additions and 2 deletions.
77 changes: 76 additions & 1 deletion src/algopy_testing/_context_helpers/txn_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,15 @@
import contextlib
import typing
from collections import defaultdict
from dataclasses import dataclass
from inspect import signature

import algosdk

from algopy_testing.constants import ARC4_RETURN_PREFIX
from algopy_testing.decorators.arc4 import _extract_group_txns, _generate_arc4_signature_from_fn
from algopy_testing.enums import OnCompleteAction
from algopy_testing.models.contract import Contract
from algopy_testing.primitives.bytes import Bytes
from algopy_testing.utils import convert_native_to_stack, get_new_scratch_space

Expand All @@ -17,7 +21,7 @@
import algopy

from algopy_testing._itxn_loader import InnerTransactionResultType
from algopy_testing.models.txn_fields import TransactionFields
from algopy_testing.models.txn_fields import ApplicationCallFields, TransactionFields


from algopy_testing import gtxn
Expand All @@ -27,6 +31,19 @@
from algopy_testing.primitives import UInt64


@dataclass
class PreparedAppCall:
txns: list[algopy.gtxn.TransactionBase]
app_txn: algopy.gtxn.ApplicationCallTransaction
method: typing.Callable[..., typing.Any]
args: tuple
kwargs: dict

def submit(self) -> object:
# This method will be called to execute the prepared call
return self.method(*self.args, **self.kwargs)


class TransactionContext:
def __init__(self) -> None:
self._groups: list[TransactionGroup] = []
Expand Down Expand Up @@ -59,6 +76,64 @@ def _maybe_implicit_txn_group(
with ctx:
yield

def txn_group_for(
self, method: typing.Callable[..., typing.Any], *args: typing.Any, **kwargs: typing.Any
) -> PreparedAppCall:
"""Prepare an application call transaction group for a contract method
without executing it.
:param method: The decorated contract method (baremethod or
abimethod).
:param args: Positional arguments for the method.
:param kwargs: Keyword arguments for the method.
:return: A PreparedAppCall object containing the transaction
group and method info.
"""

import algopy

if not hasattr(method, "__wrapped__"):
raise ValueError("The provided method must be decorated with baremethod or abimethod")

contract = method.__self__
if not isinstance(contract, Contract):
raise TypeError("The method must be bound to a Contract instance")

# Extract method signature
sig = signature(method)
bound_args = sig.bind(contract, *args, **kwargs)
bound_args.apply_defaults()

# Prepare transaction fields
txn_fields: ApplicationCallFields = {}

if hasattr(method, "is_create"):
txn_fields["on_completion"] = (
OnCompleteAction.NoOp
if not method.is_create # type: ignore[attr-defined]
else OnCompleteAction.OptIn
)

# Handle ABI methods
if hasattr(method, "__name__"):
arc4_name = getattr(method, "__arc4_name__", method.__name__)
arc4_signature = _generate_arc4_signature_from_fn(method.__wrapped__, arc4_name) # type: ignore[attr-defined]
txns = _extract_group_txns(
self,
contract=contract,
arc4_signature=arc4_signature,
args=bound_args.args[1:], # Exclude 'self' argument
)
app_txn = next(
txn for txn in txns if isinstance(txn, algopy.gtxn.ApplicationCallTransaction)
)
return PreparedAppCall(txns, app_txn, method, args, kwargs)

# Handle bare methods
txn_fields["app_id"] = contract.__app_id__
app_txn = self.any.txn.application_call(**txn_fields)
return PreparedAppCall([app_txn], app_txn, method, args, kwargs)

@contextlib.contextmanager
def scoped_execution(
self,
Expand Down
5 changes: 4 additions & 1 deletion src/algopy_testing/_value_generators/avm.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,12 @@ def uint64(self, min_value: int = 0, max_value: int = MAX_UINT64) -> algopy.UInt
raise ValueError("max_value must be less than or equal to MAX_UINT64")
if min_value > max_value:
raise ValueError("min_value must be less than or equal to max_value")
if min_value < 0 or max_value < 0:
raise ValueError("min_value and max_value must be greater than or equal to 0")

return algopy_testing.UInt64(generate_random_int(min_value, max_value))

def bytes(self, length: int = MAX_BYTES_SIZE) -> algopy.Bytes:
def bytes(self, length: int | None = None) -> algopy.Bytes:
"""Generate a random byte sequence of a specified length.
:param length: Length of the byte sequence. Defaults to
Expand All @@ -53,6 +55,7 @@ def bytes(self, length: int = MAX_BYTES_SIZE) -> algopy.Bytes:
:returns: The randomly generated byte sequence.
:rtype: algopy.Bytes
"""
length = length or MAX_BYTES_SIZE
return algopy_testing.Bytes(secrets.token_bytes(length))

def string(self, length: int = MAX_BYTES_SIZE) -> algopy.String:
Expand Down
Empty file.
110 changes: 110 additions & 0 deletions tests/value_generators/test_avm.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
import algopy
import algosdk
import pytest
from algopy_testing import algopy_testing_context
from algopy_testing.constants import MAX_BYTES_SIZE, MAX_UINT64
from algopy_testing.context import AlgopyTestContext
from algopy_testing.primitives.bytes import Bytes
from algopy_testing.primitives.string import String


@pytest.fixture()
def context() -> AlgopyTestContext:
with algopy_testing_context() as ctx:
yield ctx


def assert_value_in_range(value: int | object, min_val: int, max_val: int) -> None:
assert min_val <= value <= max_val


def assert_length(value: bytes | str | String | Bytes, expected_length: int) -> None:
if isinstance(value, bytes | Bytes):
assert len(value) == expected_length
else:
assert len(str(value)) == expected_length


@pytest.mark.parametrize(
("method", "type_", "min_val", "max_val"),
[
("uint64", algopy.UInt64, 0, MAX_UINT64),
],
)
def test_avm_uint64_generator(
context: AlgopyTestContext, method: str, type_: type, min_val: int, max_val: int
) -> None:
func = getattr(context.any, method)
value = func(min_val, max_val)
assert isinstance(value, type_)
assert_value_in_range(value, min_val, max_val)

with pytest.raises(ValueError, match="max_value must be less than or equal to MAX_UINT64"):
func(max_value=max_val + 1)

with pytest.raises(ValueError, match="min_value must be less than or equal to max_value"):
func(min_value=max_val + 1)

with pytest.raises(
ValueError, match="min_value and max_value must be greater than or equal to 0"
):
func(min_value=-1)


@pytest.mark.parametrize("length", [None, 10, MAX_BYTES_SIZE])
def test_avm_bytes_generator(context: AlgopyTestContext, length: int | None) -> None:
value = context.any.bytes(length) if length else context.any.bytes()
assert isinstance(value, algopy.Bytes)
assert_length(value, length or MAX_BYTES_SIZE)


@pytest.mark.parametrize("length", [None, 10, MAX_BYTES_SIZE])
def test_avm_string_generator(context: AlgopyTestContext, length: int | None) -> None:
value = context.any.string(length) if length else context.any.string()
assert isinstance(value, algopy.String)
assert_length(value, length or MAX_BYTES_SIZE)


def test_avm_account_generator(context: AlgopyTestContext) -> None:
account = context.any.account()
assert isinstance(account, algopy.Account)
assert context.ledger.account_exists(account.public_key)

custom_address = algosdk.account.generate_account()[1]
account = context.any.account(address=custom_address)
assert isinstance(account, algopy.Account)
assert account.public_key == custom_address
assert context.ledger.account_exists(custom_address)

with pytest.raises(ValueError, match="Account with such address already exists"):
context.any.account(address=custom_address)


def test_avm_asset_generator(context: AlgopyTestContext) -> None:
asset = context.any.asset()
assert isinstance(asset, algopy.Asset)
assert context.ledger.asset_exists(int(asset.id))

custom_id = 1000
asset = context.any.asset(asset_id=custom_id)
assert isinstance(asset, algopy.Asset)
assert int(asset.id) == custom_id
assert context.ledger.asset_exists(custom_id)

with pytest.raises(ValueError, match="Asset with such ID already exists"):
context.any.asset(asset_id=custom_id)


def test_avm_application_generator(context: AlgopyTestContext) -> None:
app = context.any.application()
assert isinstance(app, algopy.Application)
assert context.ledger.app_exists(int(app.id))

custom_id = 1000
app = context.any.application(id=custom_id)
assert isinstance(app, algopy.Application)
assert int(app.id) == custom_id
assert context.ledger.app_exists(custom_id)

with pytest.raises(ValueError, match="Application id .* has already been configured"):
context.any.application(id=custom_id)

0 comments on commit 0cb6f32

Please sign in to comment.