latest.ipv8.test.base

Module Contents

Classes

TranslatedMockEndpointListener

Automagically translate received packets into messages.

TestBase

Base TestCase to allow easy testing of IPv8 overlays.
class latest.ipv8.test.base.TranslatedMockEndpointListener(overlay: latest.ipv8.types.Community, message_classes: list[type[latest.ipv8.types.Payload]], message_filter: list[type[latest.ipv8.types.Payload]] | None, main_thread: bool = False)

Bases: latest.ipv8.test.mocking.endpoint.MockEndpointListener

Automagically translate received packets into messages.

The magic depends on three assumptions:

  • The requested overlay messages have a msg_id specified.

  • The requested overlay messages can be decoded using a single Payload.

  • The requested overlay messages follow the lazy_wrapper_* standard.

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

Callback for when a new packet is received by an endpoint.

class latest.ipv8.test.base.TestBase(methodName: str = 'runTest')

Bases: TestCaseClass, Generic[OT]

Base TestCase to allow easy testing of IPv8 overlays.

property __loop: asyncio.AbstractEventLoop

Return the asyncio event loop used for the test case.

__testing__ = True
__lockup_timestamp__ = 0
__asynctest_compatibility_mode__
MAX_TEST_TIME = 10
loop
__call_internal_in_context(func: Callable | Coroutine) → None

Call setUp or tearDown within the unittest.IsolatedAsyncioTestCase context.

This is not used for asynctest compatibility mode.

_callSetUp() → None
_callTearDown() → None
initialize(overlay_class: type[latest.ipv8.types.Community], node_count: int, settings: latest.ipv8.community.CommunitySettings | None = None, create_dht: bool = False, enable_statistics: bool = False) → None

Spawn a given number of nodes with a particular overlay class and introduce them to each other.

WARNING: Peers are directly added into each other’s Network. This skips sending messages (efficient!).

HOWEVER, custom logic based on introduction requests and responses may therefore be skipped.

_patch_overlay(overlay: latest.ipv8.overlay.Overlay) → None

Perform patches on an overlay for testing purposes.

By default, this patches out the general try-except in the packet handler, allowing crashes.

patch_overlays(i: int) → None

Method to make the packet handlers of a particular node fragile.

If you want to disable fragile packet handling for an entire class over overlays, add the class to the production_overlay_classes list.

If you want to disable fragile packet handling for a particular node, overwrite this method.

_cb_exception(loop: asyncio.AbstractEventLoop, context: dict) → None

Callback for asyncio exceptions.

Do not call self.fail or pytest.fail or any other failure mechanism in here. These methods work by raising an Exception, which will silently fail as this is not the main loop.

setUp() → None

Perform the base setup per unit test. Always call this before running your own setUp code!

async tearDown() → None

Tear down all the overlays, remove all files and check if any errors “escaped” the unit test scope. Always call this after your own tearDown!

classmethod setUpClass() → None

Perform the base setup for this entire test case. Always call this before running your own setUpClass code!

This code is responsible for detecting deadlocks in - and in between - tests. However, if you’re running pytest you should use the pytest-timeout plugin instead.

classmethod tearDownClass() → None

Signal that our test case completed without incident. Always call this after your own tearDownClass!

This - indirectly - stops setUpClass() from forcibly shutting down your tests.

create_node(settings: latest.ipv8.community.CommunitySettings | None = None, create_dht: bool = False, enable_statistics: bool = False) → latest.ipv8.test.mocking.ipv8.MockIPv8

Create a new IPv8-like container to store node related information.

add_node_to_experiment(node: latest.ipv8.test.mocking.ipv8.MockIPv8) → None

Add a new node to this experiment (use create_node()).

static is_background_task(task: asyncio.Task) → bool

Check if the given task is to be ignored.

async deliver_messages(timeout: float = 0.1) → None

Allow peers to communicate.

The strategy is as follows:

  1. Measure the amount of existing asyncio tasks

  2. After 10 milliseconds, check if we are below 2 tasks twice in a row

  3. If not, go back to handling calls (step 2) or return, if the timeout has been reached

Parameters:

timeout – the maximum time to wait for messages to be delivered

async introduce_nodes() → None

Force all known nodes to send IntroductionRequest messages to each other.

temporary_directory() → str

Create a directory that temporary files can be written to.

WARNING: If your test suite crashes, these files will be left behind (this is by design).

address(i: int) → latest.ipv8.types.Address

The endpoint address of node i.

endpoint(i: int) → latest.ipv8.types.Endpoint

The endpoint of node i.

key_bin(i: int) → bytes

The serialized public key material of node i.

key_bin_private(i: int) → bytes

The serialized PRIVATE key material of node i.

mid(i: int) → bytes

The SHA-1 encoding (20 bytes) of the public key of node 1.

my_peer(i: int) → latest.ipv8.peer.Peer

The PRIVATE peer instance of node i, containing the PRIVATE KEY.

When communicating peer info between peers, you should normally use peer(i) instead.

network(i: int) → latest.ipv8.peerdiscovery.network.Network

The Network instance of node i.

node(i: int) → latest.ipv8.test.mocking.ipv8.MockIPv8

The node identified by i.

overlay(i: int) → OT

The overlay belonging to node i.

peer(i: int) → latest.ipv8.peer.Peer

A public Peer object representing node i.

private_key(i: int) → latest.ipv8.types.PrivateKey

The PRIVATE key object of node i.

public_key(i: int) → latest.ipv8.types.PublicKey

The public key object of node i.

assertReceivedBy(i: int, message_classes: list[type[latest.ipv8.types.Payload]], ordered: bool = True, message_filter: list[type[latest.ipv8.types.Payload]] | None = None) → list[latest.ipv8.types.Payload]

Assert that a node i receives messages of a given type, potentially unordered or while ignoring other messages.

Parameters:

  • i – the node id that needs to receive messages

  • message_classes – the messages classes that you expect the node to receive

  • ordered – whether the message classes are received ordered or unordered

  • message_filter – what message classes to store (others are silently ignored)