`latest.ipv8.test.mocking.endpoint`

Module Contents

Classes

`MockEndpoint`	Endpoint that registers an address in the "internet" dictionary instead of using The Internet.
`AddressTester`	Generating addresses that are on our physical machine's actual physical LAN can lead to issues.
`AutoMockEndpoint`	Randomly generate LAN + WAN addresses that are globally unique and register them in the "internet" dictionary.
`MockEndpointListener`	Listener that simply stores all data sent to it.

Functions

crash_event_loop(→ None)

Raise an exception on the event loop.

Attributes

internet

latest.ipv8.test.mocking.endpoint.internet

async latest.ipv8.test.mocking.endpoint.crash_event_loop(forwarded_exception: Exception) → None

Raise an exception on the event loop.

Parameters:: forwarded_exception – the exception instance to raise.

class latest.ipv8.test.mocking.endpoint.MockEndpoint(lan_address: latest.ipv8.types.Address, wan_address: latest.ipv8.types.Address)

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

Endpoint that registers an address in the “internet” dictionary instead of using The Internet.

SEND_INET_EXCEPTION_TO_LOOP = True: Raise an uncaught AssertionError on the asyncio event loop if attempting to send to an unknown address. Useful for use in defensively-programmed code: bypasses most exception handling.

assert_open() → None: Throw an assertion error if this endpoint is not open.

is_open() → bool: Check if this endpoint is open.

get_address() → latest.ipv8.types.Address: Get our own registered WAN address.

send(socket_address: latest.ipv8.types.Address, packet: bytes) → None

Route a message through the “internet” dictionary.

WARNING: We schedule a call on the event loop. Otherwise, you can easily create infinite loops!

open() → None: Set this endpoint to be open.

close(timeout: float = 0.0) → None: Close this endpoint.

reset_byte_counters() → None: Reset our byte counters (we have none).

class latest.ipv8.test.mocking.endpoint.AddressTester(endpoint: latest.ipv8.messaging.interfaces.endpoint.Endpoint)

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

Generating addresses that are on our physical machine’s actual physical LAN can lead to issues.

singleton

classmethod get_singleton(endpoint: latest.ipv8.messaging.interfaces.endpoint.Endpoint) → AddressTester: Create a singleton AddressTester, you only need one.

on_packet(packet: tuple[latest.ipv8.types.Address, bytes]) → None: This should never be called.

is_lan(address: str) → bool: Check if the given address is on our physical LAN.

class latest.ipv8.test.mocking.endpoint.AutoMockEndpoint

Bases: MockEndpoint

Randomly generate LAN + WAN addresses that are globally unique and register them in the “internet” dictionary.

IPV6_ADDRESSES

_generate_address() → latest.ipv8.messaging.interfaces.udp.endpoint.UDPv4Address | latest.ipv8.messaging.interfaces.udp.endpoint.UDPv6Address

_is_lan(address: latest.ipv8.types.Address) → bool: Avoid false positives for the actual machine’s lan.

_generate_unique_address() → latest.ipv8.types.Address

class latest.ipv8.test.mocking.endpoint.MockEndpointListener(endpoint: latest.ipv8.messaging.interfaces.endpoint.Endpoint, main_thread: bool = False)