tractor/tests/ipc/test_tipc.py

'''
Unit tests for the `AF_TIPC` transport backend, `tractor.ipc._tipc`.

The kernel-touching cases are gated on `is_tipc_available()` since
the `tipc` module is NOT loaded by default (`sudo modprobe tipc`);
the pure address-algebra cases run everywhere.

'''
from __future__ import annotations
import errno
import struct
from types import SimpleNamespace
from socket import (
    SOCK_STREAM,
    SOL_SOCKET,
    SO_ACCEPTCONN,
)

import pytest
import trio

from tractor.msg.types import Aid
from tractor.discovery import _addr
from tractor.discovery._addr import wrap_address
from tractor.discovery._multiaddr import (
    mk_maddr,
    parse_maddr,
)
from tractor.ipc import _tipc
from tractor.ipc._uds import UDSAddress
from tractor.ipc._tipc import (
    AF_TIPC,
    SOL_TIPC,
    TIPC_ADDR_ID,
    TIPC_ADDR_NAME,
    TIPC_CLUSTER_SCOPE,
    TIPC_DEST_DROPPABLE,
    TIPC_HIGH_IMPORTANCE,
    TIPC_IMPORTANCE,
    TIPC_NAME_UNKNOWN,
    TIPC_NODE_SCOPE,
    TIPC_PUBLISHED,
    TIPC_SCOPE_UNKNOWN,
    TIPC_SUBSCR_TIMEOUT,
    TIPC_ZONE_SCOPE,
    TRACTOR_STYPE,
    MsgpackTIPCStream,
    TIPCAddress,
    instance_from_seed,
    is_tipc_available,
    open_topology_events,
    start_listener,
)


pytestmark = pytest.mark.tipc

requires_tipc = pytest.mark.skipif(
    not is_tipc_available(),
    reason=(
        '`tipc` kernel module not loaded (`sudo modprobe tipc`)'
    ),
)


# ------------------------------------------------------------------
# address algebra (no kernel needed)
# ------------------------------------------------------------------

@pytest.mark.parametrize(
    'addr',
    [
        TIPCAddress.get_root(),
        TIPCAddress(
            _stype=TRACTOR_STYPE,
            _instance=42,
            _scope=TIPC_NODE_SCOPE,
        ),
    ],
    ids=['root', 'node-scoped'],
)
def test_addr_unwrap_roundtrip(addr: TIPCAddress):
    '''
    `.unwrap()` is proto-keyed and `.from_addr()` inverts it — for
    both the `tuple` form and the `list` form msgpack decodes to.

    '''
    unwrapped: tuple = addr.unwrap()
    assert unwrapped[0] == 'tipc' == TIPCAddress.proto_key
    assert len(unwrapped) == 4

    assert TIPCAddress.from_addr(unwrapped) == addr
    assert TIPCAddress.from_addr(list(unwrapped)) == addr


def test_addr_scope_defaults_when_omitted():
    '''
    A 3-elem `('tipc', stype, inst)` form defaults to the
    cluster-scope bindspace.

    '''
    addr: TIPCAddress = TIPCAddress.from_addr(
        ('tipc', TRACTOR_STYPE, 99),
    )
    assert addr._scope == TIPC_CLUSTER_SCOPE
    assert addr.bindspace == TIPCAddress.def_bindspace


def test_zone_scope_normalized_to_cluster():
    '''
    `TIPC_ZONE_SCOPE` is deprecated/aliased in modern kernels;
    accept it on input, fold it to cluster.

    '''
    addr: TIPCAddress = TIPCAddress.from_addr(
        ('tipc', TRACTOR_STYPE, 7, TIPC_ZONE_SCOPE),
    )
    assert addr._scope == TIPC_CLUSTER_SCOPE
    assert addr.is_valid


def test_maddr_parse_normalizes_and_reports_bad_input():
    '''
    The interim `/tipc` parser must route through
    `TIPCAddress.from_addr()` so deprecated zone scope is normalized
    exactly like every other unwrapped-address entrypoint.

    A malformed segment count previously leaked the tuple-unpacking
    `ValueError`, which gave callers no indication that the TIPC
    multiaddr grammar itself was invalid.

    '''
    addr: TIPCAddress = parse_maddr(
        f'/tipc/{TRACTOR_STYPE}/7/{TIPC_ZONE_SCOPE}'
    )
    assert addr._scope == TIPC_CLUSTER_SCOPE

    with pytest.raises(
        ValueError,
        match='Invalid TIPC multiaddr',
    ):
        parse_maddr('/tipc/not-enough-segments')


def test_addr_from_bare_port_id_raises():
    '''
    A `TIPC_ADDR_ID` 5-tuple carries no service-name so it can
    NEVER be wrapped; it must fail loudly rather than silently
    fabricate an un-dialable addr.

    This is the invariant that lets
    `TIPCAddress.rebind_from_sockname` be `False`.

    '''
    with pytest.raises(ValueError) as excinfo:
        TIPCAddress.from_addr((TIPC_ADDR_ID, 0, 12345, 0, 0))

    assert 'port-id' in str(excinfo.value)


def test_addr_is_valid_predicate():
    assert TIPCAddress.get_root().is_valid

    # instance 0 is not a bindable name
    assert not TIPCAddress(
        _stype=TRACTOR_STYPE,
        _instance=0,
    ).is_valid

    # service-types 0..63 are TIPC-internal (`TIPC_CFG_SRV`,
    # `TIPC_TOP_SRV`, ..)
    assert not TIPCAddress(
        _stype=1,
        _instance=1616,
    ).is_valid


def test_port_id_is_annotation_only():
    '''
    `.maybe_node`/`.maybe_ref` are *observed* metadata, excluded
    from `.unwrap()` exactly like `UDSAddress.maybe_pid`.

    '''
    addr: TIPCAddress = TIPCAddress.get_root()
    annotated: TIPCAddress = addr.with_port_id(
        node=0xdead,
        ref=1234,
    )
    assert annotated.unwrap() == addr.unwrap()
    assert annotated.maybe_ref == 1234
    assert '1234' in repr(annotated)


def test_instance_from_seed_is_pure():
    '''
    Same seed -> same instance (what the follow-up registrar-less
    discovery fast-path will lean on), and always clear of the
    reserved low range.

    '''
    for seed in ('doggy@123', 'kitty@456', ''):
        inst: int = instance_from_seed(seed)
        assert inst == instance_from_seed(seed)
        assert 64 <= inst < 2**32


def test_get_random_collision_resistance():
    '''
    A `.get_random()` clash does NOT raise `EADDRINUSE` — TIPC
    accepts multiple publishers of one name and round-robins
    connects between them, so a collision is *silent crosstalk*.

    Assert the 4-byte digest spreads well enough for that to stay
    improbable.

    NOTE the bound is birthday-statistical, not absolute:
    P(collision) ~= 1 - exp(-n**2 / 2**33) ~= 1.2e-2 for n=10k, so
    a strict `== n` assert would be ~1-in-86 flaky. P(>2
    collisions) is ~1e-7, hence the slack. See plan 01 §9 for the
    escalation path if this ever trips.

    '''
    n: int = 10_000
    addrs: list[TIPCAddress] = [
        TIPCAddress.get_random()
        for _ in range(n)
    ]
    instances: set[int] = {
        addr._instance
        for addr in addrs
    }
    assert len(instances) >= n - 2

    # every one is a legal, bindable name
    assert all(addr.is_valid for addr in addrs)


def test_get_random_keys_live_actors_by_uuid(
    monkeypatch: pytest.MonkeyPatch,
):
    '''
    TIPC names are cluster-wide while PIDs are host-local. Hashing
    only `(actor name, pid)` therefore made same-named actors with
    equal PIDs on different hosts publish one service name, where
    TIPC silently round-robins connects between them.

    Hold the actor name and PID fixed while changing only its UUID;
    distinct instances prove the globally unique identity field is
    now part of the derivation.

    '''
    monkeypatch.setattr(_tipc.os, 'getpid', lambda: 1616)

    def get_addr(uuid: str) -> TIPCAddress:
        actor = SimpleNamespace(
            aid=Aid(
                name='worker',
                uuid=uuid,
                pid=1616,
            )
        )
        monkeypatch.setattr(
            _tipc,
            'current_actor',
            lambda **kwargs: actor,
        )
        return TIPCAddress.get_random()

    first: TIPCAddress = get_addr('actor-uuid-a')
    second: TIPCAddress = get_addr('actor-uuid-b')
    assert first._instance != second._instance
    assert get_addr('actor-uuid-a')._instance == first._instance


def test_get_random_honors_bindspace():
    addr: TIPCAddress = TIPCAddress.get_random(
        bindspace=TIPC_NODE_SCOPE,
    )
    assert addr.bindspace == TIPC_NODE_SCOPE == addr._scope


def test_wrap_address_dispatches_on_the_proto_key():
    '''
    The proto-keyed unwrapped form must round-trip through the
    *global* `wrap_address()` — and NOT get stolen by `tcp`s
    `(str(), int())` case nor `uds`s `(_, str())` one.

    '''
    addr: TIPCAddress = TIPCAddress.get_random()
    assert wrap_address(addr.unwrap()) == addr
    # ..and via the `list` form `msgpack` decodes to
    assert wrap_address(list(addr.unwrap())) == addr

    assert _addr._address_types['tipc'] is TIPCAddress
    assert _addr.get_address_cls('tipc') is TIPCAddress

    # the host-singleton registrar default is import-time cheap
    # (no kernel module, no I/O) and mirrors the `1616` idiom
    assert _addr._default_lo_addrs['tipc'] == (
        'tipc', TRACTOR_STYPE, 1616, TIPC_CLUSTER_SCOPE,
    )

    # A UDS directory named `tipc` is still a valid classic
    # 2-element address, not a malformed proto-keyed TIPC one.
    uds: UDSAddress = wrap_address(('tipc', 'actor.sock'))
    assert isinstance(uds, UDSAddress)
    assert UDSAddress.unwrapped_type == tuple[str, str]


def test_maddr_roundtrip():
    '''
    Interim `str`-only `/tipc/` maddr grammar (there's no
    registered `/tipc` multiaddr proto yet, gh #483), which
    `parse_maddr()` special-cases before `Multiaddr()` ever sees
    the string.

    '''
    addr: TIPCAddress = TIPCAddress.get_random()
    maddr: str = mk_maddr(addr)

    assert isinstance(maddr, str)
    assert maddr == (
        f'/tipc/{addr._stype}/{addr._instance}/{addr._scope}'
    )
    assert parse_maddr(maddr) == addr
    # ..and through the generic entrypoint
    assert wrap_address(maddr) == addr


def test_eafnosupport_is_actionable_connerr(
    monkeypatch: pytest.MonkeyPatch,
):
    '''
    With no `tipc` module the kernel answers `EAFNOSUPPORT`; that
    MUST surface as a `ConnectionError` naming the fix rather than
    a bare `OSError`.

    '''
    class _NoTIPCKernel:
        @staticmethod
        def socket(*args, **kwargs):
            raise OSError(
                errno.EAFNOSUPPORT,
                'Address family not supported by protocol',
            )

    monkeypatch.setattr(_tipc, 'trio_socket', _NoTIPCKernel)

    async def main():
        await start_listener(addr=TIPCAddress.get_root())

    with pytest.raises(ConnectionError) as excinfo:
        trio.run(main)

    report: str = str(excinfo.value)
    assert 'modprobe tipc' in report
    assert type(excinfo.value.__cause__) is OSError


def test_availability_probe_is_linux_only(
    monkeypatch: pytest.MonkeyPatch,
):
    '''
    The fallback `AF_TIPC` integer keeps imports portable but can
    alias an unrelated socket family on another OS. Socket creation
    alone must never turn that numeric collision into a false TIPC
    capability result.

    Simulate Darwin with a socket constructor that would otherwise
    succeed; proving it is never called pins the platform gate ahead
    of the syscall probe.

    '''
    socket_called: bool = False

    def fake_socket(*args, **kwargs):
        nonlocal socket_called
        socket_called = True

    monkeypatch.setattr(_tipc, '_tipc_avail', None)
    monkeypatch.setattr(_tipc.sys, 'platform', 'darwin')
    monkeypatch.setattr(_tipc.socket, 'socket', fake_socket)

    assert not is_tipc_available()
    assert not socket_called


def test_connect_reuses_tolerant_peer_observation(
    monkeypatch: pytest.MonkeyPatch,
):
    '''
    A TIPC peer can withdraw immediately after `.connect()`. The
    transport constructor already tolerates that race, but the dial
    path used to call `getpeername()` a second time and leak a raw
    `ENOTCONN` after an otherwise successful connection.

    The fake socket permits exactly the constructor's first peer
    lookup. A second lookup raises deterministically, so successful
    construction proves `connect_to()` reuses the tolerant observed
    address and transfers socket ownership only after setup.

    '''
    class _DropAfterObservation:
        def __init__(self):
            self.peer_lookups: int = 0
            self.closed: bool = False

        def setsockopt(self, *args) -> None:
            pass

        async def connect(self, addr) -> None:
            pass

        def getsockname(self) -> tuple[int, int, int, int, int]:
            return (TIPC_ADDR_ID, 1, 2, 0, TIPC_CLUSTER_SCOPE)

        def getpeername(self) -> tuple[int, int, int, int, int]:
            self.peer_lookups += 1
            if self.peer_lookups > 1:
                raise OSError(errno.ENOTCONN, 'peer withdrew')
            return (TIPC_ADDR_ID, 3, 4, 0, TIPC_CLUSTER_SCOPE)

        def close(self) -> None:
            self.closed = True

    async def main() -> None:
        sock = _DropAfterObservation()
        monkeypatch.setattr(
            _tipc.trio_socket,
            'socket',
            lambda *args: sock,
        )
        monkeypatch.setattr(
            _tipc.trio,
            'SocketStream',
            lambda raw_sock: SimpleNamespace(socket=raw_sock),
        )

        addr: TIPCAddress = TIPCAddress.get_root()
        stream: MsgpackTIPCStream = (
            await MsgpackTIPCStream.connect_to(addr)
        )
        assert sock.peer_lookups == 1
        assert stream.raddr.unwrap() == addr.unwrap()
        assert stream.raddr.maybe_ref == 4
        assert not sock.closed

    trio.run(main)


# ------------------------------------------------------------------
# kernel-touching
# ------------------------------------------------------------------

@requires_tipc
def test_listener_tolerates_so_acceptconn():
    '''
    `trio.SocketListener.__init__` asserts
    `getsockopt(SOL_SOCKET, SO_ACCEPTCONN)` is truthy, suppressing
    `OSError` for exotic families.

    Pin which of the two branches `AF_TIPC` actually takes (plan 01
    §3.1 left it as an assumption) so a kernel-side regression is
    caught here rather than as a mystery bind failure.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)
        try:
            assert lstnr.socket.getsockopt(
                SOL_SOCKET,
                SO_ACCEPTCONN,
            )
        finally:
            lstnr.socket.close()

    trio.run(main)


@requires_tipc
def test_bind_publishes_a_dialable_service_name():
    '''
    "Publishing a bind IS registration": `.bind()` a singleton
    name-seq and a second task resolves it by *name* — with NO
    `tractor` registrar in the loop.

    This is the core #378 property.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)

        accepted: list = []

        async def _accept():
            stream = await lstnr.accept()
            accepted.append(stream)
            await stream.send_all(b'woof')
            await stream.aclose()

        async with trio.open_nursery() as tn:
            tn.start_soon(_accept)
            await trio.sleep(0.05)

            sock = _tipc.trio_socket.socket(
                AF_TIPC,
                SOCK_STREAM,
            )
            # NOTE, connect by *name* -> the kernel does the
            # lookup, i.e. this call IS the discovery query.
            await sock.connect((
                TIPC_ADDR_NAME,
                addr._stype,
                addr._instance,
                0,  # domain: 0 == "anywhere in scope"
                addr._scope,
            ))
            stream = trio.SocketStream(sock)
            assert await stream.receive_some(16) == b'woof'
            await stream.aclose()

        assert len(accepted) == 1
        lstnr.socket.close()

    trio.run(main)


@requires_tipc
def test_getsockname_is_a_port_id_not_the_bound_name():
    '''
    The reason `TIPCAddress.rebind_from_sockname` is `False`.

    A NAMESEQ-bound listener's `getsockname()` answers a
    `TIPC_ADDR_ID` port-id, which never equals `.unwrap()` and
    cannot be wrapped back into a service name.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)
        try:
            sockname: tuple = lstnr.socket.getsockname()
            assert sockname[0] == TIPC_ADDR_ID
            assert sockname != addr.unwrap()
            with pytest.raises(ValueError):
                TIPCAddress.from_addr(sockname)
        finally:
            lstnr.socket.close()

    trio.run(main)


@requires_tipc
def test_msgpack_roundtrip_over_service_name():
    '''
    Two `trio` tasks in ONE proc exchange `msgpack`-framed msgs
    over a TIPC service name — no `tractor` runtime involved.

    Also pins the `(laddr, raddr)` story of plan 01 §3.4a: the
    dialling side knows the name it dialled, the accepting side
    only ever learns a port-id.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)

        ping = Aid(name='doggy', uuid='abc123', pid=1)
        pong = Aid(name='kitty', uuid='def456', pid=2)
        srv_got: list = []

        async def _serve():
            stream = await lstnr.accept()
            tpt = MsgpackTIPCStream(stream)

            # accepting side can NOT know the peer's service name
            assert not tpt.raddr.is_valid
            assert tpt.raddr._instance == TIPC_NAME_UNKNOWN
            assert tpt.raddr.maybe_ref is not None

            srv_got.append(await tpt.recv())
            await tpt.send(pong)
            await stream.aclose()

        async with trio.open_nursery() as tn:
            tn.start_soon(_serve)
            await trio.sleep(0.05)

            cli: MsgpackTIPCStream = await MsgpackTIPCStream.connect_to(
                destaddr=addr,
            )
            assert cli.connected()

            # dialling side DOES know the name, and it round-trips
            assert cli.raddr.is_valid
            assert cli.raddr.unwrap() == addr.unwrap()
            assert cli.raddr.maybe_ref is not None

            await cli.send(ping)
            assert await cli.recv() == pong
            await cli.stream.aclose()

        assert srv_got == [ping]
        lstnr.socket.close()

    trio.run(main)


@requires_tipc
def test_dial_unpublished_name_is_connerr():
    '''
    Contract §4: a dead/absent peer must normalize to
    `ConnectionError` — the discovery-ping path depends on it.

    XXX TIPC answers an unpublished-name lookup with
    `EHOSTUNREACH`, which python maps to a **bare** `OSError` (NOT
    a `ConnectionError` subtype the way `ECONNREFUSED` maps to
    `ConnectionRefusedError`), so the normalization is load-bearing
    rather than cosmetic.

    '''
    async def main():
        # nothing has ever `.bind()`ed this one
        nowhere = TIPCAddress(
            _stype=TRACTOR_STYPE,
            _instance=0xDEADBEEF,
        )
        with trio.fail_after(5):
            await MsgpackTIPCStream.connect_to(destaddr=nowhere)

    with pytest.raises(ConnectionError) as excinfo:
        trio.run(main)

    src_exc = excinfo.value.__cause__
    assert src_exc.errno == errno.EHOSTUNREACH
    assert 'No TIPC publisher' in str(excinfo.value)


@requires_tipc
def test_importance_sockopt_roundtrips():
    '''
    The `TIPC_IMPORTANCE` QoS knob (plan 01 §3.3) is settable and
    readable back — TIPC can rank a conn's traffic under
    congestion, which no other backend can do.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)

        # XXX hold the accepted conn open for the duration; a
        # peer that closes first makes `getpeername()` (called
        # from `MsgpackTransport.__init__`) raise `ENOTCONN`.
        done = trio.Event()

        async def _accept():
            stream = await lstnr.accept()
            await done.wait()
            await stream.aclose()

        async with trio.open_nursery() as tn:
            tn.start_soon(_accept)
            await trio.sleep(0.05)
            cli = await MsgpackTIPCStream.connect_to(
                destaddr=addr,
                importance=TIPC_HIGH_IMPORTANCE,
            )
            sock = cli.stream.socket
            assert sock.getsockopt(
                SOL_TIPC,
                TIPC_IMPORTANCE,
            ) == TIPC_HIGH_IMPORTANCE
            assert sock.getsockopt(
                SOL_TIPC,
                TIPC_DEST_DROPPABLE,
            ) == 0
            await cli.stream.aclose()
            done.set()

        lstnr.socket.close()

    trio.run(main)


@requires_tipc
def test_dropped_peer_does_not_kill_the_listener():
    '''
    A peer that connects then drops BEFORE we read must cost us
    an addr, not the runtime.

    XXX unlike tcp/uds — where the kernel keeps answering the
    peer addr until *we* close — TIPC answers `ENOTCONN` on
    `getpeername()` once the peer is gone. Since
    `MsgpackTransport.__init__()` calls `.get_stream_addrs()`
    (via `Channel.from_stream()`) BEFORE the handshake, an
    unguarded `OSError` there escapes
    `handle_stream_from_peer()`s handshake tolerance (contract
    §4) and tears down the whole actor.

    Real-world triggers: a port scan, a liveness probe (our own
    `tests/discovery/conftest.py::daemon` readiness poll does
    exactly this!), or a cancelled dial.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        lstnr = await start_listener(addr=addr)

        tpts: list = []

        async def _accept():
            stream = await lstnr.accept()
            # MUST NOT raise even though the peer is already gone
            tpts.append(MsgpackTIPCStream(stream))

        async with trio.open_nursery() as tn:
            tn.start_soon(_accept)
            await trio.sleep(0.05)

            # connect-then-immediately-drop
            sock = _tipc.trio_socket.socket(AF_TIPC, SOCK_STREAM)
            await sock.connect((
                TIPC_ADDR_NAME,
                addr._stype,
                addr._instance,
                0,
                addr._scope,
            ))
            sock.close()
            await trio.sleep(0.2)

        # NOTE the assertion that matters is simply that
        # `MsgpackTIPCStream()` above did NOT raise; whether
        # `getpeername()` still answers is a kernel-side race on
        # the disconnect indication, so don't pin `.maybe_ref`.
        assert len(tpts) == 1
        raddr: TIPCAddress = tpts[0].raddr
        assert not raddr.is_valid
        # ..and it still reprs cleanly for the con-status logs
        assert 'unknown-service' in repr(raddr)

        lstnr.socket.close()

    trio.run(main)


def test_observed_addr_tolerates_a_dead_peer():
    '''
    The deterministic half of the above: `_maybe_sockaddr()`
    swallows the `ENOTCONN` and `_observed_addr()` still yields
    a usable (name-less, port-id-less) addr.

    '''
    def _enotconn():
        raise OSError(
            errno.ENOTCONN,
            'Transport endpoint is not connected',
        )

    assert _tipc._maybe_sockaddr(_enotconn) is None

    addr: TIPCAddress = _tipc._observed_addr(None)
    assert not addr.is_valid
    assert addr.maybe_node is None
    assert addr.maybe_ref is None
    assert 'unknown-service' in repr(addr)


@requires_tipc
def test_duplicate_name_bind_does_not_raise():
    '''
    Unlike every other backend, TIPC permits *two* publishers of
    one service name and round-robins connects between them.

    Pin that observed behaviour — it's the whole reason
    `.get_random()` bothers with a well-spread digest, and a
    future kernel that starts raising `EADDRINUSE` here would be
    very good news worth noticing.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        first = await start_listener(addr=addr)
        second = await start_listener(addr=addr)
        try:
            assert first.socket.getsockname() != second.socket.getsockname()
        finally:
            first.socket.close()
            second.socket.close()

    trio.run(main)


# ------------------------------------------------------------------
# layer B — the topology service (`TIPC_TOP_SRV`)
# ------------------------------------------------------------------

def test_topology_struct_layouts():
    '''
    Pin the `include/uapi/linux/tipc.h` struct sizes.

    XXX the event is **48** bytes (`4+4+4+8+28`), NOT the 40 an
    earlier revision of plan 01 §5.2 claimed.

    '''
    assert struct.calcsize(_tipc._SUBSCR_FMT) == 28
    assert struct.calcsize(_tipc._EVENT_FMT) == 48
    assert _tipc._EVENT_SIZE == 48

    # ..and the subscription we actually emit is exactly that
    sub: bytes = _tipc._mk_subscr(
        stype=TRACTOR_STYPE,
        lower=0,
        upper=0xFFFF_FFFF,
        filt=_tipc.TIPC_SUB_SERVICE,
        timeout=_tipc.TIPC_WAIT_FOREVER,
    )
    assert len(sub) == 28


def _pack_topology_event(
    event: int,
    instance: int = 71,
) -> bytes:
    return struct.pack(
        _tipc._EVENT_FMT,
        event,
        instance,
        instance,
        123,
        456,
        0, 0, 0, 0, 0,
        b'\0' * 8,
    )


def test_wait_forever_is_masked_for_packing():
    '''
    Python exposes `TIPC_WAIT_FOREVER` as `-1`, which `struct`
    refuses to pack into an unsigned `'I'`; it MUST be masked.

    '''
    assert _tipc.TIPC_WAIT_FOREVER == -1

    with pytest.raises(struct.error):
        struct.pack('=I', _tipc.TIPC_WAIT_FOREVER)

    sub: bytes = _tipc._mk_subscr(
        stype=TRACTOR_STYPE,
        lower=0,
        upper=0,
        filt=_tipc.TIPC_SUB_SERVICE,
        timeout=_tipc.TIPC_WAIT_FOREVER,
    )
    _, _, _, timeout, _, _ = struct.unpack(_tipc._SUBSCR_FMT, sub)
    assert timeout == 0xFFFF_FFFF


def test_decode_name_event_rejects_junk():
    '''
    Runt frames and unknown event codes are dropped, never raised
    — a confused kernel must not kill the reader task.

    '''
    assert _tipc._decode_name_event(
        b'\x00' * 12,
        stype=TRACTOR_STYPE,
    ) is None

    bogus: bytes = struct.pack(
        _tipc._EVENT_FMT,
        99,  # not a known event code
        1, 1, 0, 0,
        0, 0, 0, 0, 0,
        b'\0' * 8,
    )
    assert _tipc._decode_name_event(
        bogus,
        stype=TRACTOR_STYPE,
    ) is None


def test_topology_event_scope_is_unknown():
    '''
    Neither `struct tipc_subscr` nor `struct tipc_event` carries a
    publication scope. The old decoder copied caller context into
    each `TIPCAddress`, falsely presenting cluster scope as kernel-
    observed data even for a node-scoped publisher.

    Decode a valid publication and assert its address uses the
    explicit unknown sentinel rather than inventing reachability.

    TIPC address types and publication scopes:
    https://docs.kernel.org/networking/tipc.html
    Event wire format (which has no scope field):
    https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h

    '''
    event: _tipc.TIPCNameEvent|None = _tipc._decode_name_event(
        _pack_topology_event(TIPC_PUBLISHED),
        stype=TRACTOR_STYPE,
    )
    assert event is not None
    assert event.addr._scope == TIPC_SCOPE_UNKNOWN
    assert not event.addr.is_valid


def test_topology_stream_surfaces_overflow():
    '''
    Topology transitions are `TIPC_PUBLISHED`/`TIPC_WITHDRAWN`
    changes to the set of ports matching a subscribed name sequence,
    as emitted by the kernel topology server:
    https://github.com/torvalds/linux/blob/master/net/tipc/topsrv.c

    Blocking this reader on `tx.send()` stops draining the topology
    socket and merely pushes pressure into the kernel queue. Silently
    dropping with `send_nowait()` is worse: a push registry keeps a
    stale view without knowing it missed a transition.

    A zero-capacity channel deterministically fills the user-space
    boundary. The dedicated overflow error proves the reader remains
    non-blocking while forcing consumers to resubscribe and rebuild
    instead of trusting incomplete state.

    '''
    class _OneEventSocket:
        def __init__(self):
            self.sent: bool = False

        async def recv(self, size: int) -> bytes:
            if not self.sent:
                self.sent = True
                return _pack_topology_event(TIPC_PUBLISHED)
            raise AssertionError('reader continued after overflow')

    async def main() -> None:
        tx, rx = trio.open_memory_channel(0)
        with pytest.raises(
            _tipc.TIPCNameEventOverflow,
            match='resubscribe and rebuild',
        ) as excinfo:
            await _tipc._stream_name_events(
                _OneEventSocket(),
                TRACTOR_STYPE,
                tx,
            )
        assert excinfo.value.event.kind == 'published'
        with pytest.raises(trio.EndOfChannel):
            await rx.receive()

    trio.run(main)


def test_topology_timeout_closes_stream():
    '''
    A finite kernel subscription expires after its timeout event.
    The old reader forwarded that event and waited forever for a
    second frame that could never arrive, so consumers blocked on a
    channel that looked live despite having no subscription.

    Feed one timeout frame directly to the reader; receiving that
    event followed by `EndOfChannel` proves the stream terminates at
    the kernel subscription boundary.

    Subscription timeout semantics:
    https://github.com/torvalds/linux/blob/master/include/uapi/linux/tipc.h

    '''
    class _TimeoutSocket:
        async def recv(self, size: int) -> bytes:
            return _pack_topology_event(TIPC_SUBSCR_TIMEOUT)

    async def main() -> None:
        tx, rx = trio.open_memory_channel(1)
        await _tipc._stream_name_events(
            _TimeoutSocket(),
            TRACTOR_STYPE,
            tx,
        )
        assert (await rx.receive()).kind == 'timeout'
        with pytest.raises(trio.EndOfChannel):
            await rx.receive()

    trio.run(main)


@requires_tipc
def test_topology_reports_publish_and_withdraw():
    '''
    "Publishing a bind IS registration" — but *observed from the
    outside*, by the kernel pushing us the name-table transition.

    This is the whole point of layer B: cluster-wide service
    (de)registration with NO registrar actor and NO polling.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        got: list = []

        async with open_topology_events(
            stype=addr._stype,
        ) as events:
            # publish..
            lstnr = await start_listener(addr=addr)
            with trio.fail_after(5):
                got.append(await events.receive())

            # ..then withdraw
            lstnr.socket.close()
            with trio.fail_after(5):
                got.append(await events.receive())

        kinds: list[str] = [ev.kind for ev in got]
        assert kinds == ['published', 'withdrawn']

        for ev in got:
            assert ev.addr._instance == addr._instance
            assert ev.addr._stype == addr._stype
            assert ev.addr._scope == TIPC_SCOPE_UNKNOWN
            assert isinstance(ev.ref, int)
        # both transitions name the SAME publisher port
        assert got[0].ref == got[1].ref
        assert 'published' in repr(got[0])

    trio.run(main)


@requires_tipc
def test_topology_acm_closes_cleanly():
    '''
    The `@acm` must tear down w/o leaking its reader task — exit
    closes the sock, the reader's `.recv()` raises
    `ClosedResourceError`, the nursery collapses.

    Also assert an *unconsumed* subscription doesn't wedge exit.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()
        with trio.fail_after(10):
            async with open_topology_events(stype=addr._stype):
                lstnr = await start_listener(addr=addr)
                # deliberately DON'T read the event
                await trio.sleep(0.2)
                lstnr.socket.close()

            # ..and a second open/close round still works
            async with open_topology_events(
                stype=addr._stype,
            ) as events:
                assert events is not None

    trio.run(main)


@requires_tipc
def test_topology_sub_ports_reports_each_publisher():
    '''
    `TIPC_SUB_PORTS` gives one event per *publisher*, so the
    duplicate-name/round-robin case (§2.3) is observable from the
    topology feed — which is how a push-registry would ever
    detect silent crosstalk.

    '''
    async def main():
        addr: TIPCAddress = TIPCAddress.get_random()

        async with open_topology_events(
            stype=addr._stype,
            filt=_tipc.TIPC_SUB_PORTS,
        ) as events:
            first = await start_listener(addr=addr)
            second = await start_listener(addr=addr)

            refs: set[int] = set()
            with trio.fail_after(5):
                for _ in range(2):
                    ev = await events.receive()
                    assert ev.kind == 'published'
                    assert ev.addr._instance == addr._instance
                    refs.add(ev.ref)

            # two DISTINCT publisher ports on one service name
            assert len(refs) == 2

            first.socket.close()
            second.socket.close()

    trio.run(main)