mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-03-07 14:33:09 +00:00
fix: second round review fixes for account merge
- Rename _compute_auth_methods to compute_auth_methods (public API) - Add Literal type to _handle_subscription_merge param - Add Literal type to keep_from in route handler - Add Path(min_length=32, max_length=64) on merge_token params - Import Path and Literal in account_linking routes
This commit is contained in:
@@ -5,10 +5,10 @@ Router 2 (`merge_router`): Public endpoints for merge preview and execution.
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
from typing import Any, Literal
|
||||
|
||||
import structlog
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from fastapi import APIRouter, Depends, HTTPException, Path, status
|
||||
from pydantic import BaseModel, Field
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
@@ -18,7 +18,7 @@ from app.database.crud.user import (
|
||||
set_user_oauth_provider_id,
|
||||
)
|
||||
from app.database.models import User
|
||||
from app.services.account_merge_service import _compute_auth_methods, execute_merge, get_merge_preview
|
||||
from app.services.account_merge_service import compute_auth_methods, execute_merge, get_merge_preview
|
||||
|
||||
from ..auth.merge_service import (
|
||||
MERGE_TOKEN_TTL_SECONDS,
|
||||
@@ -137,7 +137,7 @@ def _get_provider_identifier(user: User, provider: str) -> str | None:
|
||||
|
||||
def _count_auth_methods(user: User) -> int:
|
||||
"""Count how many auth methods the user has linked."""
|
||||
return len(_compute_auth_methods(user))
|
||||
return len(compute_auth_methods(user))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -364,7 +364,7 @@ merge_router = APIRouter(prefix='/auth/merge', tags=['Cabinet Account Merge'])
|
||||
|
||||
@merge_router.get('/{merge_token}', response_model=MergePreviewResponse)
|
||||
async def get_merge_preview_endpoint(
|
||||
merge_token: str,
|
||||
merge_token: str = Path(..., min_length=32, max_length=64),
|
||||
db: AsyncSession = Depends(get_cabinet_db),
|
||||
) -> MergePreviewResponse:
|
||||
"""Preview the result of merging two accounts before confirming."""
|
||||
@@ -407,8 +407,8 @@ async def get_merge_preview_endpoint(
|
||||
|
||||
@merge_router.post('/{merge_token}', response_model=MergeResponse)
|
||||
async def execute_merge_endpoint(
|
||||
merge_token: str,
|
||||
request: MergeRequest,
|
||||
merge_token: str = Path(..., min_length=32, max_length=64),
|
||||
db: AsyncSession = Depends(get_cabinet_db),
|
||||
) -> MergeResponse:
|
||||
"""Execute account merge. Consumes the merge token (one-time use)."""
|
||||
@@ -433,7 +433,7 @@ async def execute_merge_endpoint(
|
||||
)
|
||||
|
||||
# Convert user_id to 'primary'/'secondary' string for execute_merge()
|
||||
keep_from: str = 'primary' if request.keep_subscription_from == primary_user_id else 'secondary'
|
||||
keep_from: Literal['primary', 'secondary'] = 'primary' if request.keep_subscription_from == primary_user_id else 'secondary'
|
||||
|
||||
# 3. Execute merge
|
||||
try:
|
||||
|
||||
@@ -60,7 +60,7 @@ _PARTNER_STATUS_PRIORITY: dict[str, int] = {
|
||||
}
|
||||
|
||||
|
||||
def _compute_auth_methods(user: User) -> list[str]:
|
||||
def compute_auth_methods(user: User) -> list[str]:
|
||||
"""Вычисляет список методов авторизации пользователя."""
|
||||
methods: list[str] = []
|
||||
if user.telegram_id:
|
||||
@@ -104,7 +104,7 @@ def _build_user_preview(user: User) -> dict[str, Any]:
|
||||
'username': user.username,
|
||||
'first_name': user.first_name,
|
||||
'email': user.email,
|
||||
'auth_methods': _compute_auth_methods(user),
|
||||
'auth_methods': compute_auth_methods(user),
|
||||
'balance_kopeks': user.balance_kopeks,
|
||||
'subscription': _build_subscription_preview(user.subscription),
|
||||
'created_at': user.created_at,
|
||||
@@ -214,7 +214,7 @@ async def _handle_subscription_merge(
|
||||
db: AsyncSession,
|
||||
primary: User,
|
||||
secondary: User,
|
||||
keep_subscription_from: str,
|
||||
keep_subscription_from: Literal['primary', 'secondary'],
|
||||
) -> None:
|
||||
"""Обрабатывает мерж подписок между двумя аккаунтами.
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ from app.services import account_merge_service
|
||||
from app.services.account_merge_service import (
|
||||
_build_subscription_preview,
|
||||
_build_user_preview,
|
||||
_compute_auth_methods,
|
||||
compute_auth_methods,
|
||||
execute_merge,
|
||||
get_merge_preview,
|
||||
)
|
||||
@@ -105,26 +105,26 @@ def _make_db() -> SimpleNamespace:
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _compute_auth_methods
|
||||
# compute_auth_methods
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestComputeAuthMethods:
|
||||
def test_no_methods(self):
|
||||
user = _make_user()
|
||||
assert _compute_auth_methods(user) == []
|
||||
assert compute_auth_methods(user) == []
|
||||
|
||||
def test_telegram_only(self):
|
||||
user = _make_user(telegram_id=12345)
|
||||
assert _compute_auth_methods(user) == ['telegram']
|
||||
assert compute_auth_methods(user) == ['telegram']
|
||||
|
||||
def test_email_only(self):
|
||||
user = _make_user(email='test@example.com', password_hash='hash123')
|
||||
assert _compute_auth_methods(user) == ['email']
|
||||
assert compute_auth_methods(user) == ['email']
|
||||
|
||||
def test_email_without_password_not_counted(self):
|
||||
user = _make_user(email='test@example.com')
|
||||
assert _compute_auth_methods(user) == []
|
||||
assert compute_auth_methods(user) == []
|
||||
|
||||
def test_all_methods(self):
|
||||
user = _make_user(
|
||||
@@ -136,11 +136,11 @@ class TestComputeAuthMethods:
|
||||
discord_id='d123',
|
||||
vk_id=99999,
|
||||
)
|
||||
assert _compute_auth_methods(user) == ['telegram', 'email', 'google', 'yandex', 'discord', 'vk']
|
||||
assert compute_auth_methods(user) == ['telegram', 'email', 'google', 'yandex', 'discord', 'vk']
|
||||
|
||||
def test_oauth_only(self):
|
||||
user = _make_user(google_id='g123', discord_id='d123')
|
||||
assert _compute_auth_methods(user) == ['google', 'discord']
|
||||
assert compute_auth_methods(user) == ['google', 'discord']
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user