counterweight/arbret
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

first implementation

Browse source
This commit is contained in:
counterweight 2025-12-20 11:12:11 +01:00
parent 79458bcba4
commit 870804e7b9
Signed by: counterweight
GPG key ID: 883EDBAA726BD96C
24 changed files with 5485 additions and 184 deletions
Download patch file Download diff file Expand all files Collapse all files

44
backend/tests/helpers.py
View file

@ -1,7 +1,51 @@
import uuid
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import User, Invite, InviteStatus, ROLE_ADMIN
from invite_utils import generate_invite_identifier
def unique_email(prefix: str = "test") -> str:
"""Generate a unique email for tests sharing the same database."""
return f"{prefix}-{uuid.uuid4().hex[:8]}@example.com"
async def create_invite_for_registration(db: AsyncSession, godfather_email: str) -> str:
"""
Create an invite that can be used for registration.
Returns the invite identifier.
"""
# Find godfather
result = await db.execute(select(User).where(User.email == godfather_email))
godfather = result.scalar_one_or_none()
if not godfather:
# Create a godfather user (admin can create invites)
from auth import get_password_hash
from models import Role
result = await db.execute(select(Role).where(Role.name == ROLE_ADMIN))
admin_role = result.scalar_one_or_none()
godfather = User(
email=godfather_email,
hashed_password=get_password_hash("password123"),
roles=[admin_role] if admin_role else [],
)
db.add(godfather)
await db.flush()
# Create invite
identifier = generate_invite_identifier()
invite = Invite(
identifier=identifier,
godfather_id=godfather.id,
status=InviteStatus.READY,
)
db.add(invite)
await db.commit()
return identifier

217
backend/tests/test_auth.py
View file

@ -1,16 +1,31 @@
"""Tests for authentication endpoints.
Note: Registration now requires an invite code. Tests that need to register
users will create invites first via the helper function.
"""
import pytest
from auth import COOKIE_NAME
from tests.helpers import unique_email
from tests.helpers import unique_email, create_invite_for_registration
# Registration tests
# Registration tests (with invite)
@pytest.mark.asyncio
async def test_register_success(client):
async def test_register_success(client_factory):
"""Can register with valid invite code."""
email = unique_email("register")
response = await client.post(
# Create invite
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("godfather"))
response = await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
assert response.status_code == 200
data = response.json()
@ -25,62 +40,110 @@ async def test_register_success(client):
@pytest.mark.asyncio
async def test_register_duplicate_email(client):
async def test_register_duplicate_email(client_factory):
"""Cannot register with already-used email."""
email = unique_email("duplicate")
await client.post(
# Create two invites
async with client_factory.get_db_session() as db:
invite1 = await create_invite_for_registration(db, unique_email("gf1"))
invite2 = await create_invite_for_registration(db, unique_email("gf2"))
# First registration
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite1,
},
)
response = await client.post(
# Second registration with same email
response = await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "differentpass"},
json={
"email": email,
"password": "differentpass",
"invite_identifier": invite2,
},
)
assert response.status_code == 400
assert response.json()["detail"] == "Email already registered"
@pytest.mark.asyncio
async def test_register_invalid_email(client):
response = await client.post(
async def test_register_invalid_email(client_factory):
"""Cannot register with invalid email format."""
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
response = await client_factory.post(
"/api/auth/register",
json={"email": "notanemail", "password": "password123"},
json={
"email": "notanemail",
"password": "password123",
"invite_identifier": invite_code,
},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_register_missing_password(client):
"""Cannot register without password."""
response = await client.post(
"/api/auth/register",
json={"email": unique_email()},
json={"email": unique_email(), "invite_identifier": "some-code-00"},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_register_missing_email(client):
"""Cannot register without email."""
response = await client.post(
"/api/auth/register",
json={"password": "password123"},
json={"password": "password123", "invite_identifier": "some-code-00"},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_register_missing_invite(client):
"""Cannot register without invite code."""
response = await client.post(
"/api/auth/register",
json={"email": unique_email(), "password": "password123"},
)
assert response.status_code == 422
@pytest.mark.asyncio
async def test_register_empty_body(client):
"""Cannot register with empty body."""
response = await client.post("/api/auth/register", json={})
assert response.status_code == 422
# Login tests
@pytest.mark.asyncio
async def test_login_success(client):
async def test_login_success(client_factory):
"""Can login with valid credentials."""
email = unique_email("login")
await client.post(
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
response = await client.post(
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "password123"},
)
@ -93,13 +156,22 @@ async def test_login_success(client):
@pytest.mark.asyncio
async def test_login_wrong_password(client):
async def test_login_wrong_password(client_factory):
"""Cannot login with wrong password."""
email = unique_email("wrongpass")
await client.post(
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "correctpassword"},
json={
"email": email,
"password": "correctpassword",
"invite_identifier": invite_code,
},
)
response = await client.post(
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "wrongpassword"},
)
@ -109,6 +181,7 @@ async def test_login_wrong_password(client):
@pytest.mark.asyncio
async def test_login_nonexistent_user(client):
"""Cannot login with non-existent user."""
response = await client.post(
"/api/auth/login",
json={"email": unique_email("nonexistent"), "password": "password123"},
@ -119,6 +192,7 @@ async def test_login_nonexistent_user(client):
@pytest.mark.asyncio
async def test_login_invalid_email_format(client):
"""Cannot login with invalid email format."""
response = await client.post(
"/api/auth/login",
json={"email": "invalidemail", "password": "password123"},
@ -128,6 +202,7 @@ async def test_login_invalid_email_format(client):
@pytest.mark.asyncio
async def test_login_missing_fields(client):
"""Cannot login with missing fields."""
response = await client.post("/api/auth/login", json={})
assert response.status_code == 422
@ -135,16 +210,22 @@ async def test_login_missing_fields(client):
# Get current user tests
@pytest.mark.asyncio
async def test_get_me_success(client_factory):
"""Can get current user info when authenticated."""
email = unique_email("me")
# Register and get cookies
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg_response = await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg_response.cookies)
# Use authenticated client
async with client_factory.create(cookies=cookies) as authed:
response = await authed.get("/api/auth/me")
@ -158,12 +239,14 @@ async def test_get_me_success(client_factory):
@pytest.mark.asyncio
async def test_get_me_no_cookie(client):
"""Cannot get current user without auth cookie."""
response = await client.get("/api/auth/me")
assert response.status_code == 401
@pytest.mark.asyncio
async def test_get_me_invalid_cookie(client_factory):
"""Cannot get current user with invalid cookie."""
async with client_factory.create(cookies={COOKIE_NAME: "invalidtoken123"}) as authed:
response = await authed.get("/api/auth/me")
assert response.status_code == 401
@ -172,6 +255,7 @@ async def test_get_me_invalid_cookie(client_factory):
@pytest.mark.asyncio
async def test_get_me_expired_token(client_factory):
"""Cannot get current user with expired token."""
bad_token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOjEsImV4cCI6MH0.invalid"
async with client_factory.create(cookies={COOKIE_NAME: bad_token}) as authed:
response = await authed.get("/api/auth/me")
@ -181,11 +265,19 @@ async def test_get_me_expired_token(client_factory):
# Cookie validation tests
@pytest.mark.asyncio
async def test_cookie_from_register_works_for_me(client_factory):
"""Auth cookie from registration works for subsequent requests."""
email = unique_email("tokentest")
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg_response = await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg_response.cookies)
@ -198,11 +290,19 @@ async def test_cookie_from_register_works_for_me(client_factory):
@pytest.mark.asyncio
async def test_cookie_from_login_works_for_me(client_factory):
"""Auth cookie from login works for subsequent requests."""
email = unique_email("logintoken")
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
login_response = await client_factory.post(
"/api/auth/login",
@ -220,16 +320,29 @@ async def test_cookie_from_login_works_for_me(client_factory):
# Multiple users tests
@pytest.mark.asyncio
async def test_multiple_users_isolated(client_factory):
"""Multiple users have isolated sessions."""
email1 = unique_email("user1")
email2 = unique_email("user2")
async with client_factory.get_db_session() as db:
invite1 = await create_invite_for_registration(db, unique_email("gf1"))
invite2 = await create_invite_for_registration(db, unique_email("gf2"))
resp1 = await client_factory.post(
"/api/auth/register",
json={"email": email1, "password": "password1"},
json={
"email": email1,
"password": "password1",
"invite_identifier": invite1,
},
)
resp2 = await client_factory.post(
"/api/auth/register",
json={"email": email2, "password": "password2"},
json={
"email": email2,
"password": "password2",
"invite_identifier": invite2,
},
)
cookies1 = dict(resp1.cookies)
@ -248,13 +361,22 @@ async def test_multiple_users_isolated(client_factory):
# Password tests
@pytest.mark.asyncio
async def test_password_is_hashed(client):
async def test_password_is_hashed(client_factory):
"""Passwords are properly hashed (can login with correct password)."""
email = unique_email("hashtest")
await client.post(
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "mySecurePassword123"},
json={
"email": email,
"password": "mySecurePassword123",
"invite_identifier": invite_code,
},
)
response = await client.post(
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "mySecurePassword123"},
)
@ -262,13 +384,22 @@ async def test_password_is_hashed(client):
@pytest.mark.asyncio
async def test_case_sensitive_password(client):
async def test_case_sensitive_password(client_factory):
"""Passwords are case-sensitive."""
email = unique_email("casetest")
await client.post(
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "Password123"},
json={
"email": email,
"password": "Password123",
"invite_identifier": invite_code,
},
)
response = await client.post(
response = await client_factory.post(
"/api/auth/login",
json={"email": email, "password": "password123"},
)
@ -278,11 +409,19 @@ async def test_case_sensitive_password(client):
# Logout tests
@pytest.mark.asyncio
async def test_logout_success(client_factory):
"""Can logout successfully."""
email = unique_email("logout")
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg_response = await client_factory.post(
"/api/auth/register",
json={"email": email, "password": "password123"},
json={
"email": email,
"password": "password123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg_response.cookies)

59
backend/tests/test_counter.py
View file

@ -1,7 +1,11 @@
"""Tests for counter endpoints.
Note: Registration now requires an invite code.
"""
import pytest
from auth import COOKIE_NAME
from tests.helpers import unique_email
from tests.helpers import unique_email, create_invite_for_registration
# Protected endpoint tests - without auth
@ -34,9 +38,16 @@ async def test_increment_counter_invalid_cookie(client_factory):
# Authenticated counter tests
@pytest.mark.asyncio
async def test_get_counter_authenticated(client_factory):
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "testpass123"},
json={
"email": unique_email(),
"password": "testpass123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg.cookies)
@ -49,9 +60,16 @@ async def test_get_counter_authenticated(client_factory):
@pytest.mark.asyncio
async def test_increment_counter(client_factory):
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "testpass123"},
json={
"email": unique_email(),
"password": "testpass123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg.cookies)
@ -68,9 +86,16 @@ async def test_increment_counter(client_factory):
@pytest.mark.asyncio
async def test_increment_counter_multiple(client_factory):
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "testpass123"},
json={
"email": unique_email(),
"password": "testpass123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg.cookies)
@ -89,9 +114,16 @@ async def test_increment_counter_multiple(client_factory):
@pytest.mark.asyncio
async def test_get_counter_after_increment(client_factory):
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
reg = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "testpass123"},
json={
"email": unique_email(),
"password": "testpass123",
"invite_identifier": invite_code,
},
)
cookies = dict(reg.cookies)
@ -109,10 +141,19 @@ async def test_get_counter_after_increment(client_factory):
# Counter is shared between users
@pytest.mark.asyncio
async def test_counter_shared_between_users(client_factory):
# Create invites for two users
async with client_factory.get_db_session() as db:
invite1 = await create_invite_for_registration(db, unique_email("gf1"))
invite2 = await create_invite_for_registration(db, unique_email("gf2"))
# Create first user
reg1 = await client_factory.post(
"/api/auth/register",
json={"email": unique_email("share1"), "password": "testpass123"},
json={
"email": unique_email("share1"),
"password": "testpass123",
"invite_identifier": invite1,
},
)
cookies1 = dict(reg1.cookies)
@ -127,7 +168,11 @@ async def test_counter_shared_between_users(client_factory):
# Create second user - should see the increments
reg2 = await client_factory.post(
"/api/auth/register",
json={"email": unique_email("share2"), "password": "testpass123"},
json={
"email": unique_email("share2"),
"password": "testpass123",
"invite_identifier": invite2,
},
)
cookies2 = dict(reg2.cookies)

1013
backend/tests/test_invites.py Normal file
View file

File diff suppressed because it is too large Load diff

11
backend/tests/test_permissions.py
View file

@ -305,11 +305,18 @@ class TestSecurityBypassAttempts:
Test that new registrations cannot claim admin role.
New users should only get 'regular' role by default.
"""
from tests.helpers import unique_email
from tests.helpers import unique_email, create_invite_for_registration
async with client_factory.get_db_session() as db:
invite_code = await create_invite_for_registration(db, unique_email("gf"))
response = await client_factory.post(
"/api/auth/register",
json={"email": unique_email(), "password": "password123"},
json={
"email": unique_email(),
"password": "password123",
"invite_identifier": invite_code,
},
)
assert response.status_code == 200

54
backend/tests/test_profile.py
View file

@ -397,3 +397,57 @@ class TestProfilePrivacy:
assert "telegram" not in data
assert "signal" not in data
assert "nostr_npub" not in data
class TestProfileGodfather:
"""Tests for godfather information in profile."""
async def test_profile_shows_godfather_email(self, client_factory, admin_user, regular_user):
"""Profile shows godfather email for users who signed up with invite."""
from tests.helpers import unique_email
from sqlalchemy import select
from models import User
# Create invite
async with client_factory.create(cookies=admin_user["cookies"]) as client:
async with client_factory.get_db_session() as db:
result = await db.execute(
select(User).where(User.email == regular_user["email"])
)
godfather = result.scalar_one()
create_resp = await client.post(
"/api/admin/invites",
json={"godfather_id": godfather.id},
)
identifier = create_resp.json()["identifier"]
# Register new user with invite
new_email = unique_email("godchild")
async with client_factory.create() as client:
reg_resp = await client.post(
"/api/auth/register",
json={
"email": new_email,
"password": "password123",
"invite_identifier": identifier,
},
)
new_user_cookies = dict(reg_resp.cookies)
# Check profile shows godfather
async with client_factory.create(cookies=new_user_cookies) as client:
response = await client.get("/api/profile")
assert response.status_code == 200
data = response.json()
assert data["godfather_email"] == regular_user["email"]
async def test_profile_godfather_null_for_seeded_users(self, client_factory, regular_user):
"""Profile shows null godfather for users without one (e.g., seeded users)."""
async with client_factory.create(cookies=regular_user["cookies"]) as client:
response = await client.get("/api/profile")
assert response.status_code == 200
data = response.json()
assert data["godfather_email"] is None