576 lines
23 KiB
Python
576 lines
23 KiB
Python
import asyncio
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from src import main
|
|
from src.session_store import SessionStoreError
|
|
|
|
|
|
@pytest.fixture
|
|
def access_control(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-attempts.sqlite3"))
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_MAX_FAILURES", "3")
|
|
monkeypatch.setenv("STACKCHAIN_LOGIN_WINDOW_SECONDS", "60")
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_default_operator_mode_fails_closed_before_gitea_when_secrets_are_missing(monkeypatch):
|
|
monkeypatch.delenv("STACKCHAIN_DASHBOARD_AUTH_MODE", raising=False)
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
health = await client.get("/healthz")
|
|
responses = [
|
|
await client.get("/readyz"),
|
|
await client.get("/login"),
|
|
await client.get("/"),
|
|
await client.get("/api/v1/context"),
|
|
]
|
|
|
|
assert health.status_code == 200
|
|
assert [response.status_code for response in responses] == [503, 503, 503, 503]
|
|
assert all(response.headers["cache-control"] == "no-store" for response in responses)
|
|
assert all(response.json() == {
|
|
"detail": "Dashboard authentication is not configured"
|
|
} for response in responses)
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_operator_mode_rejects_reused_or_incomplete_secrets(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "same-secret-with-at-least-thirty-two-characters")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "same-secret-with-at-least-thirty-two-characters")
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
reused = await client.get("/login")
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_SESSION_SECRET", "")
|
|
incomplete = await client.get("/login")
|
|
|
|
assert reused.status_code == 503
|
|
assert incomplete.status_code == 503
|
|
assert "same-secret" not in reused.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_insecure_local_mode_is_restricted_to_loopback(monkeypatch):
|
|
monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "insecure-local")
|
|
local_transport = httpx.ASGITransport(app=main.app, client=("127.0.0.1", 1234))
|
|
remote_transport = httpx.ASGITransport(app=main.app, client=("203.0.113.9", 1234))
|
|
async with httpx.AsyncClient(transport=local_transport, base_url="http://test") as client:
|
|
local = await client.get("/login")
|
|
async with httpx.AsyncClient(transport=remote_transport, base_url="http://test") as client:
|
|
remote = await client.get("/login")
|
|
|
|
assert local.status_code == 200
|
|
assert remote.status_code == 403
|
|
assert remote.json() == {"detail": "Insecure local mode requires a loopback client"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_anonymous_private_request_is_rejected_before_gitea(access_control, monkeypatch):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.get("/api/v1/context")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json() == {"detail": "Authentication required"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert called is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_creates_secure_session_without_echoing_access_token(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authenticated": True}
|
|
cookies = response.headers.get_list("set-cookie")
|
|
assert any("stackchain_session=" in value and "HttpOnly" in value and "Secure" in value and "SameSite=strict" in value for value in cookies)
|
|
assert any("stackchain_csrf=" in value and "Secure" in value and "SameSite=strict" in value and "HttpOnly" not in value for value in cookies)
|
|
assert "correct horse battery staple" not in response.text
|
|
assert response.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_throttles_repeated_failures_with_retry_guidance(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.7", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
failures = [
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
for _ in range(3)
|
|
]
|
|
blocked = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert [response.status_code for response in failures] == [401, 401, 401]
|
|
assert blocked.status_code == 429
|
|
assert blocked.json() == {"detail": "Too many sign-in attempts"}
|
|
assert blocked.headers["retry-after"].isdigit()
|
|
assert blocked.headers["cache-control"] == "no-store"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_successful_sign_in_clears_prior_failures(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, client=("203.0.113.8", 1234))
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
for _ in range(2):
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
success = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
after_success = [
|
|
await client.post("/api/v1/session", json={"access_token": "wrong"})
|
|
for _ in range(3)
|
|
]
|
|
|
|
assert success.status_code == 200
|
|
assert [response.status_code for response in after_success] == [401, 401, 401]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_get_reaches_private_api(access_control, monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy", "full_name": "", "email": ""}
|
|
|
|
async def empty():
|
|
return []
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
monkeypatch.setattr(main, "repos", empty)
|
|
monkeypatch.setattr(main, "issues", empty)
|
|
monkeypatch.setattr(main, "pull_requests", empty)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
signed_in = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
response = await client.get("/api/v1/context")
|
|
|
|
assert signed_in.status_code == 200
|
|
assert response.status_code == 200
|
|
assert response.json()["user"]["login"] == "timmy"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_status_reuses_the_middleware_validation(access_control, monkeypatch):
|
|
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"}
|
|
)
|
|
original_store = main.dashboard_auth._session_store()
|
|
lookups = 0
|
|
|
|
class CountingStore:
|
|
def is_active(self, session_id, expires_at):
|
|
nonlocal lookups
|
|
lookups += 1
|
|
return original_store.is_active(session_id, expires_at)
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: CountingStore())
|
|
response = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["authenticated"] is True
|
|
assert lookups == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_authenticated_session_status_exposes_only_csrf_proof(access_control):
|
|
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"}
|
|
)
|
|
response = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"authenticated": True,
|
|
"csrf_token": client.cookies["stackchain_csrf"],
|
|
}
|
|
assert "correct horse battery staple" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_mutation_requires_same_origin_and_session_csrf(access_control, monkeypatch):
|
|
calls = 0
|
|
|
|
async def mark_read(notification_id):
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": notification_id, "read": True}
|
|
|
|
monkeypatch.setattr(main, "mark_notification_read", mark_read)
|
|
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"}
|
|
)
|
|
missing = await client.patch("/api/v1/notifications/7/read")
|
|
foreign = await client.patch(
|
|
"/api/v1/notifications/7/read",
|
|
headers={
|
|
"Origin": "https://evil.example",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
valid = await client.patch(
|
|
"/api/v1/notifications/7/read",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
|
|
assert missing.status_code == 403
|
|
assert foreign.status_code == 403
|
|
assert valid.status_code == 200
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_clears_session_and_blocks_private_routes(access_control):
|
|
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"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
private = await client.get("/api/v1/background-identity")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {"authenticated": False, "clear_private_device_data": True}
|
|
assert private.status_code == 401
|
|
assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_revokes_every_existing_session(access_control):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with (
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as phone,
|
|
httpx.AsyncClient(transport=transport, base_url="https://test") as laptop,
|
|
):
|
|
await phone.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
await laptop.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
response = await phone.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://test",
|
|
"X-CSRF-Token": phone.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
phone_private = await phone.get("/api/v1/background-identity")
|
|
laptop_private = await laptop.get("/api/v1/background-identity")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"authenticated": False,
|
|
"all_sessions_revoked": True,
|
|
"clear_private_device_data": True,
|
|
}
|
|
assert phone_private.status_code == 401
|
|
assert laptop_private.status_code == 401
|
|
assert all("Max-Age=0" in value for value in response.headers.get_list("set-cookie"))
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_rejects_cross_site_requests_without_revoking(access_control):
|
|
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"}
|
|
)
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={
|
|
"Origin": "https://evil.example",
|
|
"X-CSRF-Token": client.cookies["stackchain_csrf"],
|
|
},
|
|
)
|
|
session = await client.get("/api/v1/session")
|
|
|
|
assert response.status_code == 403
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert session.status_code == 200
|
|
assert session.json()["authenticated"] is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_out_all_devices_registry_failure_sets_no_cookies(access_control, monkeypatch):
|
|
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"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
class BrokenStore:
|
|
def is_active(self, session_id, expires_at):
|
|
return True
|
|
|
|
def revoke_all(self):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.delete(
|
|
"/api/v1/sessions",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_revokes_a_captured_cookie_before_gitea(access_control, monkeypatch):
|
|
calls = 0
|
|
|
|
async def user():
|
|
nonlocal calls
|
|
calls += 1
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
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"}
|
|
)
|
|
captured = client.cookies["stackchain_session"]
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://test",
|
|
headers={"Cookie": f"stackchain_session={captured}"},
|
|
) as replay:
|
|
before = await replay.get("/api/v1/background-identity")
|
|
|
|
await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=transport,
|
|
base_url="https://test",
|
|
headers={"Cookie": f"stackchain_session={captured}"},
|
|
) as replay:
|
|
after = await replay.get("/api/v1/background-identity")
|
|
|
|
assert before.status_code == 200
|
|
assert after.status_code == 401
|
|
assert after.headers["cache-control"] == "no-store"
|
|
assert calls == 1
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_registry_latency_does_not_block_the_event_loop(access_control, monkeypatch):
|
|
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"}
|
|
)
|
|
|
|
class SlowStore:
|
|
def is_active(self, session_id, expires_at):
|
|
time.sleep(0.15)
|
|
return True
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: SlowStore())
|
|
private_request = asyncio.create_task(client.get("/api/v1/session"))
|
|
await asyncio.sleep(0)
|
|
started = time.perf_counter()
|
|
await asyncio.sleep(0.01)
|
|
heartbeat_elapsed = time.perf_counter() - started
|
|
response = await private_request
|
|
|
|
assert response.status_code == 200
|
|
assert heartbeat_elapsed < 0.08
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_session_registry_read_failure_fails_closed_before_gitea(
|
|
access_control, monkeypatch
|
|
):
|
|
called = False
|
|
|
|
async def user():
|
|
nonlocal called
|
|
called = True
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
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"}
|
|
)
|
|
|
|
class BrokenStore:
|
|
def is_active(self, session_id, expires_at):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.get("/api/v1/background-identity")
|
|
health = await client.get("/healthz")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert "database path" not in response.text
|
|
assert called is False
|
|
assert health.status_code == 200
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_sign_in_registry_write_failure_issues_no_cookie(access_control, monkeypatch):
|
|
class BrokenStore:
|
|
def activate(self, session_id, expires_at):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
response = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logout_registry_failure_does_not_claim_revocation(access_control, monkeypatch):
|
|
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"}
|
|
)
|
|
csrf = client.cookies["stackchain_csrf"]
|
|
|
|
class BrokenStore:
|
|
def is_active(self, session_id, expires_at):
|
|
return True
|
|
|
|
def revoke(self, session_id):
|
|
raise SessionStoreError("database path and secret details")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
response = await client.delete(
|
|
"/api/v1/session",
|
|
headers={"Origin": "https://test", "X-CSRF-Token": csrf},
|
|
)
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"detail": "Session registry is temporarily unavailable"}
|
|
assert response.headers["cache-control"] == "no-store"
|
|
assert response.headers.get_list("set-cookie") == []
|
|
assert "database path" not in response.text
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_public_routes_skip_session_registry_validation(access_control, monkeypatch):
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
signed_in = await client.post(
|
|
"/api/v1/session", json={"access_token": "correct horse battery staple"}
|
|
)
|
|
|
|
class BrokenStore:
|
|
def is_active(self, session_id, expires_at):
|
|
raise SessionStoreError("public routes must not read the registry")
|
|
|
|
monkeypatch.setattr(main.dashboard_auth, "_session_store", lambda now=None: BrokenStore())
|
|
login = await client.get("/login")
|
|
manifest = await client.get("/manifest.webmanifest")
|
|
static = await client.get("/static/session.js")
|
|
|
|
assert signed_in.status_code == 200
|
|
assert [login.status_code, manifest.status_code, static.status_code] == [200, 200, 200]
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_public_routes_remain_available_and_readiness_hides_identity(access_control, monkeypatch):
|
|
async def user():
|
|
return {"id": 1, "login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", user)
|
|
transport = httpx.ASGITransport(app=main.app)
|
|
async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
|
|
health = await client.get("/healthz")
|
|
login = await client.get("/login")
|
|
ready = await client.get("/readyz")
|
|
|
|
assert health.status_code == 200
|
|
assert login.status_code == 200
|
|
assert 'name="access_token"' in login.text
|
|
assert ready.status_code == 200
|
|
assert ready.json() == {"status": "ready", "service": "stackchain-dashboard"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_subpath_deployment_scopes_routes_and_session_cookies(access_control):
|
|
transport = httpx.ASGITransport(app=main.app, root_path="/dashboard")
|
|
async with httpx.AsyncClient(
|
|
transport=transport, base_url="https://test/dashboard/"
|
|
) as client:
|
|
login = await client.get("login")
|
|
signed_in = await client.post(
|
|
"api/v1/session",
|
|
json={"access_token": "correct horse battery staple"},
|
|
)
|
|
status = await client.get("api/v1/session")
|
|
|
|
assert login.status_code == 200
|
|
assert signed_in.status_code == 200
|
|
assert status.status_code == 200
|
|
assert any(
|
|
"stackchain_session=" in value and "Path=/dashboard" in value
|
|
for value in signed_in.headers.get_list("set-cookie")
|
|
)
|