mirror of
https://github.com/lnbits/lnbits.git
synced 2026-08-19 13:27:39 +02:00
697 lines
23 KiB
Python
697 lines
23 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
|
|
from lnbits.core.wasm_ext.routes.api import (
|
|
WasmRequestBodyTooLargeError,
|
|
WasmRoutePayload,
|
|
_read_api_payload,
|
|
_read_json_object_with_size,
|
|
_wasm_extension_api_export,
|
|
_wasm_route_owner_id,
|
|
register_wasm_extension_api_routes,
|
|
unregister_wasm_extension_api_routes,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.assets import (
|
|
WASM_EXTENSION_STATIC_MIME_TYPES,
|
|
_reject_html_like_wasm_static_asset,
|
|
_wasm_extension_core_asset_response,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.register import _format_wasm_extension_size
|
|
from lnbits.core.wasm_ext.routes.security import (
|
|
consume_wasm_extension_frame_token,
|
|
wasm_extension_frame_csp,
|
|
wasm_extension_frame_url,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.ui import (
|
|
_match_wasm_extension_ui_route,
|
|
_wasm_extension_bridge_api_routes,
|
|
_wasm_extension_entrypoint,
|
|
register_wasm_extension_ui_routes,
|
|
)
|
|
from lnbits.core.wasm_ext.wasm.config import parse_wasm_extension_config
|
|
from lnbits.core.wasm_ext.wasm.loader import WasmExtension
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("size_bytes", "formatted_size"),
|
|
[
|
|
(128_235, "128.24 KB"),
|
|
(1_000_000, "1.00 MB"),
|
|
(12_823_598, "12.82 MB"),
|
|
],
|
|
)
|
|
def test_format_wasm_extension_size(size_bytes: int, formatted_size: str):
|
|
assert _format_wasm_extension_size(size_bytes) == formatted_size
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_json_reader_rejects_large_content_length_without_reading():
|
|
request = _FakeRequest([b"{}"], content_length="11")
|
|
|
|
with pytest.raises(WasmRequestBodyTooLargeError, match="11 bytes"):
|
|
await _read_json_object_with_size(cast(Request, request), max_body_bytes=10)
|
|
|
|
assert request.stream_started is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_json_reader_rejects_large_stream_without_content_length():
|
|
request = _FakeRequest([b'{"value":"', b"x" * 20, b'"}'])
|
|
|
|
with pytest.raises(WasmRequestBodyTooLargeError):
|
|
await _read_json_object_with_size(cast(Request, request), max_body_bytes=16)
|
|
|
|
assert request.stream_started is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_api_payload_records_actual_body_bytes():
|
|
body = b'{"amount":21}'
|
|
request = _FakeRequest(
|
|
[body],
|
|
path_params={"invoice_id": "abc"},
|
|
query_params={"include_paid": "true"},
|
|
)
|
|
|
|
payload = await _read_api_payload(
|
|
cast(Request, request),
|
|
{"invoice_id": "invoiceId"},
|
|
max_body_bytes=100,
|
|
)
|
|
|
|
assert payload.data == {
|
|
"invoiceId": "abc",
|
|
"includePaid": "true",
|
|
"amount": 21,
|
|
}
|
|
assert payload.request_bytes == len(body)
|
|
|
|
|
|
def test_wasm_api_export_visibility_is_enforced(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
|
|
assert _wasm_extension_api_export(extension, "render") == "render"
|
|
assert _wasm_extension_api_export(extension, "private_render") == "private_render"
|
|
with pytest.raises(PermissionError, match="not callable over HTTP"):
|
|
_wasm_extension_api_export(extension, "on_invoice_paid")
|
|
with pytest.raises(KeyError, match="has no export"):
|
|
_wasm_extension_api_export(extension, "missing")
|
|
|
|
|
|
def test_wasm_ui_entrypoint_rejects_escape_static_and_non_html(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
(tmp_path / "index.html").write_text("<html></html>", encoding="utf-8")
|
|
(tmp_path / "index.txt").write_text("text", encoding="utf-8")
|
|
(tmp_path / "static").mkdir()
|
|
(tmp_path / "static" / "index.html").write_text("<html></html>", encoding="utf-8")
|
|
|
|
assert (
|
|
_wasm_extension_entrypoint(extension, "index.html")
|
|
== (tmp_path / "index.html").resolve()
|
|
)
|
|
with pytest.raises(ValueError, match="escapes extension root"):
|
|
_wasm_extension_entrypoint(extension, "../outside.html")
|
|
with pytest.raises(ValueError, match="must not be inside the static"):
|
|
_wasm_extension_entrypoint(extension, "static/index.html")
|
|
with pytest.raises(ValueError, match="must be an HTML file"):
|
|
_wasm_extension_entrypoint(extension, "index.txt")
|
|
|
|
|
|
def test_wasm_frame_token_is_one_time_and_user_bound(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
frame_path = "/ext-frame/demoext/0"
|
|
frame_url = wasm_extension_frame_url(extension, frame_path, "user-1")
|
|
token = frame_url.split("frame_token=", 1)[1]
|
|
|
|
with pytest.raises(HTTPException) as wrong_user:
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-2",
|
|
)
|
|
assert wrong_user.value.status_code == 404
|
|
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-1",
|
|
)
|
|
with pytest.raises(HTTPException) as reused:
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-1",
|
|
)
|
|
assert reused.value.status_code == 404
|
|
|
|
|
|
def test_wasm_frame_csp_is_locked_to_extension_assets(tmp_path: Path):
|
|
csp = wasm_extension_frame_csp(
|
|
_request_with_query("token"),
|
|
_wasm_extension(tmp_path),
|
|
)
|
|
|
|
assert "sandbox allow-scripts" in csp
|
|
assert "default-src 'none'" in csp
|
|
assert "connect-src 'none'" in csp
|
|
assert "frame-ancestors 'self'" in csp
|
|
assert "http://testserver/ext-assets/demoext/" in csp
|
|
|
|
|
|
def test_wasm_ui_route_matching_and_bridge_public_api_filtering(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
|
|
matched = _match_wasm_extension_ui_route(extension, "/ext/demo/abc")
|
|
public_routes = _wasm_extension_bridge_api_routes(extension, public=True)
|
|
private_routes = _wasm_extension_bridge_api_routes(extension, public=False)
|
|
|
|
assert matched["auth"] == "user"
|
|
assert matched["route_params"] == {"item_id": "abc"}
|
|
assert public_routes == [
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/ext/demoext/public/{item_id}",
|
|
"pattern": "^/api/v1/ext/demoext/public/[^/]+$",
|
|
}
|
|
]
|
|
assert {route["path"] for route in private_routes} == {
|
|
"/api/v1/ext/demoext/public/{item_id}",
|
|
"/api/v1/ext/demoext/private/{item_id}",
|
|
}
|
|
|
|
|
|
def test_wasm_api_routes_are_included_in_openapi(tmp_path: Path):
|
|
app = FastAPI()
|
|
app.openapi_schema = {"stale": True}
|
|
|
|
register_wasm_extension_api_routes(app, _wasm_extension(tmp_path))
|
|
|
|
assert app.openapi_schema is None
|
|
schema = app.openapi()
|
|
assert "/api/v1/ext/demoext/public/{item_id}" in schema["paths"]
|
|
assert "/api/v1/ext/demoext/private/{item_id}" in schema["paths"]
|
|
assert schema["paths"]["/api/v1/ext/demoext/public/{item_id}"]["get"]["tags"] == [
|
|
"Demo"
|
|
]
|
|
assert schema["paths"]["/api/v1/ext/demoext/private/{item_id}"]["post"]["tags"] == [
|
|
"Demo"
|
|
]
|
|
|
|
|
|
def test_wasm_api_routes_load_openapi_operation_fragment(tmp_path: Path):
|
|
app = FastAPI()
|
|
openapi_dir = tmp_path / "wasm"
|
|
openapi_dir.mkdir()
|
|
(openapi_dir / "openapi.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schemas": {
|
|
"DemoItem": {
|
|
"type": "object",
|
|
"required": ["id", "name"],
|
|
"properties": {
|
|
"id": {"type": "string"},
|
|
"name": {"type": "string"},
|
|
},
|
|
}
|
|
},
|
|
"routes": {
|
|
"list_demo_items": {
|
|
"summary": "List demo items",
|
|
"description": "Returns demo items.",
|
|
"operationId": "demoext_list_demo_items",
|
|
"tags": ["Ignored"],
|
|
"responses": {
|
|
"200": {
|
|
"description": "Demo item list",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"type": "object",
|
|
"properties": {
|
|
"items": {
|
|
"type": "array",
|
|
"items": {
|
|
"$ref": "#/schemas/DemoItem"
|
|
},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
register_wasm_extension_api_routes(
|
|
app,
|
|
_wasm_extension(
|
|
tmp_path,
|
|
openapi="wasm/openapi.json",
|
|
extra_exports=[{"name": "list-demo-items", "visibility": "public"}],
|
|
api_routes=[
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "list-demo-items",
|
|
"auth": "public",
|
|
}
|
|
],
|
|
),
|
|
)
|
|
|
|
operation = app.openapi()["paths"]["/api/v1/ext/demoext/public/{item_id}"]["get"]
|
|
item_schema = operation["responses"]["200"]["content"]["application/json"][
|
|
"schema"
|
|
]["properties"]["items"]["items"]
|
|
assert operation["summary"] == "List demo items"
|
|
assert operation["description"] == "Returns demo items."
|
|
assert operation["operationId"] == "demoext_list_demo_items"
|
|
assert operation["tags"] == ["Demo"]
|
|
assert item_schema["properties"]["name"] == {"type": "string"}
|
|
|
|
|
|
def test_wasm_api_routes_allow_route_openapi_fragment_override(tmp_path: Path):
|
|
app = FastAPI()
|
|
openapi_dir = tmp_path / "wasm"
|
|
openapi_dir.mkdir()
|
|
(openapi_dir / "openapi.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"routes": {
|
|
"render": {
|
|
"summary": "Default render docs",
|
|
"operationId": "demoext_render",
|
|
},
|
|
"custom-render": {
|
|
"summary": "Custom render docs",
|
|
"operationId": "demoext_custom_render",
|
|
},
|
|
}
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
register_wasm_extension_api_routes(
|
|
app,
|
|
_wasm_extension(
|
|
tmp_path,
|
|
openapi="wasm/openapi.json",
|
|
api_routes=[
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "render",
|
|
"auth": "public",
|
|
"openapi": "#/routes/custom-render",
|
|
}
|
|
],
|
|
),
|
|
)
|
|
|
|
operation = app.openapi()["paths"]["/api/v1/ext/demoext/public/{item_id}"]["get"]
|
|
assert operation["summary"] == "Custom render docs"
|
|
assert operation["operationId"] == "demoext_custom_render"
|
|
|
|
|
|
def test_wasm_api_routes_add_success_response_example_for_redoc(tmp_path: Path):
|
|
app = FastAPI()
|
|
openapi_dir = tmp_path / "wasm"
|
|
openapi_dir.mkdir()
|
|
(openapi_dir / "openapi.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"schemas": {
|
|
"DemoItem": {
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "string"},
|
|
"count": {"type": "integer"},
|
|
},
|
|
},
|
|
"DemoResponse": {
|
|
"type": "object",
|
|
"required": ["ok", "data"],
|
|
"properties": {
|
|
"ok": {"type": "boolean", "enum": [True]},
|
|
"data": {
|
|
"type": "object",
|
|
"properties": {
|
|
"items": {
|
|
"type": "array",
|
|
"items": {"$ref": "#/schemas/DemoItem"},
|
|
}
|
|
},
|
|
},
|
|
},
|
|
},
|
|
"ErrorResponse": {
|
|
"type": "object",
|
|
"required": ["ok", "error"],
|
|
"properties": {
|
|
"ok": {"type": "boolean", "enum": [False]},
|
|
"error": {"type": "string"},
|
|
},
|
|
},
|
|
},
|
|
"routes": {
|
|
"render": {
|
|
"summary": "Render",
|
|
"responses": {
|
|
"200": {
|
|
"description": "Success or extension-level error.",
|
|
"content": {
|
|
"application/json": {
|
|
"schema": {
|
|
"oneOf": [
|
|
{"$ref": "#/schemas/DemoResponse"},
|
|
{"$ref": "#/schemas/ErrorResponse"},
|
|
]
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
register_wasm_extension_api_routes(
|
|
app,
|
|
_wasm_extension(
|
|
tmp_path,
|
|
openapi="wasm/openapi.json",
|
|
api_routes=[
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "render",
|
|
"auth": "public",
|
|
}
|
|
],
|
|
),
|
|
)
|
|
|
|
json_content = app.openapi()["paths"]["/api/v1/ext/demoext/public/{item_id}"][
|
|
"get"
|
|
]["responses"]["200"]["content"]["application/json"]
|
|
assert json_content["example"] == {
|
|
"ok": True,
|
|
"data": {"items": [{"id": "string", "count": 0}]},
|
|
}
|
|
|
|
|
|
def test_wasm_api_routes_ignore_missing_openapi_fragment(tmp_path: Path):
|
|
app = FastAPI()
|
|
|
|
register_wasm_extension_api_routes(
|
|
app,
|
|
_wasm_extension(
|
|
tmp_path,
|
|
api_routes=[
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "render",
|
|
"auth": "public",
|
|
"openapi": "wasm/missing.json#/routes/list_demo_items",
|
|
}
|
|
],
|
|
),
|
|
)
|
|
|
|
operation = app.openapi()["paths"]["/api/v1/ext/demoext/public/{item_id}"]["get"]
|
|
assert operation["summary"] == "GET /public/{item_id}"
|
|
assert operation["operationId"] == "demoext_get_public_item_id"
|
|
|
|
|
|
def test_wasm_api_routes_replace_same_extension_routes_on_upgrade(tmp_path: Path):
|
|
app = FastAPI()
|
|
route_path = "/api/v1/ext/demoext/public/{item_id}"
|
|
|
|
async def legacy_handler() -> dict[str, bool]:
|
|
return {"legacy": True}
|
|
|
|
app.add_api_route(
|
|
route_path,
|
|
legacy_handler,
|
|
methods=["GET"],
|
|
name=f"demoext:GET:{route_path}",
|
|
include_in_schema=False,
|
|
)
|
|
app.openapi_schema = {"stale": True}
|
|
|
|
register_wasm_extension_api_routes(app, _wasm_extension(tmp_path))
|
|
|
|
routes = _matching_routes(app, route_path, "GET")
|
|
assert len(routes) == 1
|
|
assert routes[0].endpoint != legacy_handler
|
|
assert routes[0].include_in_schema is True
|
|
assert app.openapi_schema is None
|
|
assert route_path in app.openapi()["paths"]
|
|
|
|
|
|
def test_wasm_api_routes_remove_obsolete_routes_on_upgrade(tmp_path: Path):
|
|
app = FastAPI()
|
|
route_path = "/api/v1/ext/demoext/removed"
|
|
|
|
async def removed_handler() -> dict[str, bool]:
|
|
return {"removed": True}
|
|
|
|
app.add_api_route(
|
|
route_path,
|
|
removed_handler,
|
|
methods=["GET"],
|
|
name=f"demoext:GET:{route_path}",
|
|
)
|
|
app.openapi_schema = {"stale": True}
|
|
|
|
register_wasm_extension_api_routes(app, _wasm_extension(tmp_path))
|
|
|
|
assert _matching_routes(app, route_path, "GET") == []
|
|
assert app.openapi_schema is None
|
|
assert route_path not in app.openapi()["paths"]
|
|
|
|
|
|
def test_wasm_api_route_cleanup_preserves_ui_frame_config_route(tmp_path: Path):
|
|
app = FastAPI()
|
|
(tmp_path / "index.html").write_text("<html></html>", encoding="utf-8")
|
|
extension = _wasm_extension(tmp_path)
|
|
frame_config_path = "/api/v1/ext/demoext/_ui/frame"
|
|
api_route_path = "/api/v1/ext/demoext/public/{item_id}"
|
|
|
|
register_wasm_extension_ui_routes(app, extension)
|
|
assert _matching_routes(app, frame_config_path, "POST") != []
|
|
|
|
register_wasm_extension_api_routes(app, extension)
|
|
|
|
assert _matching_routes(app, frame_config_path, "POST") != []
|
|
assert _matching_routes(app, api_route_path, "GET") != []
|
|
|
|
assert unregister_wasm_extension_api_routes(app, "demoext") is True
|
|
assert _matching_routes(app, api_route_path, "GET") == []
|
|
assert _matching_routes(app, frame_config_path, "POST") != []
|
|
|
|
|
|
def test_wasm_api_routes_are_removed_from_openapi_on_uninstall(tmp_path: Path):
|
|
app = FastAPI()
|
|
route_path = "/api/v1/ext/demoext/public/{item_id}"
|
|
|
|
register_wasm_extension_api_routes(app, _wasm_extension(tmp_path))
|
|
assert route_path in app.openapi()["paths"]
|
|
|
|
assert unregister_wasm_extension_api_routes(app, "demoext") is True
|
|
|
|
assert app.openapi_schema is None
|
|
assert _matching_routes(app, route_path, "GET") == []
|
|
assert route_path not in app.openapi()["paths"]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_api_route_owner_context_uses_configured_storage_row(
|
|
tmp_path: Path, mocker
|
|
):
|
|
extension = _wasm_extension(tmp_path)
|
|
route_config = parse_wasm_extension_config(
|
|
"demoext",
|
|
{
|
|
"id": "demoext",
|
|
"name": "Demo",
|
|
"short_description": "Demo extension",
|
|
"version": "1.0.0",
|
|
"extension_type": "wasm",
|
|
"wasm": {
|
|
"module": "extension.wasm",
|
|
"exports": [{"name": "finish", "visibility": "public"}],
|
|
},
|
|
"api_routes": [
|
|
{
|
|
"method": "POST",
|
|
"path": "/games/{game_id}/finish",
|
|
"export": "finish",
|
|
"auth": "public",
|
|
"path_params": {"game_id": "gameId"},
|
|
"ownerContext": {"table": "games", "idParam": "gameId"},
|
|
}
|
|
],
|
|
},
|
|
).api_routes[0]
|
|
owner_lookup = mocker.patch(
|
|
"lnbits.core.wasm_ext.routes.api.storage_get_row_owner_id",
|
|
mocker.AsyncMock(return_value="owner-1"),
|
|
)
|
|
|
|
owner_id = await _wasm_route_owner_id(
|
|
extension,
|
|
route_config,
|
|
WasmRoutePayload({"gameId": "game-1"}, request_bytes=10),
|
|
)
|
|
|
|
assert owner_id == "owner-1"
|
|
owner_lookup.assert_awaited_once_with("demoext", "games", "game-1")
|
|
|
|
|
|
def test_wasm_static_core_assets_and_html_like_text_assets_are_guarded(tmp_path: Path):
|
|
response = _wasm_extension_core_asset_response("_lnbits/material-icons.css")
|
|
assert response.headers["X-Content-Type-Options"] == "nosniff"
|
|
assert response.headers["Cache-Control"] == "no-store"
|
|
assert WASM_EXTENSION_STATIC_MIME_TYPES[".ogg"] == "audio/ogg"
|
|
|
|
for path in ["_lnbits/../bundle.min.css", "_lnbits/missing.css"]:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_wasm_extension_core_asset_response(path)
|
|
assert exc_info.value.status_code == 404
|
|
|
|
script_path = tmp_path / "app.js"
|
|
script_path.write_text("<script>alert(1)</script>", encoding="utf-8")
|
|
with pytest.raises(HTTPException) as html_like:
|
|
_reject_html_like_wasm_static_asset(script_path)
|
|
assert html_like.value.status_code == 404
|
|
|
|
|
|
class _FakeRequest:
|
|
method = "POST"
|
|
|
|
def __init__(
|
|
self,
|
|
chunks: list[bytes],
|
|
*,
|
|
content_length: str | None = None,
|
|
path_params: dict[str, str] | None = None,
|
|
query_params: dict[str, str] | None = None,
|
|
) -> None:
|
|
self._chunks = chunks
|
|
self.headers: dict[str, str] = {}
|
|
if content_length is not None:
|
|
self.headers["content-length"] = content_length
|
|
self.path_params = path_params or {}
|
|
self.query_params = query_params or {}
|
|
self.stream_started = False
|
|
|
|
async def stream(self) -> AsyncIterator[bytes]:
|
|
self.stream_started = True
|
|
for chunk in self._chunks:
|
|
yield chunk
|
|
|
|
|
|
def _wasm_extension(
|
|
root_path: Path,
|
|
*,
|
|
api_routes: list[dict] | None = None,
|
|
openapi: str | None = None,
|
|
extra_exports: list[dict] | None = None,
|
|
) -> WasmExtension:
|
|
extension_config = {
|
|
"id": "demoext",
|
|
"name": "Demo",
|
|
"short_description": "Demo extension",
|
|
"version": "1.0.0",
|
|
"extension_type": "wasm",
|
|
"wasm": {
|
|
"module": "extension.wasm",
|
|
"exports": [
|
|
{"name": "render", "visibility": "public"},
|
|
{"name": "private_render", "visibility": "authenticated"},
|
|
{"name": "on_invoice_paid", "visibility": "event"},
|
|
*(extra_exports or []),
|
|
],
|
|
},
|
|
"ui_routes": [
|
|
{
|
|
"path": "/demo/{item_id}",
|
|
"entrypoint": "index.html",
|
|
"auth": "user",
|
|
}
|
|
],
|
|
"api_routes": api_routes
|
|
or [
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "render",
|
|
"auth": "public",
|
|
},
|
|
{
|
|
"method": "POST",
|
|
"path": "/private/{item_id}",
|
|
"export": "private_render",
|
|
"auth": "user",
|
|
},
|
|
],
|
|
}
|
|
if openapi:
|
|
extension_config["openapi"] = openapi
|
|
config = parse_wasm_extension_config("demoext", extension_config)
|
|
return WasmExtension(
|
|
id="demoext",
|
|
name="Demo",
|
|
version="1.0.0",
|
|
root_path=root_path,
|
|
module_path=root_path / "extension.wasm",
|
|
wit_path=None,
|
|
world="",
|
|
exports=config.wasm.exports,
|
|
config=config,
|
|
)
|
|
|
|
|
|
def _request_with_query(token: str) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "GET",
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"path": "/ext-frame/demoext/0",
|
|
"query_string": f"frame_token={token}".encode(),
|
|
"headers": [],
|
|
}
|
|
)
|
|
|
|
|
|
def _matching_routes(app: FastAPI, route_path: str, method: str) -> list:
|
|
return [
|
|
route
|
|
for route in app.router.routes
|
|
if getattr(route, "path", None) == route_path
|
|
and method in (getattr(route, "methods", set()) or set())
|
|
]
|