implemented
This commit is contained in:
parent
40ca82bb45
commit
409e0df9a6
16 changed files with 2451 additions and 4 deletions
|
|
@ -10,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
||||
from database import engine, get_db, Base
|
||||
from models import Counter, User, SumRecord, CounterRecord, Permission, Role, ROLE_REGULAR
|
||||
from validation import validate_profile_fields
|
||||
|
||||
|
||||
R = TypeVar("R", bound=BaseModel)
|
||||
|
|
@ -323,3 +324,85 @@ async def get_sum_records(
|
|||
per_page=per_page,
|
||||
total_pages=total_pages,
|
||||
)
|
||||
|
||||
|
||||
# Profile endpoints
|
||||
class ProfileResponse(BaseModel):
|
||||
"""Response model for profile data."""
|
||||
contact_email: str | None
|
||||
telegram: str | None
|
||||
signal: str | None
|
||||
nostr_npub: str | None
|
||||
|
||||
|
||||
class ProfileUpdate(BaseModel):
|
||||
"""Request model for updating profile."""
|
||||
contact_email: str | None = None
|
||||
telegram: str | None = None
|
||||
signal: str | None = None
|
||||
nostr_npub: str | None = None
|
||||
|
||||
|
||||
def require_regular_user():
|
||||
"""Dependency that requires the user to have the 'regular' role."""
|
||||
async def checker(
|
||||
current_user: User = Depends(get_current_user),
|
||||
) -> User:
|
||||
if ROLE_REGULAR not in current_user.role_names:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="Profile access is only available to regular users",
|
||||
)
|
||||
return current_user
|
||||
return checker
|
||||
|
||||
|
||||
@app.get("/api/profile", response_model=ProfileResponse)
|
||||
async def get_profile(
|
||||
current_user: User = Depends(require_regular_user()),
|
||||
):
|
||||
"""Get the current user's profile (contact details)."""
|
||||
return ProfileResponse(
|
||||
contact_email=current_user.contact_email,
|
||||
telegram=current_user.telegram,
|
||||
signal=current_user.signal,
|
||||
nostr_npub=current_user.nostr_npub,
|
||||
)
|
||||
|
||||
|
||||
@app.put("/api/profile", response_model=ProfileResponse)
|
||||
async def update_profile(
|
||||
data: ProfileUpdate,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
current_user: User = Depends(require_regular_user()),
|
||||
):
|
||||
"""Update the current user's profile (contact details)."""
|
||||
# Validate all fields
|
||||
errors = validate_profile_fields(
|
||||
contact_email=data.contact_email,
|
||||
telegram=data.telegram,
|
||||
signal=data.signal,
|
||||
nostr_npub=data.nostr_npub,
|
||||
)
|
||||
|
||||
if errors:
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail={"field_errors": errors},
|
||||
)
|
||||
|
||||
# Update fields
|
||||
current_user.contact_email = data.contact_email
|
||||
current_user.telegram = data.telegram
|
||||
current_user.signal = data.signal
|
||||
current_user.nostr_npub = data.nostr_npub
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(current_user)
|
||||
|
||||
return ProfileResponse(
|
||||
contact_email=current_user.contact_email,
|
||||
telegram=current_user.telegram,
|
||||
signal=current_user.signal,
|
||||
nostr_npub=current_user.nostr_npub,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -101,6 +101,12 @@ class User(Base):
|
|||
email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
|
||||
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
|
||||
|
||||
# Contact details (all optional)
|
||||
contact_email: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
telegram: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
signal: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
nostr_npub: Mapped[str | None] = mapped_column(String(64), nullable=True)
|
||||
|
||||
# Relationship to roles
|
||||
roles: Mapped[list[Role]] = relationship(
|
||||
"Role",
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ dependencies = [
|
|||
"bcrypt>=4.0.0",
|
||||
"python-jose[cryptography]>=3.3.0",
|
||||
"email-validator>=2.0.0",
|
||||
"bech32>=1.2.0",
|
||||
]
|
||||
|
||||
[dependency-groups]
|
||||
|
|
|
|||
|
|
@ -67,6 +67,8 @@ async def upsert_user(db: AsyncSession, email: str, password: str, role_names: l
|
|||
|
||||
async def seed() -> None:
|
||||
async with engine.begin() as conn:
|
||||
# Drop all tables and recreate to ensure schema is up to date
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
|
||||
async with async_session() as db:
|
||||
|
|
|
|||
400
backend/tests/test_profile.py
Normal file
400
backend/tests/test_profile.py
Normal file
|
|
@ -0,0 +1,400 @@
|
|||
"""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
|
||||
|
||||
215
backend/tests/test_validation.py
Normal file
215
backend/tests/test_validation.py
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
"""Tests for profile field validation."""
|
||||
import pytest
|
||||
|
||||
from validation import (
|
||||
validate_contact_email,
|
||||
validate_telegram,
|
||||
validate_signal,
|
||||
validate_nostr_npub,
|
||||
validate_profile_fields,
|
||||
)
|
||||
|
||||
|
||||
class TestValidateContactEmail:
|
||||
"""Tests for contact email validation."""
|
||||
|
||||
def test_none_is_valid(self):
|
||||
assert validate_contact_email(None) is None
|
||||
|
||||
def test_empty_string_is_valid(self):
|
||||
assert validate_contact_email("") is None
|
||||
|
||||
def test_valid_email(self):
|
||||
assert validate_contact_email("user@example.com") is None
|
||||
|
||||
def test_valid_email_with_plus(self):
|
||||
assert validate_contact_email("user+tag@example.com") is None
|
||||
|
||||
def test_valid_email_subdomain(self):
|
||||
assert validate_contact_email("user@mail.example.com") is None
|
||||
|
||||
def test_invalid_email_no_at(self):
|
||||
result = validate_contact_email("userexample.com")
|
||||
assert result is not None
|
||||
assert "not valid" in result.lower() or "@" in result
|
||||
|
||||
def test_invalid_email_no_domain(self):
|
||||
result = validate_contact_email("user@")
|
||||
assert result is not None
|
||||
|
||||
def test_invalid_email_spaces(self):
|
||||
result = validate_contact_email("user @example.com")
|
||||
assert result is not None
|
||||
|
||||
|
||||
class TestValidateTelegram:
|
||||
"""Tests for Telegram handle validation."""
|
||||
|
||||
def test_none_is_valid(self):
|
||||
assert validate_telegram(None) is None
|
||||
|
||||
def test_empty_string_is_valid(self):
|
||||
assert validate_telegram("") is None
|
||||
|
||||
def test_valid_handle(self):
|
||||
assert validate_telegram("@alice") is None
|
||||
|
||||
def test_valid_handle_with_numbers(self):
|
||||
assert validate_telegram("@alice123") is None
|
||||
|
||||
def test_valid_handle_with_underscore(self):
|
||||
assert validate_telegram("@alice_bob") is None
|
||||
|
||||
def test_valid_handle_min_length(self):
|
||||
# 5 characters after @
|
||||
assert validate_telegram("@abcde") is None
|
||||
|
||||
def test_valid_handle_max_length(self):
|
||||
# 32 characters after @
|
||||
assert validate_telegram("@" + "a" * 32) is None
|
||||
|
||||
def test_missing_at_prefix(self):
|
||||
result = validate_telegram("alice")
|
||||
assert result is not None
|
||||
assert "@" in result
|
||||
|
||||
def test_just_at_sign(self):
|
||||
result = validate_telegram("@")
|
||||
assert result is not None
|
||||
|
||||
def test_too_short(self):
|
||||
# Less than 5 characters after @
|
||||
result = validate_telegram("@abcd")
|
||||
assert result is not None
|
||||
assert "5" in result
|
||||
|
||||
def test_too_long(self):
|
||||
# More than 32 characters after @
|
||||
result = validate_telegram("@" + "a" * 33)
|
||||
assert result is not None
|
||||
assert "32" in result
|
||||
|
||||
def test_starts_with_number(self):
|
||||
result = validate_telegram("@1alice")
|
||||
assert result is not None
|
||||
assert "letter" in result.lower()
|
||||
|
||||
def test_invalid_characters(self):
|
||||
result = validate_telegram("@alice-bob")
|
||||
assert result is not None
|
||||
|
||||
|
||||
class TestValidateSignal:
|
||||
"""Tests for Signal username validation."""
|
||||
|
||||
def test_none_is_valid(self):
|
||||
assert validate_signal(None) is None
|
||||
|
||||
def test_empty_string_is_valid(self):
|
||||
assert validate_signal("") is None
|
||||
|
||||
def test_valid_username(self):
|
||||
assert validate_signal("alice.42") is None
|
||||
|
||||
def test_valid_simple_username(self):
|
||||
assert validate_signal("alice") is None
|
||||
|
||||
def test_valid_with_numbers(self):
|
||||
assert validate_signal("alice123") is None
|
||||
|
||||
def test_whitespace_only_invalid(self):
|
||||
result = validate_signal(" ")
|
||||
assert result is not None
|
||||
assert "empty" in result.lower()
|
||||
|
||||
def test_too_long(self):
|
||||
result = validate_signal("a" * 65)
|
||||
assert result is not None
|
||||
assert "64" in result
|
||||
|
||||
|
||||
class TestValidateNostrNpub:
|
||||
"""Tests for Nostr npub validation."""
|
||||
|
||||
# Valid npub for testing (32 zero bytes encoded as bech32)
|
||||
VALID_NPUB = "npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqzqujme"
|
||||
|
||||
def test_none_is_valid(self):
|
||||
assert validate_nostr_npub(None) is None
|
||||
|
||||
def test_empty_string_is_valid(self):
|
||||
assert validate_nostr_npub("") is None
|
||||
|
||||
def test_valid_npub(self):
|
||||
# Generate a valid npub for testing
|
||||
# This is the npub for 32 zero bytes
|
||||
assert validate_nostr_npub(self.VALID_NPUB) is None
|
||||
|
||||
def test_wrong_prefix(self):
|
||||
result = validate_nostr_npub("nsec1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqwcv5dz")
|
||||
assert result is not None
|
||||
assert "npub" in result.lower()
|
||||
|
||||
def test_invalid_checksum(self):
|
||||
# Change last character to break checksum
|
||||
result = validate_nostr_npub("npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqsutgpd")
|
||||
assert result is not None
|
||||
assert "checksum" in result.lower()
|
||||
|
||||
def test_too_short(self):
|
||||
result = validate_nostr_npub("npub1short")
|
||||
assert result is not None
|
||||
|
||||
def test_not_starting_with_npub1(self):
|
||||
result = validate_nostr_npub("npub2qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqsutgpc")
|
||||
assert result is not None
|
||||
assert "npub1" in result
|
||||
|
||||
def test_random_string(self):
|
||||
result = validate_nostr_npub("not_a_valid_npub_at_all")
|
||||
assert result is not None
|
||||
|
||||
|
||||
class TestValidateProfileFields:
|
||||
"""Tests for validating all profile fields together."""
|
||||
|
||||
def test_all_none_is_valid(self):
|
||||
errors = validate_profile_fields()
|
||||
assert errors == {}
|
||||
|
||||
def test_all_valid(self):
|
||||
errors = validate_profile_fields(
|
||||
contact_email="user@example.com",
|
||||
telegram="@alice",
|
||||
signal="alice.42",
|
||||
nostr_npub="npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqzqujme",
|
||||
)
|
||||
assert errors == {}
|
||||
|
||||
def test_single_invalid_field(self):
|
||||
errors = validate_profile_fields(
|
||||
contact_email="user@example.com",
|
||||
telegram="alice", # Missing @
|
||||
signal="alice.42",
|
||||
)
|
||||
assert "telegram" in errors
|
||||
assert len(errors) == 1
|
||||
|
||||
def test_multiple_invalid_fields(self):
|
||||
errors = validate_profile_fields(
|
||||
contact_email="not-an-email",
|
||||
telegram="alice", # Missing @
|
||||
)
|
||||
assert "contact_email" in errors
|
||||
assert "telegram" in errors
|
||||
assert len(errors) == 2
|
||||
|
||||
def test_empty_values_are_valid(self):
|
||||
errors = validate_profile_fields(
|
||||
contact_email="",
|
||||
telegram="",
|
||||
signal="",
|
||||
nostr_npub="",
|
||||
)
|
||||
assert errors == {}
|
||||
|
||||
135
backend/validation.py
Normal file
135
backend/validation.py
Normal file
|
|
@ -0,0 +1,135 @@
|
|||
"""Validation utilities for user profile fields."""
|
||||
import re
|
||||
from email_validator import validate_email, EmailNotValidError
|
||||
from bech32 import bech32_decode
|
||||
|
||||
|
||||
def validate_contact_email(value: str | None) -> str | None:
|
||||
"""
|
||||
Validate contact email format.
|
||||
|
||||
Returns None if valid, error message if invalid.
|
||||
Empty/None values are valid (field is optional).
|
||||
"""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
try:
|
||||
validate_email(value, check_deliverability=False)
|
||||
return None
|
||||
except EmailNotValidError as e:
|
||||
return str(e)
|
||||
|
||||
|
||||
def validate_telegram(value: str | None) -> str | None:
|
||||
"""
|
||||
Validate Telegram handle.
|
||||
|
||||
Must start with @ if provided.
|
||||
Returns None if valid, error message if invalid.
|
||||
Empty/None values are valid (field is optional).
|
||||
"""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
if not value.startswith("@"):
|
||||
return "Telegram handle must start with @"
|
||||
|
||||
if len(value) < 2:
|
||||
return "Telegram handle must have at least one character after @"
|
||||
|
||||
# Telegram usernames: 5-32 characters, alphanumeric and underscores
|
||||
# But we store with @, so check 6-33 total
|
||||
handle = value[1:] # Remove @
|
||||
if len(handle) < 5:
|
||||
return "Telegram handle must be at least 5 characters (after @)"
|
||||
|
||||
if len(handle) > 32:
|
||||
return "Telegram handle must be at most 32 characters (after @)"
|
||||
|
||||
if not re.match(r'^[a-zA-Z][a-zA-Z0-9_]*$', handle):
|
||||
return "Telegram handle must start with a letter and contain only letters, numbers, and underscores"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def validate_signal(value: str | None) -> str | None:
|
||||
"""
|
||||
Validate Signal username.
|
||||
|
||||
Any non-empty string is valid.
|
||||
Returns None if valid, error message if invalid.
|
||||
Empty/None values are valid (field is optional).
|
||||
"""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
# Signal usernames are fairly permissive, just check it's not empty
|
||||
if len(value.strip()) == 0:
|
||||
return "Signal username cannot be empty"
|
||||
|
||||
if len(value) > 64:
|
||||
return "Signal username must be at most 64 characters"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def validate_nostr_npub(value: str | None) -> str | None:
|
||||
"""
|
||||
Validate Nostr npub (public key in bech32 format).
|
||||
|
||||
Must be valid bech32 with 'npub' prefix.
|
||||
Returns None if valid, error message if invalid.
|
||||
Empty/None values are valid (field is optional).
|
||||
"""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
if not value.startswith("npub1"):
|
||||
return "Nostr npub must start with 'npub1'"
|
||||
|
||||
# Decode bech32 to validate checksum
|
||||
hrp, data = bech32_decode(value)
|
||||
|
||||
if hrp is None or data is None:
|
||||
return "Invalid Nostr npub: bech32 checksum failed"
|
||||
|
||||
if hrp != "npub":
|
||||
return "Nostr npub must have 'npub' prefix"
|
||||
|
||||
# npub should decode to 32 bytes (256 bits) for a public key
|
||||
# In bech32, each character encodes 5 bits, so 32 bytes = 52 characters of data
|
||||
if len(data) != 52:
|
||||
return "Invalid Nostr npub: incorrect length"
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def validate_profile_fields(
|
||||
contact_email: str | None = None,
|
||||
telegram: str | None = None,
|
||||
signal: str | None = None,
|
||||
nostr_npub: str | None = None,
|
||||
) -> dict[str, str]:
|
||||
"""
|
||||
Validate all profile fields at once.
|
||||
|
||||
Returns a dict of field_name -> error_message for any invalid fields.
|
||||
Empty dict means all fields are valid.
|
||||
"""
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if err := validate_contact_email(contact_email):
|
||||
errors["contact_email"] = err
|
||||
|
||||
if err := validate_telegram(telegram):
|
||||
errors["telegram"] = err
|
||||
|
||||
if err := validate_signal(signal):
|
||||
errors["signal"] = err
|
||||
|
||||
if err := validate_nostr_npub(nostr_npub):
|
||||
errors["nostr_npub"] = err
|
||||
|
||||
return errors
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue