latest.ipv8.test.REST.rest_base
Module Contents
Classes
Transport to route over the fake IPv8 internet instead of the actual Internet. |
|
Fake server to hijack TCP sessions and route them over the IPv8 fake internet. |
|
Pretend to host a TCP site (actually UDP over fake IPv8 internet). |
|
Hook a fake site into the real rest manager. |
|
Manager for IPv8 related objects during REST tests. |
|
Connector that routes over the fake IPv8 internet. |
|
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.
- 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
- shutdown(timeout: float) asyncio.Future[bool]
Shut down this fake server (no action required).
- 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).
- 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.
- 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.
- 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.
- 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 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