Add `parse_maddr()` tests + registrar maddr integ test

Cover `parse_maddr()` with unit tests for tcp/ipv4,
tcp/ipv6, uds, and unsupported-protocol error paths,
plus full `addr -> mk_maddr -> str -> parse_maddr`
roundtrip verification.

Adds,
- a `_maddr_to_tpt_proto` inverse-mapping assertion.
- an `wrap_address()` maddr-string acceptance test.
- a `test_reg_then_unreg_maddr` end-to-end suite which audits passing
  the registry addr as multiaddr str through the entire runtime.

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
subint_spawner_backend
Gud Boi 2026-03-26 19:10:20 -04:00
parent 90ba0e3658
commit 5c4438bacc
2 changed files with 158 additions and 2 deletions

View File

@ -1,6 +1,7 @@
'''
Multiaddr construction and round-trip tests for
`tractor.discovery._multiaddr.mk_maddr()`.
Multiaddr construction, parsing, and round-trip tests for
`tractor.discovery._multiaddr.mk_maddr()` and
`tractor.discovery._multiaddr.parse_maddr()`.
'''
from pathlib import Path
@ -13,8 +14,11 @@ from tractor.ipc._tcp import TCPAddress
from tractor.ipc._uds import UDSAddress
from tractor.discovery._multiaddr import (
mk_maddr,
parse_maddr,
_tpt_proto_to_maddr,
_maddr_to_tpt_proto,
)
from tractor.discovery._addr import wrap_address
def test_tpt_proto_to_maddr_mapping():
@ -134,3 +138,110 @@ def test_mk_maddr_roundtrip(addr):
assert reparsed == maddr
assert str(reparsed) == str(maddr)
# ------ parse_maddr() tests ------
def test_maddr_to_tpt_proto_mapping():
'''
`_maddr_to_tpt_proto` is the exact inverse of
`_tpt_proto_to_maddr`.
'''
assert _maddr_to_tpt_proto == {
'tcp': 'tcp',
'unix': 'uds',
}
def test_parse_maddr_tcp_ipv4():
'''
`parse_maddr()` on an IPv4 TCP multiaddr string
produce a `TCPAddress` with the correct host and port.
'''
result = parse_maddr('/ip4/127.0.0.1/tcp/1234')
assert isinstance(result, TCPAddress)
assert result.unwrap() == ('127.0.0.1', 1234)
def test_parse_maddr_tcp_ipv6():
'''
`parse_maddr()` on an IPv6 TCP multiaddr string
produce a `TCPAddress` with the correct host and port.
'''
result = parse_maddr('/ip6/::1/tcp/5678')
assert isinstance(result, TCPAddress)
assert result.unwrap() == ('::1', 5678)
def test_parse_maddr_uds():
'''
`parse_maddr()` on a `/unix/...` multiaddr string
produce a `UDSAddress` with the correct dir and filename.
'''
result = parse_maddr('/unix/tractor_test/test.sock')
assert isinstance(result, UDSAddress)
filedir, filename = result.unwrap()
assert filename == 'test.sock'
assert 'tractor_test' in str(filedir)
def test_parse_maddr_unsupported():
'''
`parse_maddr()` raise `ValueError` for an unsupported
protocol combination like UDP.
'''
with pytest.raises(
ValueError,
match='Unsupported multiaddr protocol combo',
):
parse_maddr('/ip4/127.0.0.1/udp/1234')
@pytest.mark.parametrize(
'addr',
[
pytest.param(
TCPAddress('127.0.0.1', 9999),
id='tcp-ipv4',
),
pytest.param(
UDSAddress(
filedir='tractor_rt',
filename='roundtrip.sock',
),
id='uds',
),
],
)
def test_parse_maddr_roundtrip(addr):
'''
Full round-trip: `addr -> mk_maddr -> str -> parse_maddr`
produce an `Address` whose `.unwrap()` matches the original.
'''
maddr: Multiaddr = mk_maddr(addr)
maddr_str: str = str(maddr)
parsed = parse_maddr(maddr_str)
assert type(parsed) is type(addr)
assert parsed.unwrap() == addr.unwrap()
def test_wrap_address_maddr_str():
'''
`wrap_address()` accept a multiaddr-format string and
return the correct `Address` type.
'''
result = wrap_address('/ip4/127.0.0.1/tcp/9999')
assert isinstance(result, TCPAddress)
assert result.unwrap() == ('127.0.0.1', 9999)

View File

@ -16,6 +16,8 @@ import subprocess
import tractor
from tractor.trionics import collapse_eg
from tractor._testing import tractor_test
from tractor.discovery._addr import wrap_address
from tractor.discovery._multiaddr import mk_maddr
import trio
@ -53,6 +55,49 @@ async def test_reg_then_unreg(
assert not sockaddrs
@tractor_test
async def test_reg_then_unreg_maddr(
reg_addr: tuple,
):
'''
Same as `test_reg_then_unreg` but pass the registry
address as a multiaddr string to verify `wrap_address()`
multiaddr parsing end-to-end through the runtime.
'''
# tuple -> Address -> multiaddr string
addr_obj = wrap_address(reg_addr)
maddr_str: str = str(mk_maddr(addr_obj))
actor = tractor.current_actor()
assert actor.is_registrar
async with tractor.open_nursery(
registry_addrs=[maddr_str],
) as n:
portal = await n.start_actor(
'actor_maddr',
enable_modules=[__name__],
)
uid = portal.channel.aid.uid
async with tractor.get_registry(maddr_str) as aportal:
assert actor is aportal.actor
async with tractor.wait_for_actor('actor_maddr'):
assert uid in aportal.actor._registry
sockaddrs = actor._registry[uid]
assert sockaddrs
await n.cancel()
await trio.sleep(0.1)
assert uid not in aportal.actor._registry
sockaddrs = actor._registry.get(uid)
assert not sockaddrs
the_line = 'Hi my name is {}'