russell@unturf.com
531c26fcfb
Python dict round-trips as an Erlang/Elixir map. Golden vectors from
real Erlang term_to_binary verify wire compatibility for:
- empty map (#{})
- keyed map with binary keys + mixed values
- {ok, #{status => ok, balance => N}} (Elixir gen_server reply shape)
122 tests green. Required for the unfeed ErlangDistTransport which
decodes Elixir wallet RPC replies.
133 lines
5.2 KiB
Python
133 lines
5.2 KiB
Python
"""ETF codec tests.
|
|
|
|
Golden vectors below were generated by a real Erlang VM via
|
|
``term_to_binary/1``. See ``docs/etf_vectors.erl`` for the generator script.
|
|
Decode tests prove we read the wire spec correctly; round-trip tests prove
|
|
the encoder is internally consistent.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from erldistpy.etf import (
|
|
Atom,
|
|
ETFError,
|
|
Pid,
|
|
Reference,
|
|
decode,
|
|
encode,
|
|
)
|
|
|
|
# Golden vectors: (name, hex_from_real_vm, expected_python_value)
|
|
GOLDEN = [
|
|
("small_int_0", "836100", 0),
|
|
("small_int_42", "83612a", 42),
|
|
("small_int_255", "8361ff", 255),
|
|
("int_256", "836200000100", 256),
|
|
("int_neg_1", "8362ffffffff", -1),
|
|
("int_max32", "83627fffffff", 2147483647),
|
|
("int_min32", "836280000000", -2147483648),
|
|
("big_pos", "836e0700ffffc06ff28623", 9999999999999999),
|
|
("big_neg", "836e0701ffffc06ff28623", -9999999999999999),
|
|
("atom_hello", "8364000568656c6c6f", Atom("hello")),
|
|
("atom_true", "8364000474727565", True),
|
|
("atom_false", "8364000566616c7365", False),
|
|
("atom_nil", "836400036e696c", None),
|
|
("atom_unicode", "83770cd0bfd180d0b8d0b2d0b5d182", Atom("привет")),
|
|
("binary_empty", "836d00000000", b""),
|
|
("binary_hi", "836d000000026869", b"hi"),
|
|
("binary_utf8", "836d0000000668c3a96c6c6f", "héllo".encode()),
|
|
("tuple_empty", "836800", ()),
|
|
("tuple_pair", "8368026400026f6b6d0000000576616c7565", (Atom("ok"), b"value")),
|
|
("tuple_triple", "836803610161026103", (1, 2, 3)),
|
|
("list_empty", "836a", []),
|
|
# NOTE: small ints all < 256 → Erlang emits STRING_EXT. Decodes to list of ints.
|
|
("list_ints", "836b0003010203", [1, 2, 3]),
|
|
("list_mixed", "836c0000000364000568656c6c6f6d00000005776f726c64612a6a",
|
|
[Atom("hello"), b"world", 42]),
|
|
("nested",
|
|
"83680364000867656e5f63616c6c6c0000000168026400046e6f646564000c77616c6c65744063616d6d796a680264000b6765745f62616c616e6365640003786d72",
|
|
(Atom("gen_call"), [(Atom("node"), Atom("wallet@cammy"))], (Atom("get_balance"), Atom("xmr")))),
|
|
("empty_map", "837400000000", {}),
|
|
("addr_map",
|
|
"8374000000026d00000007616464726573736d00000005786d722d316d0000000863757272656e6379640003786d72",
|
|
{b"address": b"xmr-1", b"currency": Atom("xmr")}),
|
|
("nested_ok",
|
|
"8368026400026f6b740000000264000762616c616e6365620012d6876400067374617475736400026f6b",
|
|
(Atom("ok"), {Atom("balance"): 1234567, Atom("status"): Atom("ok")})),
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("name,hex_in,expected", GOLDEN, ids=[g[0] for g in GOLDEN])
|
|
def test_decode_golden(name, hex_in, expected):
|
|
assert decode(bytes.fromhex(hex_in)) == expected
|
|
|
|
|
|
# Round-trip — our encoder may pick different tags (e.g. SMALL_ATOM_UTF8_EXT
|
|
# instead of legacy ATOM_EXT) but the term must survive a round-trip intact.
|
|
@pytest.mark.parametrize("term", [
|
|
0, 1, 255, 256, -1, 2**31 - 1, -(2**31),
|
|
10**20, -(10**20),
|
|
Atom("hello"), Atom("ok"), True, False, None,
|
|
b"", b"hi", b"\x00\x01\x02",
|
|
(), (1,), (Atom("ok"), b"value"),
|
|
[], [1, 2, 3], [Atom("a"), b"b", 3],
|
|
(Atom("gen_call"), [], (Atom("ping"),)),
|
|
{}, {Atom("k"): 1}, {b"a": b"b", b"c": 42},
|
|
(Atom("ok"), {Atom("status"): Atom("ok"), Atom("count"): 7}),
|
|
])
|
|
def test_round_trip(term):
|
|
assert decode(encode(term)) == term
|
|
|
|
|
|
def test_str_encodes_as_binary():
|
|
# Python str → utf-8 binary (matches Elixir convention)
|
|
assert decode(encode("hello")) == b"hello"
|
|
|
|
|
|
def test_pid_round_trip():
|
|
p = Pid(node=Atom("wallet@cammy"), id=42, serial=1, creation=7)
|
|
assert decode(encode(p)) == p
|
|
|
|
|
|
def test_reference_round_trip():
|
|
r = Reference(node=Atom("wallet@cammy"), creation=3, ids=(100, 200, 300))
|
|
assert decode(encode(r)) == r
|
|
|
|
|
|
def test_bool_before_int():
|
|
# bool is a subclass of int; must encode as atom, not as integer
|
|
out = encode(True)
|
|
assert decode(out) is True
|
|
assert decode(encode(False)) is False
|
|
|
|
|
|
def test_atom_too_long():
|
|
with pytest.raises(ETFError):
|
|
Atom("x" * 256)
|
|
|
|
|
|
def test_bad_magic():
|
|
with pytest.raises(ETFError, match="bad magic"):
|
|
decode(b"\x00\x00")
|
|
|
|
|
|
def test_unknown_tag():
|
|
with pytest.raises(ETFError, match="unknown tag"):
|
|
decode(bytes([131, 0xFE]))
|
|
|
|
|
|
def test_trailing_bytes():
|
|
with pytest.raises(ETFError, match="trailing bytes"):
|
|
decode(bytes.fromhex("836100") + b"\x00")
|
|
|
|
|
|
def test_unencodable_type():
|
|
with pytest.raises(ETFError, match="cannot encode"):
|
|
encode(object())
|
|
|
|
|
|
def test_empty_input():
|
|
with pytest.raises(ETFError, match="bad magic"):
|
|
decode(b"")
|