phase 0 + 1: repo bones and ETF codec
Repo scaffolding (LICENSE, Makefile, pyproject.toml, README) matching unfeed conventions. Flat package layout, ruff config, Unlicense. ETF codec covers the subset needed for gen_call against an Elixir node: small/int/big_int, atom_utf8 (legacy atom_ext on decode), binary, nil, list, small/large tuple, new_pid, newer_reference. Booleans round-trip as atoms true/false; Python None as atom nil; str encodes to utf-8 binary to match Elixir convention. Golden vectors were generated from real Erlang term_to_binary/1 output (generator script at docs/etf_vectors.erl). Decode tests verify wire compatibility; round-trip tests verify encoder consistency. make all green: 58 passed, lint clean. Next phases tracked in docs/ROADMAP.md (EPMD, handshake, channel, gen_call, TLS, unfeed integration).
This commit is contained in:
commit
b9fd28ca3b
11 changed files with 680 additions and 0 deletions
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
124
tests/test_etf.py
Normal file
124
tests/test_etf.py
Normal file
|
|
@ -0,0 +1,124 @@
|
|||
"""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")))),
|
||||
]
|
||||
|
||||
|
||||
@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"),)),
|
||||
])
|
||||
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"")
|
||||
Loading…
Add table
Add a link
Reference in a new issue