102 lines
2.9 KiB
Python
102 lines
2.9 KiB
Python
 
|
"""B-67: /qr/{address} must validate the address for real and must not render
|
||
|
|
QR codes on the event loop for every anonymous request."""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from httpx import ASGITransport, AsyncClient
|
||
|
|
|
||
|
|
from app.api.routes import qr
|
||
|
|
|
||
|
|
|
||
|
|
VALID_ADDRESS = "plm1q0xcqpzrky6eff2g52qdye53xkk9jxkvraxkkwd"
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def client():
|
||
|
|
qr._render_png.cache_clear()
|
||
|
|
app = FastAPI()
|
||
|
|
app.include_router(qr.router)
|
||
|
|
transport = ASGITransport(app=app)
|
||
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
|
|
yield ac
|
||
|
|
|
||
|
|
|
||
|
|
async def test_valid_address_renders_a_png(client):
|
||
|
|
resp = await client.get(f"/qr/{VALID_ADDRESS}")
|
||
|
|
assert resp.status_code == 200
|
||
|
|
assert resp.headers["content-type"] == "image/png"
|
||
|
|
assert resp.content.startswith(b"\x89PNG")
|
||
|
|
assert "max-age" in resp.headers["cache-control"]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize(
|
||
|
|
"address",
|
||
|
|
[
|
||
|
|
"plm1qbogus0000000000000000000000000000000000", # right shape, broken checksum
|
||
|
|
"bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4", # valid bech32, wrong chain
|
||
|
|
"plm1q", # too short to be anything
|
||
|
|
"P" + "a" * 40, # not bech32 at all
|
||
|
|
"plm1" + "q" * 200, # over the length guard
|
||
|
|
],
|
||
|
|
)
|
||
|
|
async def test_non_addresses_are_rejected_without_rendering(client, address, monkeypatch):
|
||
|
|
def explode(*args, **kwargs): # pragma: no cover - must never run
|
||
|
|
raise AssertionError("rendered a QR for a non-address")
|
||
|
|
|
||
|
|
monkeypatch.setattr(qr.qrcode, "make", explode)
|
||
|
|
|
||
|
|
resp = await client.get(f"/qr/{address}")
|
||
|
|
assert resp.status_code == 400
|
||
|
|
assert resp.json()["detail"]["code"] == "invalid_address"
|
||
|
|
|
||
|
|
|
||
|
|
async def test_render_is_memoized_per_address(client):
|
||
|
|
calls = 0
|
||
|
|
original = qr.qrcode.make
|
||
|
|
|
||
|
|
def counting_make(data, *args, **kwargs):
|
||
|
|
nonlocal calls
|
||
|
|
calls += 1
|
||
|
|
return original(data, *args, **kwargs)
|
||
|
|
|
||
|
|
qr.qrcode.make = counting_make
|
||
|
|
try:
|
||
|
|
for _ in range(3):
|
||
|
|
assert (await client.get(f"/qr/{VALID_ADDRESS}")).status_code == 200
|
||
|
|
finally:
|
||
|
|
qr.qrcode.make = original
|
||
|
|
|
||
|
|
assert calls == 1
|
||
|
|
|
||
|
|
|
||
|
|
async def test_render_does_not_block_the_event_loop(client):
|
||
|
|
"""The render runs in a threadpool, so the loop stays responsive while it does."""
|
||
|
|
ticks = 0
|
||
|
|
|
||
|
|
async def ticker():
|
||
|
|
nonlocal ticks
|
||
|
|
while True:
|
||
|
|
ticks += 1
|
||
|
|
await asyncio.sleep(0)
|
||
|
|
|
||
|
|
original = qr.qrcode.make
|
||
|
|
|
||
|
|
def slow_make(data, *args, **kwargs):
|
||
|
|
# Blocking sleep: on the event loop this would freeze the ticker.
|
||
|
|
import time
|
||
|
|
|
||
|
|
time.sleep(0.05)
|
||
|
|
return original(data, *args, **kwargs)
|
||
|
|
|
||
|
|
qr.qrcode.make = slow_make
|
||
|
|
task = asyncio.create_task(ticker())
|
||
|
|
try:
|
||
|
|
assert (await client.get(f"/qr/{VALID_ADDRESS}")).status_code == 200
|
||
|
|
finally:
|
||
|
|
qr.qrcode.make = original
|
||
|
|
task.cancel()
|
||
|
|
|
||
|
|
assert ticks > 1
|