latest.ipv8.test.REST.rest_base

Module Contents

Classes

IPv8Transport

Transport to route over the fake IPv8 internet instead of the actual Internet.

MockServer

Fake server to hijack TCP sessions and route them over the IPv8 fake internet.

MockedSite

Pretend to host a TCP site (actually UDP over fake IPv8 internet).

MockedRESTManager

Hook a fake site into the real rest manager.

MockRestIPv8

Manager for IPv8 related objects during REST tests.

MockConnector

Connector that routes over the fake IPv8 internet.

RESTTestBase

HTTP request superclass, which defines the common behavior between the different types of HTTP REST requests.

class latest.ipv8.test.REST.rest_base.IPv8Transport(callback: Callable[[bytes], None], host: str | None = None, port: int | None = None, server_port: int = 80)

Bases: asyncio.Transport, latest.ipv8.messaging.interfaces.endpoint.EndpointListener

Transport to route over the fake IPv8 internet instead of the actual Internet.

is_closing() bool

Whether we are in the process of shutting down.

is_reading() bool

Whether we are open.

close() None

Close our transport.

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

Callback for when the fake IPv8 internet sends us a message.

We forward this call to the registered callback.

write(data: bytes) None

Write to the other end of this transport.

class latest.ipv8.test.REST.rest_base.MockServer(server: aiohttp.web_server.Server)

Fake server to hijack TCP sessions and route them over the IPv8 fake internet.

START_PORT = 80
close() None

Close this fake server (no action required).

shutdown(timeout: float) asyncio.Future[bool]

Shut down this fake server (no action required).

received_data(data: bytes) None

Forward all received data to our handler.

class latest.ipv8.test.REST.rest_base.MockedSite(runner: BaseRunner, host: str | None = None, port: int | None = None, *, shutdown_timeout: float = 60.0, ssl_context: ssl.SSLContext | None = None, backlog: int = 128, reuse_address: bool | None = None, reuse_port: bool | None = None)

Bases: aiohttp.web.TCPSite

Pretend to host a TCP site (actually UDP over fake IPv8 internet).

async start() None

Start our nefarious rerouting.

async stop() None

The wait_for implementation of our super does not work.

Instead, we perform our own light-weight shutdown.

class latest.ipv8.test.REST.rest_base.MockedRESTManager(session: object, root_endpoint_class: type[latest.ipv8.REST.base_endpoint.BaseEndpoint] | None = None)

Bases: latest.ipv8.REST.rest_manager.RESTManager

Hook a fake site into the real rest manager.

async start_site(runner: aiohttp.web_runner.BaseRunner, host: str, port: int, ssl_context: ssl.SSLContext | None) None

Start accepting connections to the REST API.

class latest.ipv8.test.REST.rest_base.MockRestIPv8(crypto_curve: str, overlay_classes: list[type[latest.ipv8.types.Community]], settings: list[latest.ipv8.community.CommunitySettings])

Manager for IPv8 related objects during REST tests.

Note that this is not the same as an IPv8 instance, neither is it the same as a MockIPv8 instance! However, many of the same functionalities are offered.

get_overlay(overlay_cls: type[latest.ipv8.types.Community]) latest.ipv8.types.Community | None

Get any loaded overlay instance from a given class type, if it exists.

add_strategy(overlay: latest.ipv8.types.Community, strategy: latest.ipv8.peerdiscovery.discovery.DiscoveryStrategy, target_peers: int) None

Register a strategy to call every tick unless a target number of peers has been reached. If the target_peers is equal to -1, the strategy is always called.

unload_overlay(instance: latest.ipv8.types.Community) Awaitable

Unregister and unload a given community instance.

async produce_anonymized_endpoint() latest.ipv8.messaging.anonymization.endpoint.TunnelEndpoint

Create an anonymized endpoint.

async start_api() None

Start the REST API.

async stop() None

Stop serving the REST API.

class latest.ipv8.test.REST.rest_base.MockConnector(*, keepalive_timeout: object | None | float = sentinel, force_close: bool = False, limit: int = 100, limit_per_host: int = 0, enable_cleanup_closed: bool = False, loop: asyncio.AbstractEventLoop | None = None, timeout_ceil_threshold: float = 5)

Bases: aiohttp.BaseConnector

Connector that routes over the fake IPv8 internet.

async _create_connection(req: aiohttp.ClientRequest, traces: list[aiohttp.tracing.Trace], timeout: aiohttp.ClientTimeout) aiohttp.client_proto.ResponseHandler
class latest.ipv8.test.REST.rest_base.RESTTestBase(methodName: str = 'runTest')

Bases: latest.ipv8.test.base.TestBase

HTTP request superclass, which defines the common behavior between the different types of HTTP REST requests.

async initialize(overlay_classes: list[type[latest.ipv8.types.Community]], node_count: int, settings: list[latest.ipv8.community.CommunitySettings]) None

Initialize a given number of nodes with instances of the given Community classes.

async create_node(settings: list[latest.ipv8.community.CommunitySettings]) MockRestIPv8

Create a new MockRestIPv8 and start its REST API.

node(i: int) MockRestIPv8

MockRestIPv8 is not actually a MockIPv8. So, we bend the rules here a little bit.

async introduce_nodes() None

Have each node send an introduction request to each other node.

async make_request(node: MockRestIPv8, endpoint: str, request_type: str, arguments: dict[str, str] | None = None, json_response: bool = True, json: dict | None = None, expected_status: int = 200) Any

Forward an HTTP request of the specified type to a url, with the specified set of arguments.

Parameters:
  • node – the destination node

  • endpoint – the endpoint of this request (i.e. http://<interface>:<port>/<endpoint>)

  • request_type – the type of request (GET, POST, PUT, DELETE, etc.)

  • arguments – the arguments to be attached to the request. This should be a dictionary or None

  • json_response – whether the response is expected to be JSON

  • json – a JSON-serializable dictionary that is sent when making the request

  • expected_status – the status code returned in the response, defaults to 200

Returns:

a dictionary object with the response