3.0.ipv8.test.base ================== .. py:module:: 3.0.ipv8.test.base Classes ------- .. autoapisummary:: 3.0.ipv8.test.base.TranslatedMockEndpointListener 3.0.ipv8.test.base.TestBase Module Contents --------------- .. py:class:: TranslatedMockEndpointListener(overlay: 3.0.ipv8.types.Community, message_classes: list[type[3.0.ipv8.types.Payload]], message_filter: list[type[3.0.ipv8.types.Payload]] | None, main_thread: bool = False) Bases: :py:obj:`3.0.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. .. py:attribute:: overlay .. py:attribute:: message_class_map .. py:attribute:: message_filter_set .. py:attribute:: received_messages :value: [] .. py:method:: on_packet(packet: tuple[3.0.ipv8.types.Address, bytes]) -> None Callback for when a new packet is received by an endpoint. .. py:class:: TestBase(methodName: str = 'runTest') Bases: :py:obj:`unittest.IsolatedAsyncioTestCase`, :py:obj:`Generic`\ [\ :py:obj:`OT`\ ] Base TestCase to allow easy testing of IPv8 overlays. .. py:attribute:: __testing__ :value: True .. py:attribute:: __lockup_timestamp__ :value: 0 .. py:attribute:: MAX_TEST_TIME :value: 10 .. py:attribute:: nodes :value: [] .. py:attribute:: overlay_class :type: type[OT] | None :value: None .. py:attribute:: _tempdirs :value: [] .. py:attribute:: production_overlay_classes :value: [] .. py:attribute:: _uncaught_async_failure :value: None .. py:property:: loop :type: asyncio.AbstractEventLoop Return the asyncio event loop used for the test case. .. py:method:: __call_internal_in_context(func: Callable | collections.abc.Coroutine) -> None Call setUp or tearDown within the unittest.IsolatedAsyncioTestCase context. .. py:method:: _callSetUp() -> None .. py:method:: _callTearDown() -> None .. py:method:: initialize(overlay_class: type[3.0.ipv8.types.Community], node_count: int, settings: 3.0.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. .. py:method:: _patch_overlay(overlay: 3.0.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. .. py:method:: 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. .. py: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. .. py:method:: setUp() -> None Perform the base setup per unit test. Always call this before running your own setUp code! .. py:method:: tearDown() -> None :async: 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! .. py:method:: setUpClass() -> None :classmethod: 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. .. py:method:: tearDownClass() -> None :classmethod: 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. .. py:method:: create_node(settings: 3.0.ipv8.community.CommunitySettings | None = None, create_dht: bool = False, enable_statistics: bool = False) -> 3.0.ipv8.test.mocking.ipv8.MockIPv8 Create a new IPv8-like container to store node related information. .. py:method:: add_node_to_experiment(node: 3.0.ipv8.test.mocking.ipv8.MockIPv8) -> None Add a new node to this experiment (use `create_node()`). .. py:method:: is_background_task(task: asyncio.Task) -> bool :staticmethod: Check if the given task is to be ignored. .. py:method:: deliver_messages(timeout: float = 0.1) -> None :async: 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 :param timeout: the maximum time to wait for messages to be delivered .. py:method:: introduce_nodes() -> None :async: Force all known nodes to send IntroductionRequest messages to each other. .. py:method:: 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). .. py:method:: address(i: int) -> 3.0.ipv8.types.Address The endpoint address of node i. .. py:method:: endpoint(i: int) -> 3.0.ipv8.types.Endpoint The endpoint of node i. .. py:method:: key_bin(i: int) -> bytes The serialized public key material of node i. .. py:method:: key_bin_private(i: int) -> bytes The serialized PRIVATE key material of node i. .. py:method:: mid(i: int) -> bytes The SHA-1 encoding (20 bytes) of the public key of node 1. .. py:method:: my_peer(i: int) -> 3.0.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. .. py:method:: network(i: int) -> 3.0.ipv8.peerdiscovery.network.Network The Network instance of node i. .. py:method:: node(i: int) -> 3.0.ipv8.test.mocking.ipv8.MockIPv8 The node identified by i. .. py:method:: overlay(i: int) -> OT The overlay belonging to node i. .. py:method:: peer(i: int) -> 3.0.ipv8.peer.Peer A public Peer object representing node i. .. py:method:: private_key(i: int) -> 3.0.ipv8.types.PrivateKey The PRIVATE key object of node i. .. py:method:: public_key(i: int) -> 3.0.ipv8.types.PublicKey The public key object of node i. .. py:method:: assertReceivedBy(i: int, message_classes: list[type[3.0.ipv8.types.Payload]], ordered: bool = True, message_filter: list[type[3.0.ipv8.types.Payload]] | None = None) -> list[3.0.ipv8.types.Payload] Assert that a node i receives messages of a given type, potentially unordered or while ignoring other messages. :param i: the node id that needs to receive messages :param message_classes: the messages classes that you expect the node to receive :param ordered: whether the message classes are received ordered or unordered :param message_filter: what message classes to store (others are silently ignored)