Merge pull request #141 from firedancer-io/mjain/reorg-utils

chore: consolidate codec utils

Author: mjain-jump

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):

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:

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
 
 

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