ilia
480e5b6ce6
Read-only demo boards, per-IP rate limits, opaque voter sessions, session leave, upload sniffing, path containment, and retire /v1.
300 lines
11 KiB
Python
300 lines
11 KiB
Python
"""API tests for Stork MVP."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Minimal EBML/WebM header so upload sniffing accepts the fixture.
|
|
WEBM_BYTES = b"\x1a\x45\xdf\xa3" + b"\x00" * 32
|
|
|
|
|
|
def test_health(client: TestClient) -> None:
|
|
res = client.get("/api/health")
|
|
assert res.status_code == 200
|
|
assert res.json()["ok"] is True
|
|
|
|
|
|
def test_logos_gallery(client: TestClient) -> None:
|
|
res = client.get("/logos")
|
|
assert res.status_code == 200
|
|
assert b"Pick a Stork mark" in res.content
|
|
|
|
|
|
def test_home_is_paper_board(client: TestClient) -> None:
|
|
res = client.get("/")
|
|
assert res.status_code == 200
|
|
assert b"Start a private board" in res.content
|
|
|
|
|
|
def test_v2_redirects_home(client: TestClient) -> None:
|
|
res = client.get("/v2", follow_redirects=False)
|
|
assert res.status_code == 302
|
|
assert res.headers["location"] == "/"
|
|
|
|
|
|
def test_v1_redirects_home(client: TestClient) -> None:
|
|
res = client.get("/v1", follow_redirects=False)
|
|
assert res.status_code == 302
|
|
assert res.headers["location"] == "/"
|
|
|
|
|
|
def test_public_create_board_and_share_url(client: TestClient) -> None:
|
|
created = client.post("/api/boards", json={"title": "Baby Levkin"})
|
|
assert created.status_code == 200
|
|
board_id = created.json()["id"]
|
|
assert board_id.startswith("b_")
|
|
assert created.json()["url"].endswith(f"/b/{board_id}")
|
|
assert created.json()["readonly"] is False
|
|
page = client.get(f"/b/{board_id}")
|
|
assert page.status_code == 200
|
|
assert b"Start a private board" in page.content or b"Your name" in page.content
|
|
|
|
|
|
def test_join_board_with_display_name(client: TestClient) -> None:
|
|
created = client.post("/api/boards", json={"title": "Friends"}).json()
|
|
session = client.post(
|
|
"/api/session",
|
|
json={"board_id": created["id"], "display_name": "Dan"},
|
|
)
|
|
assert session.status_code == 200
|
|
assert session.json()["board"]["id"] == created["id"]
|
|
assert client.post("/api/names", json={"kind": "c1", "spelling": "Noa"}).status_code == 200
|
|
|
|
|
|
def test_boards_are_isolated(client: TestClient) -> None:
|
|
a = client.post("/api/boards", json={"title": "A"}).json()
|
|
b = client.post("/api/boards", json={"title": "B"}).json()
|
|
client.post("/api/session", json={"board_id": a["id"], "display_name": "Ilia"})
|
|
assert client.post("/api/names", json={"kind": "c1", "spelling": "Roze"}).status_code == 200
|
|
a_names = client.get("/api/names?kind=c1").json()["items"]
|
|
client.post("/api/session", json={"board_id": b["id"], "display_name": "Dan"})
|
|
assert client.post("/api/names", json={"kind": "c1", "spelling": "Roze"}).status_code == 200
|
|
b_names = client.get("/api/names?kind=c1").json()["items"]
|
|
assert len(a_names) == 1 and len(b_names) == 1
|
|
assert a_names[0]["id"] != b_names[0]["id"]
|
|
|
|
|
|
def test_default_one_column(authed: TestClient) -> None:
|
|
cols = authed.get("/api/columns").json()["items"]
|
|
assert len(cols) == 1
|
|
assert cols[0]["id"] == "c1"
|
|
assert cols[0]["label"] == "First"
|
|
|
|
|
|
def test_add_rename_columns(authed: TestClient) -> None:
|
|
c2 = authed.post("/api/columns")
|
|
assert c2.status_code == 200
|
|
assert c2.json()["id"] == "c2"
|
|
renamed = authed.patch("/api/columns/c2", json={"label": "Hebrew"})
|
|
assert renamed.status_code == 200
|
|
assert renamed.json()["label"] == "Hebrew"
|
|
assert authed.post("/api/names", json={"kind": "c2", "spelling": "שי"}).status_code == 200
|
|
|
|
|
|
def test_column_delete_by_member(authed: TestClient) -> None:
|
|
assert authed.post("/api/columns").status_code == 200
|
|
assert authed.delete("/api/columns/c2").status_code == 200
|
|
assert len(authed.get("/api/columns").json()["items"]) == 1
|
|
|
|
|
|
def test_cannot_delete_first_column(authed: TestClient) -> None:
|
|
assert authed.post("/api/columns").status_code == 200
|
|
assert authed.delete("/api/columns/c1").status_code == 400
|
|
assert len(authed.get("/api/columns").json()["items"]) == 2
|
|
|
|
|
|
def test_max_four_columns(authed: TestClient) -> None:
|
|
for _ in range(3):
|
|
assert authed.post("/api/columns").status_code == 200
|
|
assert authed.post("/api/columns").status_code == 400
|
|
|
|
|
|
def test_cannot_delete_last_column(authed: TestClient) -> None:
|
|
assert authed.delete("/api/columns/c1").status_code == 400
|
|
|
|
|
|
def test_recording_upload_and_play(authed: TestClient) -> None:
|
|
created = authed.post("/api/names", json={"kind": "c1", "spelling": "Roze"}).json()
|
|
name_id = created["id"]
|
|
files = {"file": ("voice.webm", WEBM_BYTES, "audio/webm")}
|
|
up = authed.post(f"/api/names/{name_id}/recording/en", files=files)
|
|
assert up.status_code == 200
|
|
assert up.json()["recordings"]["en"] is True
|
|
audio = authed.get(f"/api/names/{name_id}/recording/en")
|
|
assert audio.status_code == 200
|
|
assert audio.content == WEBM_BYTES
|
|
|
|
|
|
def test_recording_rejects_bad_magic(authed: TestClient) -> None:
|
|
created = authed.post("/api/names", json={"kind": "c1", "spelling": "BadAudio"}).json()
|
|
files = {"file": ("voice.webm", b"not-audio", "audio/webm")}
|
|
up = authed.post(f"/api/names/{created['id']}/recording/en", files=files)
|
|
assert up.status_code == 400
|
|
|
|
|
|
def test_session_rejects_missing_board(client: TestClient) -> None:
|
|
res = client.post("/api/session", json={"display_name": "Ilia", "board_id": "b_missing"})
|
|
assert res.status_code == 404
|
|
|
|
|
|
def test_names_require_session(client: TestClient) -> None:
|
|
res = client.get("/api/names")
|
|
assert res.status_code in (401, 404)
|
|
|
|
|
|
def test_header_name_alone_rejected(client: TestClient, family_board: dict) -> None:
|
|
headers = {
|
|
"X-Stork-Board": family_board["id"],
|
|
"X-Stork-Display-Name": "Seed Bot",
|
|
}
|
|
res = client.post(
|
|
"/api/names",
|
|
headers=headers,
|
|
json={"kind": "c1", "spelling": "HeaderOnly"},
|
|
)
|
|
assert res.status_code == 401
|
|
|
|
|
|
def test_header_auth_with_opaque_voter(client: TestClient, family_board: dict) -> None:
|
|
headers = {
|
|
"X-Stork-Board": family_board["id"],
|
|
"X-Stork-Display-Name": "Seed Bot",
|
|
"X-Stork-Voter": "opaque-voter-token-1",
|
|
}
|
|
res = client.post(
|
|
"/api/names",
|
|
headers=headers,
|
|
json={"kind": "c1", "spelling": "HeaderOnly"},
|
|
)
|
|
assert res.status_code == 200
|
|
listed = client.get("/api/names?kind=c1", headers=headers)
|
|
assert any(n["spelling"] == "HeaderOnly" for n in listed.json()["items"])
|
|
|
|
|
|
def test_clear_session(client: TestClient, family_board: dict) -> None:
|
|
client.post(
|
|
"/api/session",
|
|
json={"board_id": family_board["id"], "display_name": "Ilia"},
|
|
)
|
|
assert client.get("/api/session").json()["authenticated"] is True
|
|
assert client.delete("/api/session").status_code == 200
|
|
assert client.get("/api/session").json()["authenticated"] is False
|
|
|
|
|
|
def test_add_vote_and_rank(authed: TestClient) -> None:
|
|
a = authed.post(
|
|
"/api/names",
|
|
json={
|
|
"kind": "c1",
|
|
"spelling": "Noa",
|
|
"locales": {
|
|
"he": {"pronunciation": "No-ah", "origin": "Hebrew", "meaning": "motion"},
|
|
"en": {"pronunciation": "NO-uh", "origin": "Hebrew", "meaning": "movement"},
|
|
},
|
|
},
|
|
)
|
|
assert a.status_code == 200
|
|
noa_id = a.json()["id"]
|
|
b = authed.post("/api/names", json={"kind": "c1", "spelling": "Levi"})
|
|
levi_id = b.json()["id"]
|
|
assert authed.post(f"/api/names/{noa_id}/vote", json={"value": 1}).status_code == 200
|
|
assert authed.post(f"/api/names/{levi_id}/vote", json={"value": -1}).status_code == 200
|
|
listed = authed.get("/api/names?kind=c1").json()["items"]
|
|
assert [n["spelling"] for n in listed] == ["Noa", "Levi"]
|
|
|
|
|
|
def test_duplicate_name_rejected(authed: TestClient) -> None:
|
|
assert authed.post("/api/names", json={"kind": "c1", "spelling": "Maya"}).status_code == 200
|
|
assert authed.post("/api/names", json={"kind": "c1", "spelling": "maya"}).status_code == 400
|
|
|
|
|
|
def test_patch_locales(authed: TestClient) -> None:
|
|
created = authed.post("/api/names", json={"kind": "c1", "spelling": "Eden"}).json()
|
|
patched = authed.patch(
|
|
f"/api/names/{created['id']}",
|
|
json={"locales": {"ru": {"pronunciation": "Э-ден", "origin": "иврит", "meaning": "рай"}}},
|
|
)
|
|
assert patched.status_code == 200
|
|
assert patched.json()["locales"]["ru"]["meaning"] == "рай"
|
|
assert patched.json()["locales"]["en"]["pronunciation"] == ""
|
|
|
|
|
|
def test_admin_delete_name(authed: TestClient) -> None:
|
|
created = authed.post("/api/names", json={"kind": "c1", "spelling": "Temp"}).json()
|
|
assert authed.delete(f"/api/names/{created['id']}").status_code == 403
|
|
assert (
|
|
authed.delete(
|
|
f"/api/names/{created['id']}", headers={"X-Stork-Admin": "test-admin-token"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
|
|
def test_legacy_invite_resolve(client: TestClient, family_board: dict) -> None:
|
|
res = client.get("/api/resolve", params={"invite": "test-invite-token"})
|
|
assert res.status_code == 200
|
|
assert res.json()["id"] == family_board["id"]
|
|
assert res.json()["readonly"] is False
|
|
|
|
|
|
def test_readonly_board_blocks_mutations(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("STORK_DATA", str(tmp_path / "data"))
|
|
monkeypatch.setenv("STORK_INVITE_TOKEN", "test-invite-token")
|
|
monkeypatch.setenv("STORK_ADMIN_TOKEN", "test-admin-token")
|
|
monkeypatch.setenv("STORK_COOKIE_SECURE", "false")
|
|
monkeypatch.setenv("STORK_READONLY_BOARD_IDS", "")
|
|
|
|
import stork.app as app_mod
|
|
|
|
importlib.reload(app_mod)
|
|
with TestClient(app_mod.app) as client:
|
|
board = client.post("/api/boards", json={"title": "Demo"}).json()
|
|
board_id = board["id"]
|
|
# Reload with this board marked readonly
|
|
monkeypatch.setenv("STORK_READONLY_BOARD_IDS", board_id)
|
|
importlib.reload(app_mod)
|
|
with TestClient(app_mod.app) as client2:
|
|
info = client2.get(f"/api/boards/{board_id}")
|
|
assert info.status_code == 200
|
|
assert info.json()["readonly"] is True
|
|
client2.post(
|
|
"/api/session",
|
|
json={"board_id": board_id, "display_name": "Visitor"},
|
|
)
|
|
assert (
|
|
client2.post("/api/names", json={"kind": "c1", "spelling": "X"}).status_code
|
|
== 403
|
|
)
|
|
assert client2.get("/api/names?kind=c1").status_code == 200
|
|
app_mod.store.close()
|
|
# close first store if still open
|
|
try:
|
|
app_mod.store.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def test_rate_limit_create_board(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("STORK_DATA", str(tmp_path / "data"))
|
|
monkeypatch.setenv("STORK_INVITE_TOKEN", "test-invite-token")
|
|
monkeypatch.setenv("STORK_ADMIN_TOKEN", "test-admin-token")
|
|
monkeypatch.setenv("STORK_COOKIE_SECURE", "false")
|
|
monkeypatch.setenv("STORK_RL_BOARDS_PER_HOUR", "2")
|
|
|
|
import stork.app as app_mod
|
|
import stork.rate_limit as rl
|
|
|
|
importlib.reload(rl)
|
|
importlib.reload(app_mod)
|
|
app_mod.limiter.reset()
|
|
app_mod.limiter.configure("create_board", 2, 3600)
|
|
with TestClient(app_mod.app) as client:
|
|
assert client.post("/api/boards", json={"title": "One"}).status_code == 200
|
|
assert client.post("/api/boards", json={"title": "Two"}).status_code == 200
|
|
third = client.post("/api/boards", json={"title": "Three"})
|
|
assert third.status_code == 429
|
|
assert "Retry-After" in third.headers
|
|
app_mod.store.close()
|