Add empty() growable-buffer initializer for Python encoders

asn1python/src/asn1python/bitstream.py

@@ -26,23 +26,44 @@ def position_invariant(bit_position: int, byte_position: int, buf_length: int) -
         return (bit_position >= 0 and bit_position < NO_OF_BITS_IN_BYTE and
                 byte_position >= 0 and ((byte_position < buf_length) or (bit_position == 0 and byte_position == buf_length)))
 
-    def __init__(self, data: bytearray):
+    def __init__(self, data: bytearray, growable: bool = False):
         """
         Initialize a BitStream.
 
         Args:
             data: Initial data buffer
+            growable: When True, write operations that would exceed the buffer
+                grow it automatically instead of raising. Used by encoders created
+                without a size estimate (see Encoder.empty()). Left False for
+                fixed-size buffers (of_size / from_buffer) and decoders, where
+                exceeding the buffer must stay an error.
         """
         self._buffer = bytearray(data)
         self._current_bit = 0  # Current bit within byte (0-7)
         self._current_byte = 0  # Current byte position (0-based)
+        self._growable = growable
 
     @classmethod
     def from_bitstream(cls, other: 'BitStream') -> 'BitStream':
         """Method to create a BitStream from an existing BitStream. Copies buffer and segments"""
-        result = cls(other._buffer)
+        result = cls(other._buffer, growable=other._growable)
         return result
 
+    def _grow_to_fit(self, additional_bits: int) -> None:
+        """Grow the buffer so that `additional_bits` more bits fit past the current position.
+
+        Doubles capacity (amortised O(1) appends) until the required byte count is
+        reached. No-op if the buffer is already large enough.
+        """
+        needed_bits = self.current_used_bits + additional_bits
+        needed_bytes = (needed_bits + NO_OF_BITS_IN_BYTE - 1) // NO_OF_BITS_IN_BYTE
+        if needed_bytes <= self.buffer_size:
+            return
+        new_size = max(self.buffer_size, 1)
+        while new_size < needed_bytes:
+            new_size *= 2
+        self._buffer.extend(bytearray(new_size - self.buffer_size))
+
     def get_data(self) -> bytearray:
         """Get the used data buffer"""
         used_bytes = self.current_used_bytes
@@ -181,8 +202,11 @@ def __write_bit(self, bit: bool) -> None:
 
     def write_bit(self, bit: bool) -> None:
         if self.remaining_bits < 1:
-            raise BitStreamError("Cannot write beyond end of bitstream")
-
+            if self._growable:
+                self._grow_to_fit(1)
+            else:
+                raise BitStreamError("Cannot write beyond end of bitstream")
+
         return self.__write_bit(bit)
 
     def write_bits(self, value: int, bit_count: int) -> None:
@@ -191,7 +215,10 @@ def write_bits(self, value: int, bit_count: int) -> None:
             raise BitStreamError(f"Bit count {bit_count} out of range [0, 64]")
 
         if self.remaining_bits < bit_count:
-            raise BitStreamError("Cannot write beyond end of bitstream")
+            if self._growable:
+                self._grow_to_fit(bit_count)
+            else:
+                raise BitStreamError("Cannot write beyond end of bitstream")
 
         # Check if value fits in bit_count bits
         if value < 0 or value >= (1 << bit_count):

asn1python/src/asn1python/encoder.py

@@ -1,6 +1,7 @@
 from abc import abstractmethod, ABC
-from typing import Optional, List, Union
+from typing import Optional, List, Self, Union
 
+from .bitstream import BitStream
 from .codec import Codec, EncodeResult, ENCODE_OK, BitStreamError, ERROR_INVALID_VALUE, \
     ERROR_CONSTRAINT_VIOLATION
 
@@ -13,6 +14,18 @@ class Encoder(Codec, ABC):
     def get_decoder(self) -> Decoder:
         pass
 
+    @classmethod
+    def empty(cls) -> Self:
+        """Create an encoder over a fresh, growable buffer.
+
+        This is the go-to way to create an encoder when the output size is not
+        known in advance: the buffer grows automatically as values are written.
+        Use of_size() instead when the required size is known, to pre-allocate
+        exactly. empty() is not meaningful for decoders, which wrap existing
+        bytes via from_buffer().
+        """
+        return cls(BitStream(bytearray(), growable=True))
+
     def encode_integer(self, value: int,
                        min_val: int,
                        max_val: int,

asn1python/src/asn1python/xer_encoder.py

@@ -35,6 +35,21 @@ def of_size(cls, buffer_byte_size: int = 0) -> "XEREncoder":
         """
         return cls()
 
+    @classmethod
+    def empty(cls) -> "XEREncoder":
+        """
+        Create a new XER encoder over a fresh, growable buffer.
+
+        This is the go-to way to create an encoder when the output size is not
+        known in advance. XER is text and always builds incrementally, so this
+        is equivalent to of_size() without a hint; it exists for API consistency
+        with the binary encoders.
+
+        Returns:
+            A new XEREncoder instance.
+        """
+        return cls()
+
     def write_raw(self, text: str) -> None:
         """
         Write raw text to the encoder.

asn1python/tests/test_codec_empty.py

@@ -0,0 +1,112 @@
+"""
+Unit tests for the .empty() encoder initializer.
+
+.empty() creates an encoder over a fresh, growable buffer without requiring a
+size estimate up front. These tests cover the backlog acceptance criteria:
+  - each encoder can be created via .empty()
+  - encoding into an .empty() encoder yields the same bytes as an of_size() one
+  - a round-trip (encode with .empty(), decode with from_buffer) is lossless
+and guard the bounded-buffer semantics that of_size()/from_buffer() rely on.
+"""
+import pytest
+
+from asn1python.acn_encoder import ACNEncoder
+from asn1python.acn_decoder import ACNDecoder
+from asn1python.codec_uper import UPEREncoder, UPERDecoder
+from asn1python.bitstream import BitStream, BitStreamError
+
+try:
+    from asn1python.xer_encoder import XEREncoder
+    HAS_XER = True
+except ImportError:
+    HAS_XER = False
+
+
+BINARY_ENCODERS = [ACNEncoder, UPEREncoder]
+
+
+def _encode_sample(encoder) -> None:
+    """Encode a fixed, mixed-width sequence of values into `encoder`."""
+    assert encoder.encode_integer(42, min_val=0, max_val=255).success
+    assert encoder.append_bit(True).success
+    assert encoder.encode_integer(-7, min_val=-128, max_val=127).success
+    assert encoder.append_byte(0xAB).success
+    assert encoder.encode_integer(1000, min_val=0, max_val=65535).success
+
+
+@pytest.mark.parametrize("encoder_cls", BINARY_ENCODERS)
+def test_empty_can_be_created(encoder_cls) -> None:
+    encoder = encoder_cls.empty()
+    assert isinstance(encoder, encoder_cls)
+
+
+@pytest.mark.parametrize("encoder_cls", BINARY_ENCODERS)
+def test_empty_matches_of_size_output(encoder_cls) -> None:
+    empty_encoder = encoder_cls.empty()
+    sized_encoder = encoder_cls.of_size(1024)
+
+    _encode_sample(empty_encoder)
+    _encode_sample(sized_encoder)
+
+    assert empty_encoder.get_bitstream_buffer() == sized_encoder.get_bitstream_buffer()
+
+
+@pytest.mark.parametrize("encoder_cls,decoder_cls", [
+    (ACNEncoder, ACNDecoder),
+    (UPEREncoder, UPERDecoder),
+])
+def test_empty_round_trip(encoder_cls, decoder_cls) -> None:
+    encoder = encoder_cls.empty()
+    assert encoder.encode_integer(12345, min_val=0, max_val=100000).success
+
+    buffer = encoder.get_bitstream_buffer()
+    decoder = decoder_cls.from_buffer(buffer)
+    result = decoder.decode_integer(min_val=0, max_val=100000)
+
+    assert result.success
+    assert result.decoded_value == 12345
+
+
+@pytest.mark.parametrize("encoder_cls", BINARY_ENCODERS)
+def test_empty_grows_far_beyond_initial_capacity(encoder_cls) -> None:
+    """An .empty() encoder must accept far more than any small initial buffer."""
+    encoder = encoder_cls.empty()
+
+    num_bytes = 5000
+    for i in range(num_bytes):
+        assert encoder.append_byte(i % 256).success
+
+    buffer = encoder.get_bitstream_buffer()
+    assert len(buffer) == num_bytes
+    assert all(buffer[i] == i % 256 for i in range(num_bytes))
+
+
+def test_fixed_buffer_still_raises_on_overflow() -> None:
+    """Non-growable buffers (of_size / from_buffer / decoders) stay bounded."""
+    stream = BitStream(bytearray(1))  # one byte, not growable
+    stream.write_byte(0xFF)
+    with pytest.raises(BitStreamError):
+        stream.write_bit(True)
+
+
+def test_growable_flag_survives_copy() -> None:
+    """from_bitstream must preserve the growable flag (used by Codec.__init__)."""
+    growable = BitStream(bytearray(), growable=True)
+    copy = BitStream.from_bitstream(growable)
+    copy.write_byte(0x01)  # would raise on a zero-length fixed buffer
+    assert copy.get_data() == bytearray([0x01])
+
+    fixed = BitStream(bytearray(1))
+    fixed_copy = BitStream.from_bitstream(fixed)
+    assert fixed_copy._growable is False
+
+
+@pytest.mark.skipif(not HAS_XER, reason="XER backend not available")
+def test_xer_empty() -> None:
+    empty_encoder = XEREncoder.empty()
+    sized_encoder = XEREncoder.of_size()
+
+    for enc in (empty_encoder, sized_encoder):
+        enc.encode_integer("value", 7, 0)
+
+    assert empty_encoder.get_xml() == sized_encoder.get_xml()