107 lines
4.2 KiB
Python
107 lines
4.2 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_recent_work_api_is_authenticated_csrf_protected_no_store_and_account_scoped(
|
|
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_RECENT_WORK_DB", str(tmp_path / "recent-work.sqlite3"))
|
|
active_login = "Timmy"
|
|
|
|
async def user():
|
|
return {"id": 1, "login": active_login}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
entry = {
|
|
"kind": "issue",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 1475,
|
|
"title": "Sync recent work",
|
|
"route": "#/my-work/issue/stackchain/dashboard/1475",
|
|
}
|
|
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.post("/api/v1/recent-work", json=entry)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
saved = await client.post("/api/v1/recent-work", json=entry, headers=headers)
|
|
fetched = await client.get("/api/v1/recent-work")
|
|
active_login = "Alexander"
|
|
other_account = await client.get("/api/v1/recent-work")
|
|
|
|
assert forbidden.status_code == 403
|
|
assert saved.status_code == 200
|
|
assert saved.json() == {"items": [entry], "pinned": []}
|
|
assert fetched.json() == saved.json()
|
|
assert fetched.headers["cache-control"] == "no-store"
|
|
assert other_account.json() == {"items": [], "pinned": []}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_recent_work_pin_api_is_csrf_protected_and_account_scoped(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_RECENT_WORK_DB", str(tmp_path / "recent-work.sqlite3"))
|
|
active_login = "Timmy"
|
|
|
|
async def user():
|
|
return {"id": 1, "login": active_login}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
entry = {
|
|
"kind": "pull",
|
|
"repository": "stackchain/dashboard",
|
|
"number": 1476,
|
|
"title": "Sync recent work",
|
|
"route": "#/my-work/pull/stackchain/dashboard/1476",
|
|
}
|
|
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/recent-work/pin", json=entry)
|
|
headers = {
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
}
|
|
pinned = await client.put("/api/v1/recent-work/pin", json=entry, headers=headers)
|
|
active_login = "Alexander"
|
|
isolated = await client.get("/api/v1/recent-work")
|
|
active_login = "Timmy"
|
|
unpinned = await client.request(
|
|
"DELETE",
|
|
"/api/v1/recent-work/pin",
|
|
json={"route": entry["route"]},
|
|
headers=headers,
|
|
)
|
|
|
|
assert forbidden.status_code == 403
|
|
assert pinned.status_code == 200
|
|
assert pinned.json() == {"items": [], "pinned": [entry]}
|
|
assert isolated.json() == {"items": [], "pinned": []}
|
|
assert unpinned.status_code == 200
|
|
assert unpinned.json() == {"items": [], "pinned": []}
|