399 lines
15 KiB
Python
399 lines
15 KiB
Python
"""Tests for user profile and contact details."""
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from models import User, ROLE_REGULAR
|
|
from auth import get_password_hash
|
|
from tests.helpers import unique_email
|
|
|
|
# Valid npub for testing (32 zero bytes encoded as bech32)
|
|
VALID_NPUB = "npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqzqujme"
|
|
|
|
|
|
class TestUserContactFields:
|
|
"""Test that contact fields work correctly on the User model."""
|
|
|
|
async def test_contact_fields_default_to_none(self, client_factory):
|
|
"""New users should have all contact fields as None."""
|
|
email = unique_email("test")
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=get_password_hash("password123"),
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
|
|
assert user.contact_email is None
|
|
assert user.telegram is None
|
|
assert user.signal is None
|
|
assert user.nostr_npub is None
|
|
|
|
async def test_contact_fields_can_be_set(self, client_factory):
|
|
"""Contact fields can be set when creating a user."""
|
|
email = unique_email("test")
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=get_password_hash("password123"),
|
|
contact_email="contact@example.com",
|
|
telegram="@alice",
|
|
signal="alice.42",
|
|
nostr_npub="npub1test",
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(user)
|
|
|
|
assert user.contact_email == "contact@example.com"
|
|
assert user.telegram == "@alice"
|
|
assert user.signal == "alice.42"
|
|
assert user.nostr_npub == "npub1test"
|
|
|
|
async def test_contact_fields_persist_after_reload(self, client_factory):
|
|
"""Contact fields should persist in the database."""
|
|
email = unique_email("test")
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=get_password_hash("password123"),
|
|
contact_email="contact@example.com",
|
|
telegram="@bob",
|
|
signal="bob.99",
|
|
nostr_npub="npub1xyz",
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
user_id = user.id
|
|
|
|
# Reload from database in a new session
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
loaded_user = result.scalar_one()
|
|
|
|
assert loaded_user.contact_email == "contact@example.com"
|
|
assert loaded_user.telegram == "@bob"
|
|
assert loaded_user.signal == "bob.99"
|
|
assert loaded_user.nostr_npub == "npub1xyz"
|
|
|
|
async def test_contact_fields_can_be_updated(self, client_factory):
|
|
"""Contact fields can be updated after user creation."""
|
|
email = unique_email("test")
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=get_password_hash("password123"),
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
user_id = user.id
|
|
|
|
# Update fields
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
|
|
user.contact_email = "new@example.com"
|
|
user.telegram = "@updated"
|
|
await db.commit()
|
|
|
|
# Verify update persisted
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
|
|
assert user.contact_email == "new@example.com"
|
|
assert user.telegram == "@updated"
|
|
assert user.signal is None # Still None
|
|
assert user.nostr_npub is None # Still None
|
|
|
|
async def test_contact_fields_can_be_cleared(self, client_factory):
|
|
"""Contact fields can be set back to None."""
|
|
email = unique_email("test")
|
|
|
|
async with client_factory.get_db_session() as db:
|
|
user = User(
|
|
email=email,
|
|
hashed_password=get_password_hash("password123"),
|
|
contact_email="contact@example.com",
|
|
telegram="@alice",
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
user_id = user.id
|
|
|
|
# Clear fields
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
|
|
user.contact_email = None
|
|
user.telegram = None
|
|
await db.commit()
|
|
|
|
# Verify cleared
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one()
|
|
|
|
assert user.contact_email is None
|
|
assert user.telegram is None
|
|
|
|
|
|
class TestGetProfileEndpoint:
|
|
"""Tests for GET /api/profile endpoint."""
|
|
|
|
async def test_regular_user_can_get_profile(self, client_factory, regular_user):
|
|
"""Regular user can fetch their profile."""
|
|
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 "contact_email" in data
|
|
assert "telegram" in data
|
|
assert "signal" in data
|
|
assert "nostr_npub" in data
|
|
# All should be None for new user
|
|
assert data["contact_email"] is None
|
|
assert data["telegram"] is None
|
|
assert data["signal"] is None
|
|
assert data["nostr_npub"] is None
|
|
|
|
async def test_admin_user_cannot_access_profile(self, client_factory, admin_user):
|
|
"""Admin user gets 403 when trying to access profile."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.get("/api/profile")
|
|
|
|
assert response.status_code == 403
|
|
assert "regular users" in response.json()["detail"].lower()
|
|
|
|
async def test_unauthenticated_user_gets_401(self, client_factory):
|
|
"""Unauthenticated user gets 401."""
|
|
async with client_factory.create() as client:
|
|
response = await client.get("/api/profile")
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_profile_returns_existing_data(self, client_factory, regular_user):
|
|
"""Profile returns data that was previously set."""
|
|
# Set some data directly in DB
|
|
async with client_factory.get_db_session() as db:
|
|
result = await db.execute(
|
|
select(User).where(User.email == regular_user["email"])
|
|
)
|
|
user = result.scalar_one()
|
|
user.contact_email = "contact@test.com"
|
|
user.telegram = "@testuser"
|
|
await db.commit()
|
|
|
|
# Fetch via API
|
|
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["contact_email"] == "contact@test.com"
|
|
assert data["telegram"] == "@testuser"
|
|
assert data["signal"] is None
|
|
assert data["nostr_npub"] is None
|
|
|
|
|
|
class TestUpdateProfileEndpoint:
|
|
"""Tests for PUT /api/profile endpoint."""
|
|
|
|
async def test_regular_user_can_update_profile(self, client_factory, regular_user):
|
|
"""Regular user can update their profile."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "new@example.com",
|
|
"telegram": "@newhandle",
|
|
"signal": "signal.42",
|
|
"nostr_npub": VALID_NPUB,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["contact_email"] == "new@example.com"
|
|
assert data["telegram"] == "@newhandle"
|
|
assert data["signal"] == "signal.42"
|
|
assert data["nostr_npub"] == VALID_NPUB
|
|
|
|
async def test_profile_update_persists(self, client_factory, regular_user):
|
|
"""Profile updates persist in the database."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
await client.put(
|
|
"/api/profile",
|
|
json={"telegram": "@persisted"},
|
|
)
|
|
|
|
# Fetch again to verify
|
|
response = await client.get("/api/profile")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["telegram"] == "@persisted"
|
|
|
|
async def test_admin_user_cannot_update_profile(self, client_factory, admin_user):
|
|
"""Admin user gets 403 when trying to update profile."""
|
|
async with client_factory.create(cookies=admin_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={"telegram": "@admin"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
async def test_unauthenticated_user_gets_401(self, client_factory):
|
|
"""Unauthenticated user gets 401."""
|
|
async with client_factory.create() as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={"telegram": "@test"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_can_clear_fields(self, client_factory, regular_user):
|
|
"""User can clear fields by setting them to null."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
# First set some values
|
|
await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "test@example.com",
|
|
"telegram": "@test",
|
|
},
|
|
)
|
|
|
|
# Then clear them
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": None,
|
|
"telegram": None,
|
|
"signal": None,
|
|
"nostr_npub": None,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["contact_email"] is None
|
|
assert data["telegram"] is None
|
|
|
|
async def test_invalid_email_returns_422(self, client_factory, regular_user):
|
|
"""Invalid email format returns 422 with field error."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={"contact_email": "not-an-email"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert "field_errors" in data["detail"]
|
|
assert "contact_email" in data["detail"]["field_errors"]
|
|
|
|
async def test_invalid_telegram_returns_422(self, client_factory, regular_user):
|
|
"""Invalid telegram handle returns 422 with field error."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={"telegram": "missing_at_sign"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert "field_errors" in data["detail"]
|
|
assert "telegram" in data["detail"]["field_errors"]
|
|
|
|
async def test_invalid_npub_returns_422(self, client_factory, regular_user):
|
|
"""Invalid nostr npub returns 422 with field error."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={"nostr_npub": "npub1invalid"},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert "field_errors" in data["detail"]
|
|
assert "nostr_npub" in data["detail"]["field_errors"]
|
|
|
|
async def test_multiple_invalid_fields_returns_all_errors(self, client_factory, regular_user):
|
|
"""Multiple invalid fields return all errors."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "bad-email",
|
|
"telegram": "no-at",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
data = response.json()
|
|
assert "contact_email" in data["detail"]["field_errors"]
|
|
assert "telegram" in data["detail"]["field_errors"]
|
|
|
|
async def test_partial_update_preserves_other_fields(self, client_factory, regular_user):
|
|
"""Updating one field doesn't affect others (they get set to the request values)."""
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
# Set initial values
|
|
await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "initial@example.com",
|
|
"telegram": "@initial",
|
|
},
|
|
)
|
|
|
|
# Update only telegram, but note: PUT replaces all fields
|
|
# So we need to include all fields we want to keep
|
|
response = await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "initial@example.com",
|
|
"telegram": "@updated",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["contact_email"] == "initial@example.com"
|
|
assert data["telegram"] == "@updated"
|
|
|
|
|
|
class TestProfilePrivacy:
|
|
"""Tests to ensure profile data is private."""
|
|
|
|
async def test_profile_not_in_auth_me(self, client_factory, regular_user):
|
|
"""Contact details should NOT appear in /api/auth/me response."""
|
|
# First set some profile data
|
|
async with client_factory.create(cookies=regular_user["cookies"]) as client:
|
|
await client.put(
|
|
"/api/profile",
|
|
json={
|
|
"contact_email": "secret@example.com",
|
|
"telegram": "@secret",
|
|
},
|
|
)
|
|
|
|
# Check /api/auth/me doesn't expose it
|
|
response = await client.get("/api/auth/me")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
# These fields should NOT be in the response
|
|
assert "contact_email" not in data
|
|
assert "telegram" not in data
|
|
assert "signal" not in data
|
|
assert "nostr_npub" not in data
|