董宏凯andCursor
db333a4c33
抽出专用多轮RAG消息列表并升级会话模型；强化报告协作画布与检查器；业务链普通用户改为提交审核；扩展 agentfx 报告技能种子，并将含 admin 的用户名自动提权为管理员。 Co-authored-by: Cursor <cursoragent@cursor.com>
261 lines
9.0 KiB
Python
261 lines
9.0 KiB
Python
"""Tests for the username-based login domain handling.
|
|
|
|
Verifies that:
|
|
- The default mail domain is ``cm.com`` (with env-var override still honoured).
|
|
- ``_normalize_username_local_part`` distinguishes raw-email vs username input.
|
|
- The login flow falls back to a legacy-domain lookup so the same username
|
|
keeps mapping to the same account after the default domain changes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.gateway.auth.local_provider import LocalAuthProvider
|
|
from app.gateway.auth.models import User
|
|
from app.gateway.auth.repositories.base import UserRepository
|
|
from app.gateway.routers import auth as auth_module
|
|
from app.gateway.routers.auth import (
|
|
_get_or_create_user_by_username,
|
|
_lookup_user_by_username,
|
|
_normalize_username_local_part,
|
|
_username_to_email,
|
|
)
|
|
|
|
|
|
class FakeUserRepository(UserRepository):
|
|
"""In-memory repo just enough to exercise the local-part lookup."""
|
|
|
|
def __init__(self) -> None:
|
|
self.rows: dict[str, User] = {}
|
|
self.create_calls: list[User] = []
|
|
|
|
async def create_user(self, user: User) -> User:
|
|
for existing in self.rows.values():
|
|
if existing.email == user.email:
|
|
raise ValueError(f"Email already registered: {user.email}")
|
|
u = User(
|
|
id=user.id or uuid4(),
|
|
email=user.email,
|
|
password_hash=user.password_hash,
|
|
system_role=user.system_role,
|
|
created_at=user.created_at or datetime.now(UTC),
|
|
oauth_provider=user.oauth_provider,
|
|
oauth_id=user.oauth_id,
|
|
needs_setup=user.needs_setup,
|
|
token_version=user.token_version,
|
|
)
|
|
self.rows[str(u.id)] = u
|
|
self.create_calls.append(u)
|
|
return u
|
|
|
|
async def get_user_by_id(self, user_id: str) -> User | None:
|
|
return self.rows.get(user_id)
|
|
|
|
async def get_user_by_email(self, email: str) -> User | None:
|
|
for user in self.rows.values():
|
|
if user.email == email:
|
|
return user
|
|
return None
|
|
|
|
async def find_user_by_email_local_part(self, local_part: str) -> User | None:
|
|
prefix = f"{(local_part or '').strip().lower()}@"
|
|
candidates = [u for u in self.rows.values() if u.email.lower().startswith(prefix)]
|
|
candidates.sort(key=lambda u: u.created_at)
|
|
return candidates[0] if candidates else None
|
|
|
|
async def update_user(self, user: User) -> User:
|
|
self.rows[str(user.id)] = user
|
|
return user
|
|
|
|
async def count_users(self) -> int:
|
|
return len(self.rows)
|
|
|
|
async def count_admin_users(self) -> int:
|
|
return sum(1 for u in self.rows.values() if u.system_role == "admin")
|
|
|
|
async def list_users(self) -> list[User]:
|
|
return sorted(self.rows.values(), key=lambda user: user.created_at)
|
|
|
|
async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
|
|
return None
|
|
|
|
|
|
def _seed(repo: FakeUserRepository, email: str, *, role: str = "user") -> User:
|
|
user = User(
|
|
id=uuid4(),
|
|
email=email,
|
|
password_hash=None,
|
|
system_role=role, # type: ignore[arg-type]
|
|
created_at=datetime.now(UTC),
|
|
needs_setup=False,
|
|
)
|
|
repo.rows[str(user.id)] = user
|
|
return user
|
|
|
|
|
|
def test_normalize_returns_none_for_explicit_email() -> None:
|
|
assert _normalize_username_local_part("alice@old.com") is None
|
|
|
|
|
|
def test_normalize_lowercases_and_keeps_safe_chars() -> None:
|
|
assert _normalize_username_local_part("Alice_Smith.1") == "alice_smith.1"
|
|
|
|
|
|
def test_normalize_strips_unsafe_chars() -> None:
|
|
assert _normalize_username_local_part("hi there!") == "hi-there"
|
|
|
|
|
|
def test_username_to_email_uses_cm_com_by_default(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.delenv("DEER_FLOW_USERNAME_LOGIN_DOMAIN", raising=False)
|
|
assert _username_to_email("alice") == "alice@cm.com"
|
|
|
|
|
|
def test_username_to_email_honours_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("DEER_FLOW_USERNAME_LOGIN_DOMAIN", "example.org")
|
|
assert _username_to_email("alice") == "alice@example.org"
|
|
|
|
|
|
def test_username_to_email_passes_explicit_email_through() -> None:
|
|
assert _username_to_email("bob@legacy.io") == "bob@legacy.io"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_part_lookup_finds_legacy_domain_user() -> None:
|
|
"""Same username under legacy domain should match the new lookup."""
|
|
repo = FakeUserRepository()
|
|
_seed(repo, "alice@local.deerflow")
|
|
legacy = await repo.find_user_by_email_local_part("alice")
|
|
assert legacy is not None
|
|
assert legacy.email == "alice@local.deerflow"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_part_lookup_returns_oldest_when_collision() -> None:
|
|
"""When the same local part exists under multiple domains, prefer the
|
|
oldest row so re-logins land on the original account."""
|
|
repo = FakeUserRepository()
|
|
older = _seed(repo, "alice@local.deerflow")
|
|
older.created_at = datetime(2024, 1, 1, tzinfo=UTC)
|
|
newer = _seed(repo, "alice@cm.com")
|
|
newer.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
match = await repo.find_user_by_email_local_part("alice")
|
|
assert match is not None
|
|
assert match.email == "alice@local.deerflow"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_local_part_lookup_misses_unknown_username() -> None:
|
|
repo = FakeUserRepository()
|
|
_seed(repo, "alice@cm.com")
|
|
assert await repo.find_user_by_email_local_part("bob") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_username_login_uses_oldest_account_when_canonical_duplicate_exists(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A domain migration must not make one username alternate user ids."""
|
|
repo = FakeUserRepository()
|
|
older = _seed(repo, "alice@local.deerflow")
|
|
older.created_at = datetime(2024, 1, 1, tzinfo=UTC)
|
|
newer = _seed(repo, "alice@cm.com")
|
|
newer.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user, created = await _get_or_create_user_by_username("alice")
|
|
|
|
assert created is False
|
|
assert user.id == older.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_password_lookup_uses_same_stable_username_account(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repo = FakeUserRepository()
|
|
older = _seed(repo, "alice@local.deerflow")
|
|
older.created_at = datetime(2024, 1, 1, tzinfo=UTC)
|
|
newer = _seed(repo, "alice@cm.com")
|
|
newer.created_at = datetime(2026, 1, 1, tzinfo=UTC)
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user = await _lookup_user_by_username("alice")
|
|
|
|
assert user is not None
|
|
assert user.id == older.id
|
|
|
|
|
|
# ── 用户名含 admin → 自动管理员（注册即 admin / 历史账号登录时提升）───────────
|
|
|
|
|
|
def test_admin_username_rule_matches_substring_case_insensitive() -> None:
|
|
from app.gateway.routers.auth import _is_admin_username
|
|
|
|
assert _is_admin_username("admin") is True
|
|
assert _is_admin_username("b1admin") is True
|
|
assert _is_admin_username("Admin_Planning") is True
|
|
assert _is_admin_username("administrator") is True
|
|
assert _is_admin_username("zhangsan") is False
|
|
assert _is_admin_username("") is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_admin_substring_username_registers_as_admin(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repo = FakeUserRepository()
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user, created = await _get_or_create_user_by_username("b1admin")
|
|
|
|
assert created is True
|
|
assert user.system_role == "admin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_plain_username_still_registers_as_user(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repo = FakeUserRepository()
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user, created = await _get_or_create_user_by_username("zhangsan")
|
|
|
|
assert created is True
|
|
assert user.system_role == "user"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_admin_substring_account_promoted_on_login(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""历史已注册成普通用户的含 admin 账号，登录时自动提升为管理员。"""
|
|
repo = FakeUserRepository()
|
|
seeded = _seed(repo, "b1admin@cm.com", role="user")
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user, created = await _get_or_create_user_by_username("b1admin")
|
|
|
|
assert created is False
|
|
assert user.id == seeded.id
|
|
assert user.system_role == "admin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_plain_account_not_promoted_on_login(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
repo = FakeUserRepository()
|
|
seeded = _seed(repo, "zhangsan@cm.com", role="user")
|
|
monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
|
|
|
|
user, created = await _get_or_create_user_by_username("zhangsan")
|
|
|
|
assert created is False
|
|
assert user.id == seeded.id
|
|
assert user.system_role == "user"
|