2.12.ipv8.test.base
Module Contents
Classes
- class 2.12.ipv8.test.base.TranslatedMockEndpointListener(overlay: 2.12.ipv8.types.Community, message_classes: list[type[2.12.ipv8.types.Payload]], message_filter: list[type[2.12.ipv8.types.Payload]] | None, main_thread: bool = False)
Bases:
2.12.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.
- class 2.12.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.
- initialize(overlay_class: type[2.12.ipv8.types.Community], node_count: int, settings: 2.12.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: 2.12.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 thepytest-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: 2.12.ipv8.community.CommunitySettings | None = None, create_dht: bool = False, enable_statistics: bool = False) 2.12.ipv8.test.mocking.ipv8.MockIPv8
Create a new IPv8-like container to store node related information.
- add_node_to_experiment(node: 2.12.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:
Measure the amount of existing asyncio tasks
After 10 milliseconds, check if we are below 2 tasks twice in a row
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).
- my_peer(i: int) 2.12.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) 2.12.ipv8.peerdiscovery.network.Network
The Network instance of node i.
- node(i: int) 2.12.ipv8.test.mocking.ipv8.MockIPv8
The node identified by i.
- peer(i: int) 2.12.ipv8.peer.Peer
A public Peer object representing node i.
- assertReceivedBy(i: int, message_classes: list[type[2.12.ipv8.types.Payload]], ordered: bool = True, message_filter: list[type[2.12.ipv8.types.Payload]] | None = None) list[2.12.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)