76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
import sqlite3
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.queue_priority_store import DEFAULT_QUEUE_ORDER, QueuePriorityStore
|
|
|
|
|
|
def test_queue_priority_is_revisioned_encrypted_and_account_scoped(tmp_path):
|
|
database = tmp_path / "queue-priority.sqlite3"
|
|
store = QueuePriorityStore(database, encryption_key=b"q" * 32)
|
|
preferred = list(DEFAULT_QUEUE_ORDER)
|
|
preferred.remove("following")
|
|
preferred.insert(1, "following")
|
|
|
|
created = store.replace(" Timmy ", 0, preferred)
|
|
|
|
assert created == {"revision": 1, "order": preferred}
|
|
assert QueuePriorityStore(database, encryption_key=b"q" * 32).get("timmy") == created
|
|
assert store.get("alexander") == {"revision": 0, "order": list(DEFAULT_QUEUE_ORDER)}
|
|
with sqlite3.connect(database) as connection:
|
|
payload = connection.execute(
|
|
"SELECT queue_order FROM queue_priorities WHERE login = 'timmy'"
|
|
).fetchone()[0]
|
|
assert payload.startswith("v1:")
|
|
assert "following" not in payload
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_queue_priority_api_is_authenticated_csrf_protected_no_store_and_conflict_safe(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple")
|
|
monkeypatch.setenv(
|
|
"STACKCHAIN_DASHBOARD_SESSION_SECRET",
|
|
"a-separate-session-signing-secret-with-enough-entropy",
|
|
)
|
|
monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_QUEUE_PRIORITY_DB", str(tmp_path / "queue-priority.sqlite3"))
|
|
|
|
async def user():
|
|
return {"id": 1, "login": "Timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
preferred = list(DEFAULT_QUEUE_ORDER)
|
|
preferred.remove("following")
|
|
preferred.insert(1, "following")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
await client.post("/api/v1/session", json={"access_token": "correct horse battery staple"})
|
|
forbidden = await client.put(
|
|
"/api/v1/queue-priority", json={"revision": 0, "order": preferred}
|
|
)
|
|
headers = {"Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"]}
|
|
saved = await client.put(
|
|
"/api/v1/queue-priority", json={"revision": 0, "order": preferred}, headers=headers
|
|
)
|
|
stale = await client.put(
|
|
"/api/v1/queue-priority",
|
|
json={"revision": 0, "order": list(DEFAULT_QUEUE_ORDER)},
|
|
headers=headers,
|
|
)
|
|
fetched = await client.get("/api/v1/queue-priority")
|
|
|
|
assert forbidden.status_code == 403
|
|
assert saved.status_code == 200
|
|
assert saved.json() == {"revision": 1, "order": preferred}
|
|
assert stale.status_code == 409
|
|
assert stale.json()["detail"] == {
|
|
"message": "Queue priority changed on another device.",
|
|
"snapshot": saved.json(),
|
|
}
|
|
assert fetched.json() == saved.json()
|
|
assert fetched.headers["cache-control"] == "no-store"
|