Merge pull request #46 from algorandfoundation/feat/replace-box-ref

feat: make BoxRef methods directly accessible on Box class

Author: boblat

examples/proof_of_attendance/contract.py

@@ -45,11 +45,11 @@ def confirm_attendance_with_box_ref(self) -> None:
         minted_asset = self._mint_poa(algopy.Txn.sender)
         self.total_attendees += 1
 
-        box_ref = algopy.BoxRef(key=algopy.Txn.sender.bytes)
+        box_ref = algopy.Box(algopy.Bytes, key=algopy.Txn.sender.bytes)
         has_claimed = bool(box_ref)
         assert not has_claimed, "Already claimed POA"
 
-        box_ref.put(algopy.op.itob(minted_asset.id))
+        box_ref.value = algopy.op.itob(minted_asset.id)
 
     @algopy.arc4.abimethod()
     def confirm_attendance_with_box_map(self) -> None:
@@ -78,7 +78,7 @@ def get_poa_id_with_box(self) -> algopy.UInt64:
 
     @algopy.arc4.abimethod(readonly=True)
     def get_poa_id_with_box_ref(self) -> algopy.UInt64:
-        box_ref = algopy.BoxRef(key=algopy.Txn.sender.bytes)
+        box_ref = algopy.Box(algopy.Bytes, key=algopy.Txn.sender.bytes)
         poa_id, exists = box_ref.maybe()
         assert exists, "POA not found"
         return algopy.op.btoi(poa_id)
@@ -130,7 +130,7 @@ def claim_poa_with_box(self, opt_in_txn: algopy.gtxn.AssetTransferTransaction) -
 
     @algopy.arc4.abimethod()
     def claim_poa_with_box_ref(self, opt_in_txn: algopy.gtxn.AssetTransferTransaction) -> None:
-        box_ref = algopy.BoxRef(key=algopy.Txn.sender.bytes)
+        box_ref = algopy.Box(algopy.Bytes, key=algopy.Txn.sender.bytes)
         poa_id, exists = box_ref.maybe()
         assert exists, "POA not found, attendance validation failed!"
         assert opt_in_txn.xfer_asset.id == algopy.op.btoi(poa_id), "POA ID mismatch"

pyproject.toml

@@ -32,7 +32,7 @@ dependencies = [
     "coincurve>=19.0.1",
     # TODO: uncomment below and remove direct git reference once puya 5.0 is released
     # "algorand-python>=3",
-    "algorand-python@git+https://github.com/algorandfoundation/[email protected].7#subdirectory=stubs",
+    "algorand-python@git+https://github.com/algorandfoundation/[email protected].10#subdirectory=stubs",
 ]
 
 [project.urls]
@@ -54,7 +54,7 @@ python = "3.12"
 dependencies = [
     # TODO: uncomment below and remove direct git reference once puya 5.0 is released
     # "puyapy>=5",
-    "puyapy@git+https://github.com/algorandfoundation/[email protected].7",
+    "puyapy@git+https://github.com/algorandfoundation/[email protected].10",
     "pytest>=7.4",
     "pytest-mock>=3.10.0",
     "pytest-xdist[psutil]>=3.3",
@@ -138,7 +138,7 @@ dependencies = [
     "algokit-utils>=3.0.0",
     # TODO: uncomment below and remove direct git reference once puya 5.0 is released
     # "puyapy>=5",
-    "puyapy@git+https://github.com/algorandfoundation/[email protected].7",
+    "puyapy@git+https://github.com/algorandfoundation/[email protected].10",
 ]
 
 [tool.hatch.envs.test.scripts]
@@ -191,7 +191,7 @@ post-install-commands = [
 dependencies = [
     # TODO: uncomment below and remove direct git reference once puya 5.0 is released
     # "algorand-python>=3",
-    "algorand-python@git+https://github.com/algorandfoundation/[email protected].7#subdirectory=stubs",
+    "algorand-python@git+https://github.com/algorandfoundation/[email protected].10#subdirectory=stubs",
     "pytest>=7.4",
     "pytest-mock>=3.10.0",
     "pytest-xdist[psutil]>=3.3",

src/_algopy_testing/arc4.py

@@ -140,7 +140,7 @@ class _ABIEncoded(BytesBacked):
     def from_bytes(cls, value: algopy.Bytes | bytes, /) -> typing.Self:
         """Construct an instance from the underlying bytes (no validation)"""
         instance = cls()
-        instance._value = as_bytes(value)
+        instance._value = value.value if isinstance(value, Bytes) else value
         return instance
 
     @classmethod
@@ -556,11 +556,11 @@ class Bool(_ABIEncoded):
     _value: bytes
 
     # True value is encoded as having a 1 on the most significant bit (0x80 = 128)
-    _true_int_value = 128
-    _false_int_value = 0
+    _true_byte_value = int_to_bytes(128, 1)
+    _false_byte_value = int_to_bytes(0, 1)
 
     def __init__(self, value: bool = False, /) -> None:  # noqa: FBT001, FBT002
-        self._value = int_to_bytes(self._true_int_value if value else self._false_int_value, 1)
+        self._value = self._true_byte_value if value else self._false_byte_value
 
     def __eq__(self, other: object) -> bool:
         try:
@@ -576,8 +576,7 @@ def __bool__(self) -> bool:
     @property
     def native(self) -> bool:
         """Return the bool representation of the value after ARC4 decoding."""
-        int_value = int.from_bytes(self._value)
-        return int_value == self._true_int_value
+        return self._value == self._true_byte_value
 
     def __str__(self) -> str:
         return f"{self.native}"
@@ -669,7 +668,8 @@ def __init__(self, *_items: _TArrayItem):
                     f"item must be of type {self._type_info.item_type!r}, not {item._type_info!r}"
                 )
 
-        self._value = _encode(items)
+        item_list = list(items)
+        self._value = _encode(item_list)
 
     def __iter__(self) -> Iterator[_TArrayItem]:
         # """Returns an iterator for the items in the array"""
@@ -854,7 +854,8 @@ def __init__(self, *_items: _TArrayItem):
                 raise TypeError(
                     f"item must be of type {self._type_info.item_type!r}, not {item._type_info!r}"
                 )
-        self._value = self._encode_with_length(items)
+        item_list = list(items)
+        self._value = self._encode_with_length(item_list)
 
     def __iter__(self) -> typing.Iterator[_TArrayItem]:
         """Returns an iterator for the items in the array."""
@@ -1294,6 +1295,9 @@ def _find_bool(
     is_looking_forward = delta > 0
     is_looking_backward = delta < 0
     values_length = len(values) if isinstance(values, tuple | list) else values.length.value
+    if isinstance(values, (StaticArray | DynamicArray | list)):
+        return 0 if is_looking_backward else values_length - index - 1
+
     while True:
         curr = index + delta * until
         is_curr_at_end = curr == values_length - 1
@@ -1311,12 +1315,16 @@ def _find_bool(
     return until
 
 
-def _find_bool_types(values: typing.Sequence[_TypeInfo], index: int, delta: int) -> int:
+def _find_bool_types(
+    values: typing.Sequence[_TypeInfo], index: int, delta: int, *, is_homogeneous: bool = False
+) -> int:
     """Helper function to find consecutive booleans from current index in a tuple."""
     until = 0
     is_looking_forward = delta > 0
     is_looking_backward = delta < 0
     values_length = len(values)
+    if is_homogeneous:
+        return 0 if is_looking_backward else values_length - index - 1
     while True:
         curr = index + delta * until
         is_curr_at_end = curr == values_length - 1
@@ -1438,6 +1446,7 @@ def _encode(  # noqa: PLR0912
                     raise ValueError(
                         "expected before index should have number of bool mod 8 equal 0"
                     )
+
                 after = min(7, after)
                 consecutive_bool_list = [values[i] for i in range(i, i + after + 1)]
                 compressed_int = _compress_multiple_bool(consecutive_bool_list)
@@ -1467,14 +1476,16 @@ def _encode(  # noqa: PLR0912
     return values_length_bytes + b"".join(heads) + b"".join(tails)
 
 
-def _decode_tuple_items(  # noqa: PLR0912
+def _decode_tuple_items(  # noqa: PLR0912, PLR0915
     value: bytes, child_types: list[_TypeInfo]
 ) -> list[typing.Any]:
     dynamic_segments: list[list[int]] = []  # Store the start and end of a dynamic element
     value_partitions: list[bytes] = []
     i = 0
     array_index = 0
 
+    is_homogeneous_child_types = len(set(child_types)) == 1
+
     while i < len(child_types):
         child_type = child_types[i]
         if child_type.is_dynamic:
@@ -1489,8 +1500,12 @@ def _decode_tuple_items(  # noqa: PLR0912
             value_partitions.append(b"")
             array_index += _ABI_LENGTH_SIZE
         elif isinstance(child_type, _BoolTypeInfo):
-            before = _find_bool_types(child_types, i, -1)
-            after = _find_bool_types(child_types, i, 1)
+            before = _find_bool_types(
+                child_types, index=i, delta=-1, is_homogeneous=is_homogeneous_child_types
+            )
+            after = _find_bool_types(
+                child_types, index=i, delta=1, is_homogeneous=is_homogeneous_child_types
+            )
 
             if before % 8 != 0:
                 raise ValueError("expected before index should have number of bool mod 8 equal 0")

src/_algopy_testing/primitives/array.py

@@ -14,6 +14,9 @@
     parameterize_type,
 )
 
+if typing.TYPE_CHECKING:
+    from _typeshed import DataclassInstance
+
 _TArrayItem = typing.TypeVar("_TArrayItem")
 _TArrayLength = typing.TypeVar("_TArrayLength", bound=int)
 _T = typing.TypeVar("_T")
@@ -246,7 +249,7 @@ def _from_iter(self, items: Iterable[_TArrayItem]) -> "FixedArray[_TArrayItem, _
         return typ(items)
 
     def serialize(self) -> bytes:
-        return serialize_to_bytes(self)
+        return self._value
 
     @classmethod
     def from_bytes(cls, value: bytes, /) -> typing.Self:
@@ -509,10 +512,15 @@ def from_bytes(cls, value: bytes, /) -> typing.Self:
 class Struct(Serializable, MutableBytes):
     """Base class for Struct types."""
 
+    _field_names: typing.ClassVar[list[str]]
+
     def __init_subclass__(cls, *args: typing.Any, **kwargs: dict[str, typing.Any]) -> None:
         # make implementation not frozen, so we can conditionally control behaviour
         dataclasses.dataclass(cls, *args, **{**kwargs, "frozen": False})
         frozen = kwargs.get("frozen", False)
+        cls._field_names = [
+            f.name for f in dataclasses.fields(typing.cast("type[DataclassInstance]", cls))
+        ]
         assert isinstance(frozen, bool)
 
     def __post_init__(self) -> None:
@@ -524,6 +532,13 @@ def __getattribute__(self, name: str) -> typing.Any:
         value = super().__getattribute__(name)
         return add_mutable_callback(lambda _: self._update_backing_value(), value)
 
+    def __setattr__(self, key: str, value: typing.Any) -> None:
+        super().__setattr__(key, value)
+        # don't update backing value until base class has been init'd
+
+        if hasattr(self, "_on_mutate") and key in self._field_names:
+            self._update_backing_value()
+
     def copy(self) -> typing.Self:
         return self.__class__.from_bytes(self.serialize())
 

src/_algopy_testing/primitives/bytes.py

@@ -12,7 +12,7 @@
 
 from _algopy_testing.constants import MAX_BYTES_SIZE
 from _algopy_testing.primitives.uint64 import UInt64
-from _algopy_testing.utils import as_bytes, as_int64, check_type
+from _algopy_testing.utils import as_bytes, check_type
 
 # TypeError, ValueError are used for operations that are compile time errors
 # ArithmeticError and subclasses are used for operations that would fail during AVM execution
@@ -74,7 +74,8 @@ def __getitem__(
         if isinstance(index, slice):
             return Bytes(self.value[index])
         else:
-            int_index = as_int64(index)
+            int_index = index.value if isinstance(index, UInt64) else index
+            int_index = len(self.value) + int_index if int_index < 0 else int_index
             # my_bytes[0:1] => b'j' whereas my_bytes[0] => 106
             return Bytes(self.value[slice(int_index, int_index + 1)])
 

src/_algopy_testing/serialize.py

@@ -174,7 +174,7 @@ class TempStruct(arc4.Struct):
 
 
 def serialize_to_bytes(value: object) -> bytes:
-    return native_to_arc4(value).bytes.value
+    return native_to_arc4(value)._value
 
 
 def type_of(value: object) -> type:

src/_algopy_testing/state/box.py

@@ -3,6 +3,8 @@
 import typing
 import warnings
 
+from typing_extensions import deprecated
+
 import _algopy_testing
 from _algopy_testing.constants import MAX_BOX_SIZE
 from _algopy_testing.context_helpers import lazy_context
@@ -100,6 +102,61 @@ def value(self, value: _TValue) -> None:
     def value(self) -> None:
         lazy_context.ledger.delete_box(self.app_id, self.key)
 
+    @property
+    @deprecated("Box methods previously accessed via `.ref` are now directly available")
+    def ref(self) -> BoxRef:
+        return BoxRef(key=self.key)
+
+    def extract(
+        self, start_index: algopy.UInt64 | int, length: algopy.UInt64 | int
+    ) -> algopy.Bytes:
+        """Extract a slice of bytes from the box.
+
+        Fails if the box does not exist, or if `start_index + length > len(box)`
+
+        :arg start_index: The offset to start extracting bytes from
+        :arg length: The number of bytes to extract
+        :return: The extracted bytes
+        """
+        return _BoxRef(key=self.key).extract(start_index, length)
+
+    def resize(self, new_size: algopy.UInt64 | int) -> None:
+        """Resizes the box the specified `new_size`. Truncating existing data if the new
+        value is shorter or padding with zero bytes if it is longer.
+
+        :arg new_size: The new size of the box
+        """
+        return _BoxRef(key=self.key).resize(new_size)
+
+    def replace(self, start_index: algopy.UInt64 | int, value: algopy.Bytes | bytes) -> None:
+        """Write `value` to the box starting at `start_index`. Fails if the box does not
+        exist, or if `start_index + len(value) > len(box)`
+
+        :arg start_index: The offset to start writing bytes from
+        :arg value: The bytes to be written
+        """
+        return _BoxRef(key=self.key).replace(start_index, value)
+
+    def splice(
+        self,
+        start_index: algopy.UInt64 | int,
+        length: algopy.UInt64 | int,
+        value: algopy.Bytes | bytes,
+    ) -> None:
+        """Set box to contain its previous bytes up to index `start_index`, followed by
+        `bytes`, followed by the original bytes of the box that began at index
+        `start_index + length`
+
+        **Important: This op does not resize the box**
+        If the new value is longer than the box size, it will be truncated.
+        If the new value is shorter than the box size, it will be padded with zero bytes
+
+        :arg start_index: The index to start inserting `value`
+        :arg length: The number of bytes after `start_index` to omit from the new value
+        :arg value: The `value` to be inserted.
+        """
+        return _BoxRef(key=self.key).splice(start_index, length, value)
+
     def get(self, *, default: _TValue) -> _TValue:
         box_content, box_exists = self.maybe()
         return default if not box_exists else box_content
@@ -117,7 +174,7 @@ def length(self) -> algopy.UInt64:
         return _algopy_testing.UInt64(len(lazy_context.ledger.get_box(self.app_id, self.key)))
 
 
-class BoxRef:
+class _BoxRef:
     """BoxRef abstracts the reading and writing of boxes containing raw binary data.
 
     The size is configured manually, and can be set to values larger than what the AVM
@@ -186,14 +243,15 @@ def resize(self, new_size: algopy.UInt64 | int) -> None:
         lazy_context.ledger.set_box(self.app_id, self.key, updated_content)
 
     def replace(self, start_index: algopy.UInt64 | int, value: algopy.Bytes | bytes) -> None:
+        replace_content = value.value if isinstance(value, _algopy_testing.Bytes) else value
         box_content, box_exists = self._maybe()
         if not box_exists:
             raise RuntimeError("Box has not been created")
         start = int(start_index)
-        length = len(value)
+        length = len(replace_content)
         if (start + length) > len(box_content):
             raise ValueError("Replacement content exceeds box size")
-        updated_content = box_content[:start] + value + box_content[start + length :]
+        updated_content = box_content[:start] + replace_content + box_content[start + length :]
         lazy_context.ledger.set_box(self.app_id, self.key, updated_content)
 
     def splice(
@@ -260,6 +318,11 @@ def length(self) -> algopy.UInt64:
         return _algopy_testing.UInt64(len(box_content))
 
 
+@deprecated("Methods in BoxRef are now directly available on Box")
+class BoxRef(_BoxRef):
+    pass
+
+
 class BoxMap(typing.Generic[_TKey, _TValue]):
     """BoxMap abstracts the reading and writing of a set of boxes using a common key and
     content type.
@@ -338,3 +401,8 @@ def length(self, key: _TKey) -> algopy.UInt64:
 
     def _full_key(self, key: _TKey) -> algopy.Bytes:
         return self.key_prefix + cast_to_bytes(key)
+
+    def box(self, key: _TKey) -> Box[_TValue]:
+        """Returns a Box holding the box value at key."""
+        key_bytes = self._full_key(key)
+        return Box(self._value_type, key=key_bytes)