From 8cca12b3387e32a2f671ae37772716d01d69ebe1 Mon Sep 17 00:00:00 2001 From: Manik Jain Date: Mon, 11 Aug 2025 14:45:48 +0000 Subject: [PATCH] chore: consolidate codec utils --- src/test_suite/context/codec_utils.py | 34 ++++++++++++++++++++++++++- src/test_suite/fuzz_interface.py | 33 -------------------------- src/test_suite/instr/codec_utils.py | 8 +++++-- src/test_suite/txn/codec_utils.py | 8 +++++-- 4 files changed, 45 insertions(+), 38 deletions(-) diff --git a/src/test_suite/context/codec_utils.py b/src/test_suite/context/codec_utils.py index 39b45ec..77fb362 100644 --- a/src/test_suite/context/codec_utils.py +++ b/src/test_suite/context/codec_utils.py @@ -1,6 +1,38 @@ import test_suite.context_pb2 as context_pb import fd58 -from test_suite.fuzz_interface import encode_hex_compact, decode_hex_compact +import re + + +def decode_hex_compact(encoded): + res = bytearray() + parts = re.split(r"\.\.\.(\d+) zeros\.\.\.", encoded.decode("ascii")) + + for i, part in enumerate(parts): + if i % 2 == 0: + # Regular hex part + res.extend(bytes.fromhex(part)) + else: + # Skipped zeros part + res.extend(b"\x00" * int(part)) + + return bytes(res) + + +def encode_hex_compact(buf, gap=16): + res = "" + skipped = 0 + for i in range(0, len(buf), gap): + row = buf[i : i + gap] + if row == bytes([0] * len(row)): + skipped += len(row) + else: + if skipped > 0: + res += f"...{skipped} zeros..." + res += "".join([f"{b:0>2x}" for b in buf[i : i + gap]]) + skipped = 0 + if skipped > 0: + res += f"...{skipped} zeros..." + return bytes(res, "ascii") def encode_acct_state(acct_state: context_pb.AcctState): diff --git a/src/test_suite/fuzz_interface.py b/src/test_suite/fuzz_interface.py index 0cb4a1c..6e71169 100644 --- a/src/test_suite/fuzz_interface.py +++ b/src/test_suite/fuzz_interface.py @@ -1,7 +1,6 @@ from typing import Callable, Type, TypeVar from google.protobuf import message, descriptor, message_factory from dataclasses import dataclass, InitVar, field -import re msg_factory = message_factory.MessageFactory() @@ -27,38 +26,6 @@ """ -def decode_hex_compact(encoded): - res = bytearray() - parts = re.split(r"\.\.\.(\d+) zeros\.\.\.", encoded.decode("ascii")) - - for i, part in enumerate(parts): - if i % 2 == 0: - # Regular hex part - res.extend(bytes.fromhex(part)) - else: - # Skipped zeros part - res.extend(b"\x00" * int(part)) - - return bytes(res) - - -def encode_hex_compact(buf, gap=16): - res = "" - skipped = 0 - for i in range(0, len(buf), gap): - row = buf[i : i + gap] - if row == bytes([0] * len(row)): - skipped += len(row) - else: - if skipped > 0: - res += f"...{skipped} zeros..." - res += "".join([f"{b:0>2x}" for b in buf[i : i + gap]]) - skipped = 0 - if skipped > 0: - res += f"...{skipped} zeros..." - return bytes(res, "ascii") - - def generic_effects_prune( ctx: ContextType | None, effects: dict[str, str | None] ) -> dict[str, str | None] | None: diff --git a/src/test_suite/instr/codec_utils.py b/src/test_suite/instr/codec_utils.py index dd50f7f..b371abe 100644 --- a/src/test_suite/instr/codec_utils.py +++ b/src/test_suite/instr/codec_utils.py @@ -1,6 +1,10 @@ import fd58 -from test_suite.context.codec_utils import decode_acct_state, encode_acct_state -from test_suite.fuzz_interface import decode_hex_compact, encode_hex_compact +from test_suite.context.codec_utils import ( + decode_acct_state, + decode_hex_compact, + encode_acct_state, + encode_hex_compact, +) import test_suite.invoke_pb2 as invoke_pb diff --git a/src/test_suite/txn/codec_utils.py b/src/test_suite/txn/codec_utils.py index 36972e3..234391e 100644 --- a/src/test_suite/txn/codec_utils.py +++ b/src/test_suite/txn/codec_utils.py @@ -1,6 +1,10 @@ import fd58 -from test_suite.context.codec_utils import decode_acct_state, encode_acct_state -from test_suite.fuzz_interface import decode_hex_compact, encode_hex_compact +from test_suite.context.codec_utils import ( + decode_acct_state, + decode_hex_compact, + encode_acct_state, + encode_hex_compact, +) import test_suite.txn_pb2 as txn_pb