1c21d6d199
Part of CLI module work (fastapi-oidc-op-9lb.1). The test verifies that
/register/{token} handles users already created by initial-admin.
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
|
|
from httpx import AsyncClient
|
|
|
|
from porchlight.models import MagicLink, User
|
|
|
|
|
|
async def test_register_invalid_token_returns_error_page(client: AsyncClient) -> None:
|
|
res = await client.get("/register/nope", follow_redirects=False)
|
|
assert res.status_code == 400
|
|
assert "Invalid or expired" in res.text
|
|
|
|
|
|
async def test_register_expired_token_returns_error_page(client: AsyncClient) -> None:
|
|
app = client._transport.app # type: ignore[union-attr]
|
|
repo = app.state.magic_link_repo
|
|
await repo.create(
|
|
MagicLink(
|
|
token="expired",
|
|
username="newuser",
|
|
expires_at=datetime.now(UTC) - timedelta(hours=1),
|
|
)
|
|
)
|
|
|
|
res = await client.get("/register/expired", follow_redirects=False)
|
|
assert res.status_code == 400
|
|
assert "Invalid or expired" in res.text
|
|
|
|
|
|
async def test_register_valid_token_creates_user_and_redirects(client: AsyncClient) -> None:
|
|
app = client._transport.app # type: ignore[union-attr]
|
|
magic_link_repo = app.state.magic_link_repo
|
|
user_repo = app.state.user_repo
|
|
|
|
await magic_link_repo.create(
|
|
MagicLink(
|
|
token="t1",
|
|
username="newuser",
|
|
expires_at=datetime.now(UTC) + timedelta(hours=1),
|
|
)
|
|
)
|
|
|
|
res = await client.get("/register/t1", follow_redirects=False)
|
|
assert res.status_code in (302, 303)
|
|
assert "/manage/credentials" in res.headers["location"]
|
|
assert "setup=1" in res.headers["location"]
|
|
|
|
# Token should be marked used
|
|
link = await magic_link_repo.get_by_token("t1")
|
|
assert link is not None
|
|
assert link.used is True
|
|
|
|
# User should exist
|
|
user = await user_repo.get_by_username("newuser")
|
|
assert user is not None
|
|
assert "users" in user.groups
|
|
|
|
|
|
async def test_register_used_token_returns_error(client: AsyncClient) -> None:
|
|
app = client._transport.app # type: ignore[union-attr]
|
|
repo = app.state.magic_link_repo
|
|
await repo.create(
|
|
MagicLink(
|
|
token="used",
|
|
username="newuser",
|
|
expires_at=datetime.now(UTC) + timedelta(hours=1),
|
|
used=True,
|
|
)
|
|
)
|
|
|
|
res = await client.get("/register/used", follow_redirects=False)
|
|
assert res.status_code == 400
|
|
|
|
|
|
async def test_register_existing_user_logs_in_and_redirects(client: AsyncClient) -> None:
|
|
"""When initial-admin creates a user, the invite link should log them in."""
|
|
app = client._transport.app # type: ignore[union-attr]
|
|
magic_link_repo = app.state.magic_link_repo
|
|
user_repo = app.state.user_repo
|
|
|
|
# Pre-create the user (as initial-admin would)
|
|
user = User(userid="lusab-bansen", username="admin", groups=["admin", "users"])
|
|
await user_repo.create(user)
|
|
|
|
# Create invite for the same username
|
|
await magic_link_repo.create(
|
|
MagicLink(
|
|
token="admin-setup",
|
|
username="admin",
|
|
expires_at=datetime.now(UTC) + timedelta(hours=1),
|
|
)
|
|
)
|
|
|
|
res = await client.get("/register/admin-setup", follow_redirects=False)
|
|
assert res.status_code in (302, 303)
|
|
assert "/manage/credentials" in res.headers["location"]
|
|
assert "setup=1" in res.headers["location"]
|
|
|
|
# Token should be marked used
|
|
link = await magic_link_repo.get_by_token("admin-setup")
|
|
assert link is not None
|
|
assert link.used is True
|
|
|
|
# Original user should still exist with original groups
|
|
existing = await user_repo.get_by_username("admin")
|
|
assert existing is not None
|
|
assert existing.userid == "lusab-bansen"
|
|
assert "admin" in existing.groups
|