2.12.ipv8.test.REST.rest_base

Module Contents

Classes

class 2.12.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, 2.12.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[2.12.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 2.12.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 2.12.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.

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

Bases: 2.12.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 2.12.ipv8.test.REST.rest_base.MockRestIPv8(crypto_curve: str, overlay_classes: list[type[2.12.ipv8.types.Community]], settings: list[2.12.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[2.12.ipv8.types.Community]) → 2.12.ipv8.types.Community | None

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

add_strategy(overlay: 2.12.ipv8.types.Community, strategy: 2.12.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: 2.12.ipv8.types.Community) → Awaitable

Unregister and unload a given community instance.

async produce_anonymized_endpoint() → 2.12.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 2.12.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)

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 2.12.ipv8.test.REST.rest_base.RESTTestBase(methodName: str = 'runTest')

Bases: 2.12.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[2.12.ipv8.types.Community]], node_count: int, settings: list[2.12.ipv8.community.CommunitySettings]) → None

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

async create_node(settings: list[2.12.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