3.0.ipv8.test.mocking.endpoint

Attributes

Classes

Functions

Module Contents

3.0.ipv8.test.mocking.endpoint.internet

async 3.0.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 3.0.ipv8.test.mocking.endpoint.MockEndpoint(lan_address: 3.0.ipv8.types.Address, wan_address: 3.0.ipv8.types.Address)

Bases: 3.0.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.

lan_address

wan_address

_port

_open = False

assert_open() → None

Throw an assertion error if this endpoint is not open.

is_open() → bool

Check if this endpoint is open.

get_address() → 3.0.ipv8.types.Address

Get our own registered WAN address.

send(socket_address: 3.0.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 3.0.ipv8.test.mocking.endpoint.AddressTester(endpoint: 3.0.ipv8.messaging.interfaces.endpoint.Endpoint)

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

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

singleton = None

classmethod get_singleton(endpoint: 3.0.ipv8.messaging.interfaces.endpoint.Endpoint) → AddressTester

Create a singleton AddressTester, you only need one.

on_packet(packet: tuple[3.0.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 3.0.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

_open = False

_port = 0

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

_is_lan(address: 3.0.ipv8.types.Address) → bool

Avoid false positives for the actual machine’s lan.

_generate_unique_address() → 3.0.ipv8.types.Address

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

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

Listener that simply stores all data sent to it.

received_packets = []

on_packet(packet: tuple[3.0.ipv8.types.Address, bytes]) → None

Callback for when packets are received: simply store them.