forked from cartsnitch/cartsnitch
feat(common): sync email_inbound_token from standalone repo
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
"""Add email_inbound_token to users.
|
||||
|
||||
Revision ID: 001_add_email_inbound_token
|
||||
Revises:
|
||||
Create Date: 2026-04-02
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
|
||||
from alembic import op
|
||||
|
||||
revision: str = "001_add_email_inbound_token"
|
||||
down_revision: str | None = None
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
op.add_column("users", sa.Column("email_inbound_token", sa.String(22), nullable=True))
|
||||
op.create_unique_constraint("uq_users_email_inbound_token", "users", ["email_inbound_token"])
|
||||
|
||||
# Backfill existing users with generated tokens (PostgreSQL)
|
||||
op.execute(
|
||||
"UPDATE users SET email_inbound_token = "
|
||||
"substring(replace(gen_random_uuid()::text, '-', ''), 1, 22) "
|
||||
"WHERE email_inbound_token IS NULL"
|
||||
)
|
||||
|
||||
# Alter to non-nullable
|
||||
op.alter_column("users", "email_inbound_token", nullable=False)
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
op.drop_constraint("uq_users_email_inbound_token", "users", type_="unique")
|
||||
op.drop_column("users", "email_inbound_token")
|
||||
@@ -1,5 +1,6 @@
|
||||
"""User and UserStoreAccount models."""
|
||||
|
||||
import secrets
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import TYPE_CHECKING
|
||||
@@ -21,6 +22,9 @@ class User(UUIDPrimaryKeyMixin, TimestampMixin, Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
email: Mapped[str] = mapped_column(String(255), nullable=False, unique=True)
|
||||
email_inbound_token: Mapped[str] = mapped_column(
|
||||
String(22), nullable=False, unique=True, default=lambda: secrets.token_urlsafe(16)
|
||||
)
|
||||
hashed_password: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
display_name: Mapped[str | None] = mapped_column(String(100))
|
||||
email_verified: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default="false")
|
||||
|
||||
@@ -20,6 +20,7 @@ class UserRead(BaseModel):
|
||||
id: uuid.UUID
|
||||
email: str
|
||||
display_name: str | None
|
||||
email_inbound_token: str
|
||||
created_at: datetime
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
@@ -147,6 +147,40 @@ class TestStoreLocationModel:
|
||||
assert loc.lat == pytest.approx(42.2808)
|
||||
|
||||
|
||||
class TestUserModel:
|
||||
def test_email_inbound_token_auto_populated(self, session):
|
||||
user = User(
|
||||
id=uuid.uuid4(),
|
||||
email="token_test@example.com",
|
||||
hashed_password="hashed",
|
||||
created_at=datetime.now(UTC),
|
||||
updated_at=datetime.now(UTC),
|
||||
)
|
||||
session.add(user)
|
||||
session.commit()
|
||||
assert user.email_inbound_token is not None
|
||||
assert len(user.email_inbound_token) == 22
|
||||
|
||||
def test_email_inbound_token_unique(self, session):
|
||||
user1 = User(
|
||||
id=uuid.uuid4(),
|
||||
email="user1@example.com",
|
||||
hashed_password="hashed",
|
||||
created_at=datetime.now(UTC),
|
||||
updated_at=datetime.now(UTC),
|
||||
)
|
||||
user2 = User(
|
||||
id=uuid.uuid4(),
|
||||
email="user2@example.com",
|
||||
hashed_password="hashed",
|
||||
created_at=datetime.now(UTC),
|
||||
updated_at=datetime.now(UTC),
|
||||
)
|
||||
session.add_all([user1, user2])
|
||||
session.commit()
|
||||
assert user1.email_inbound_token != user2.email_inbound_token
|
||||
|
||||
|
||||
class TestUserStoreAccountModel:
|
||||
def test_account_status_enum(self, session):
|
||||
user = User(
|
||||
|
||||
Reference in New Issue
Block a user