3.0.ipv8.test.messaging.interfaces.dispatcher.test_endpoint

Classes

Module Contents

class 3.0.ipv8.test.messaging.interfaces.dispatcher.test_endpoint.DummyEndpointListener(endpoint: 3.0.ipv8.messaging.interfaces.endpoint.Endpoint)

Bases: 3.0.ipv8.messaging.interfaces.endpoint.EndpointListener

This class simply listens on an endpoint and stores incoming packets in a list.

incoming = []

on_packet(packet: tuple[3.0.ipv8.types.Address, bytes]) → None: Callback for incoming packets.

class 3.0.ipv8.test.messaging.interfaces.dispatcher.test_endpoint.DummyEndpoint

Bases: 3.0.ipv8.messaging.interfaces.endpoint.Endpoint

Non-functional endpoint for manual staging and inspection.

opened = False

sent = []

bytes_up = 0

bytes_down = 0

assert_open() → None: Assert open state as usual.

is_open() → bool: Check if opened as usual.

get_address() → 3.0.ipv8.types.Address: Give a fake localhost IPv4 address.

send(socket_address: 3.0.ipv8.types.Address, packet: bytes) → None: Intercept and store sends without actually sending.

async open() → bool: Do a fake open.

close() → None: Close as usual.

reset_byte_counters() → None: Reset counters as usual.

notify_listeners(packet: tuple[3.0.ipv8.types.Address, bytes]) → None: Perform usual notification.

class 3.0.ipv8.test.messaging.interfaces.dispatcher.test_endpoint.TestDispatcherEndpoint(methodName: str = 'runTest')

Bases: 3.0.ipv8.test.base.TestBase

This class contains various tests for the DispatcherEndpoint.

RANDOM_DATA = b'data'

static _produce_dummy() → tuple[3.0.ipv8.messaging.interfaces.dispatcher.endpoint.DispatcherEndpoint, DummyEndpoint, DummyEndpointListener]

Async:

Create and open a DispatcherEndpoint, dispatching to a dummy endpoint and listener.

async test_initialize_no_interfaces() → None

Check if the DispatcherEndpoint can initialize and “send” without interfaces.

This is black-hole functionality and should not crash (though we can’t really assert anything here).

async test_dispatch_receive() → None: Check if packet reception is correctly propagated from children.

async test_dispatch_send() → None: Check if packet sending is correctly propagated to children.

async test_dispatch_send_specific() → None: Check if packet sending is correctly propagated to children, with specific interface.

async test_is_open() → None: Check if is_open is correctly propagated from children.

async test_remove_listener() → None: Check if remove_listener is correctly propagated to children.

async test_byte_counters() → None: Check if byte counters are correctly propagated from children.

async test_get_address() → None: Check if get_address is correctly propagated from children.

async test_get_address_specific() → None: Check if get_address is correctly propagated from children, with specific interface.

async test_add_prefix_listener() → None: Check if add_prefix_listener is correctly propagated to children.

async test_guess_interface_ipv4() → None: Check if guess_interface guesses IPv4 interfaces correctly.

async test_guess_interface_ipv6() → None: Check if guess_interface guesses IPv6 interfaces correctly.

async test_guess_interface_ipv6_short() → None: Check if guess_interface guesses shortened IPv6 interfaces correctly.

async test_guess_interface_ipv6_very_short() → None: Check if guess_interface guesses zero-omitted IPv6 interfaces correctly.

async test_guess_interface_ipv6_ipv4map() → None: Check if guess_interface guesses IPv4 mapped onto IPv6 interfaces correctly.

async test_guess_interface_unknown() → None: Check if guess_interface guesses None if the interface is invalid.