Xiaoxia AI
143 lines
4.1 KiB
Python
143 lines
4.1 KiB
Python
import sys
|
|
from pathlib import Path
|
|
|
|
import jwt
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
API_ROOT = ROOT / "apps" / "api"
|
|
if str(API_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(API_ROOT))
|
|
|
|
from app.config import settings
|
|
|
|
from packages.application.auth.login_use_case import LoginRequest, LoginUseCase
|
|
from packages.application.auth.register_user_use_case import RegisterUserRequest, RegisterUserUseCase
|
|
from packages.domain.auth import password_hasher
|
|
from packages.domain.entities import User
|
|
|
|
|
|
class InMemoryUserRepository:
|
|
def __init__(self):
|
|
self.users = {}
|
|
|
|
def save(self, user):
|
|
self.users[user.id] = user
|
|
|
|
def find_by_id(self, user_id):
|
|
return self.users.get(user_id)
|
|
|
|
def find_by_email(self, email):
|
|
return next((user for user in self.users.values() if user.email == email), None)
|
|
|
|
def find_by_username(self, username):
|
|
return next((user for user in self.users.values() if user.username == username), None)
|
|
|
|
def find_by_verification_token(self, token):
|
|
return None
|
|
|
|
def find_by_password_reset_token(self, token):
|
|
return None
|
|
|
|
def delete(self, user_id):
|
|
return self.users.pop(user_id, None) is not None
|
|
|
|
|
|
class DummySessionStore:
|
|
def __init__(self):
|
|
self.saved = []
|
|
|
|
def save_session(self, **kwargs):
|
|
self.saved.append(kwargs)
|
|
|
|
|
|
class DummyEmailService:
|
|
def send_verification_email(self, **kwargs):
|
|
return False, "disabled"
|
|
|
|
|
|
def test_register_use_case_hashes_password_and_normalizes_email():
|
|
repo = InMemoryUserRepository()
|
|
use_case = RegisterUserUseCase(repo, email_service=DummyEmailService())
|
|
|
|
response, error = use_case.execute(
|
|
RegisterUserRequest(
|
|
email="USER@EXAMPLE.COM",
|
|
password="Password1",
|
|
username="user",
|
|
display_name="User",
|
|
)
|
|
)
|
|
|
|
assert error is None
|
|
assert response is not None
|
|
user = repo.find_by_email("user@example.com")
|
|
assert user is not None
|
|
assert user.password_hash.startswith("$2")
|
|
assert password_hasher.verify_password("Password1", user.password_hash)
|
|
|
|
|
|
def test_login_use_case_returns_verifiable_jwt():
|
|
repo = InMemoryUserRepository()
|
|
session_store = DummySessionStore()
|
|
user = User(
|
|
id="user-1",
|
|
email="user@example.com",
|
|
username="user",
|
|
display_name="User",
|
|
password_hash=password_hasher.hash_password("Password1"),
|
|
)
|
|
repo.save(user)
|
|
|
|
response, error = LoginUseCase(repo, session_store=session_store, jwt_secret_key=settings.JWT_SECRET_KEY).execute(
|
|
LoginRequest("user@example.com", "Password1")
|
|
)
|
|
|
|
assert error is None
|
|
assert response is not None
|
|
payload = jwt.decode(response.access_token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
|
|
assert response.expires_in == 1800
|
|
assert payload["sub"] == "user-1"
|
|
assert payload["type"] == "user_auth"
|
|
assert session_store.saved
|
|
|
|
|
|
def test_login_use_case_upgrades_legacy_sha256_hash():
|
|
repo = InMemoryUserRepository()
|
|
user = User(
|
|
id="user-1",
|
|
email="user@example.com",
|
|
username="user",
|
|
display_name="User",
|
|
password_hash="19513fdc9da4fb72a4a05eb66917548d3c90ff94d5419e1f2363eea89dfee1dd",
|
|
)
|
|
repo.save(user)
|
|
|
|
response, error = LoginUseCase(
|
|
repo, session_store=DummySessionStore(), jwt_secret_key=settings.JWT_SECRET_KEY
|
|
).execute(LoginRequest("user@example.com", "Password1"))
|
|
|
|
assert error is None
|
|
assert response is not None
|
|
assert user.password_hash.startswith("$2")
|
|
assert password_hasher.verify_password("Password1", user.password_hash)
|
|
|
|
|
|
def test_login_use_case_rejects_wrong_password():
|
|
repo = InMemoryUserRepository()
|
|
repo.save(
|
|
User(
|
|
id="user-1",
|
|
email="user@example.com",
|
|
username="user",
|
|
display_name="User",
|
|
password_hash=password_hasher.hash_password("Password1"),
|
|
)
|
|
)
|
|
|
|
response, error = LoginUseCase(repo, session_store=DummySessionStore()).execute(
|
|
LoginRequest("user@example.com", "WrongPassword1")
|
|
)
|
|
|
|
assert response is None
|
|
assert error == "Invalid email or password"
|