SunnyQjm/dify
1
0 Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-10-04 13:47:55 +08:00
Code Issues Projects Releases Wiki Activity

refactor(api): move web authentication into application services (#41318)

Browse Source 
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Byron.wangandautofix-ci[bot] authored and GitHub committed 2026-09-30 06:47:17 +00:00
1 parent bf41e733ba
commit 6decc78fd5
28 files changed
+3179 -1657
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/tests/unit_tests/controllers/web/test_web_forgot_password.py
+148 -211
View File
@@ -1,234 +1,171 @@
"""Unit tests for controllers.web.forgot_password endpoints."""
"""Controller-boundary tests for Web forgot-password endpoints."""
from __future__ import annotations
import base64
from collections.abc import Iterator
from unittest.mock import ANY, patch
from dataclasses import dataclass
from inspect import unwrap
from typing import override
import pytest
from flask import Flask
from controllers.console.auth.error import EmailCodeError, InvalidTokenError, PasswordMismatchError
from controllers.console.error import EmailSendIpLimitError
from controllers.web import forgot_password
from controllers.web.forgot_password import (
ForgotPasswordCheckApi,
ForgotPasswordResetApi,
ForgotPasswordSendEmailApi,
)
from enums import DeploymentEdition
from models.account import Account
from models.engine import db
from services.entities.feature_entities import SystemFeatureModel
from tests.unit_tests.config_override import config_overrides_context
from machinery.context import RequestContext
from services.entities.auth_entities import (
ForgotPasswordCheckPayload,
ForgotPasswordResetPayload,
ForgotPasswordSendPayload,
)
from services.web_authentication_service import (
WebEmailSendIPLimitedError,
WebInvalidCodeError,
WebInvalidTokenError,
WebPasswordMismatchError,
)
@pytest.fixture
def database_app() -> Iterator[Flask]:
app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db.init_app(app)
with app.app_context():
Account.__table__.create(db.engine)
yield app
def context() -> RequestContext:
return RequestContext("request-1", "trace-1", "", "", "127.0.0.1")
@pytest.fixture(autouse=True)
def _patch_wraps():
wraps_features = SystemFeatureModel(
deployment_edition=DeploymentEdition.COMMUNITY,
enable_email_password_login=True,
@dataclass(frozen=True, slots=True)
class ApplicationServicesStub:
web_authentication: object
def bind_service(monkeypatch: pytest.MonkeyPatch, service: object) -> None:
monkeypatch.setattr(
forgot_password,
"application_services",
lambda: ApplicationServicesStub(web_authentication=service),
)
with (
patch("controllers.console.wraps.db") as mock_db,
config_overrides_context(DEPLOYMENT_EDITION=DeploymentEdition.ENTERPRISE),
patch(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
return_value=wraps_features.enable_email_password_login,
class ForgotPasswordStub:
def __init__(self) -> None:
self.reset_call: tuple[str, str, str] | None = None
def send_reset_password_email(self, context: RequestContext, *, email: str, language: str | None) -> str:
assert context.remote_ip == "127.0.0.1"
assert (email, language) == ("User@Example.com", "zh-Hans")
return "verification-token"
def verify_reset_password_code(self, *, email: str, code: str, token: str) -> str:
assert (email, code, token) == ("User@Example.com", "1234", "verification-token")
return "reset-token"
def reset_password(self, *, token: str, new_password: str, password_confirmation: str) -> None:
self.reset_call = (token, new_password, password_confirmation)
def test_forgot_password_endpoints_delegate_and_serialize(
monkeypatch: pytest.MonkeyPatch,
context: RequestContext,
) -> None:
service = ForgotPasswordStub()
bind_service(monkeypatch, service)
send_result = unwrap(ForgotPasswordSendEmailApi.post)(
ForgotPasswordSendEmailApi(),
ForgotPasswordSendPayload(email="User@Example.com", language="zh-Hans"),
context,
)
check_result = unwrap(ForgotPasswordCheckApi.post)(
ForgotPasswordCheckApi(),
ForgotPasswordCheckPayload(email="User@Example.com", code="1234", token="verification-token"),
context,
)
reset_result = unwrap(ForgotPasswordResetApi.post)(
ForgotPasswordResetApi(),
ForgotPasswordResetPayload(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
),
):
yield
context,
)
assert send_result == {"result": "success", "data": "verification-token"}
assert check_result == {"is_valid": True, "email": "user@example.com", "token": "reset-token"}
assert reset_result == {"result": "success"}
assert service.reset_call == ("reset-token", "ValidPass123!", "ValidPass123!")
class TestForgotPasswordSendEmailApi:
@patch("controllers.web.forgot_password.AccountService.send_reset_password_email")
@patch("controllers.web.forgot_password.AccountService.get_account_by_email_with_case_fallback")
@patch("controllers.web.forgot_password.AccountService.is_email_send_ip_limit", return_value=False)
@patch("controllers.web.forgot_password.extract_remote_ip", return_value="127.0.0.1")
def test_should_normalize_email_before_sending(
self,
mock_extract_ip,
mock_rate_limit,
mock_get_account,
mock_send_mail,
app: Flask,
):
account = Account(name="User", email="user@example.com")
mock_get_account.return_value = account
mock_send_mail.return_value = "token-123"
@pytest.mark.parametrize(
("method", "service_error", "http_error"),
[
pytest.param("send", WebEmailSendIPLimitedError(), EmailSendIpLimitError, id="ip-limit"),
pytest.param("check", WebInvalidCodeError(), EmailCodeError, id="invalid-code"),
pytest.param("check", WebInvalidTokenError(), InvalidTokenError, id="invalid-token"),
pytest.param("reset", WebPasswordMismatchError(), PasswordMismatchError, id="password-mismatch"),
],
)
def test_forgot_password_translates_service_errors(
monkeypatch: pytest.MonkeyPatch,
context: RequestContext,
method: str,
service_error: Exception,
http_error: type[Exception],
) -> None:
class ErrorService(ForgotPasswordStub):
@override
def send_reset_password_email(
self,
context: RequestContext,
*,
email: str,
language: str | None,
) -> str:
if method == "send":
raise service_error
return super().send_reset_password_email(context, email=email, language=language)
with app.test_request_context(
"/web/forgot-password",
method="POST",
json={"email": "User@Example.com", "language": "zh-Hans"},
):
response = ForgotPasswordSendEmailApi().post()
@override
def verify_reset_password_code(self, *, email: str, code: str, token: str) -> str:
if method == "check":
raise service_error
return super().verify_reset_password_code(email=email, code=code, token=token)
assert response == {"result": "success", "data": "token-123"}
mock_get_account.assert_called_once_with("User@Example.com", session=ANY)
mock_send_mail.assert_called_once_with(account=account, email="user@example.com", language="zh-Hans")
mock_extract_ip.assert_called_once()
mock_rate_limit.assert_called_once_with("127.0.0.1")
@override
def reset_password(self, *, token: str, new_password: str, password_confirmation: str) -> None:
if method == "reset":
raise service_error
super().reset_password(
token=token,
new_password=new_password,
password_confirmation=password_confirmation,
)
bind_service(monkeypatch, ErrorService())
class TestForgotPasswordCheckApi:
@patch("controllers.web.forgot_password.AccountService.reset_forgot_password_error_rate_limit")
@patch("controllers.web.forgot_password.AccountService.generate_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.add_forgot_password_error_rate_limit")
@patch("controllers.web.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.web.forgot_password.AccountService.is_forgot_password_error_rate_limit")
def test_should_normalize_email_for_validity_checks(
self,
mock_is_rate_limit,
mock_get_data,
mock_add_rate,
mock_revoke_token,
mock_generate_token,
mock_reset_rate,
app: Flask,
):
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "User@Example.com", "code": "1234"}
mock_generate_token.return_value = (None, "new-token")
def invoke() -> None:
if method == "send":
unwrap(ForgotPasswordSendEmailApi.post)(
ForgotPasswordSendEmailApi(),
ForgotPasswordSendPayload(email="User@Example.com", language="zh-Hans"),
context,
)
elif method == "check":
unwrap(ForgotPasswordCheckApi.post)(
ForgotPasswordCheckApi(),
ForgotPasswordCheckPayload(email="User@Example.com", code="1234", token="verification-token"),
context,
)
else:
unwrap(ForgotPasswordResetApi.post)(
ForgotPasswordResetApi(),
ForgotPasswordResetPayload(
token="reset-token",
new_password="ValidPass123!",
password_confirm="ValidPass123!",
),
context,
)
with app.test_request_context(
"/web/forgot-password/validity",
method="POST",
json={"email": "User@Example.com", "code": "1234", "token": "token-123"},
):
response = ForgotPasswordCheckApi().post()
assert response == {"is_valid": True, "email": "user@example.com", "token": "new-token"}
mock_is_rate_limit.assert_called_once_with("user@example.com")
mock_add_rate.assert_not_called()
mock_revoke_token.assert_called_once_with("token-123")
mock_generate_token.assert_called_once_with(
"User@Example.com",
code="1234",
additional_data={"phase": "reset"},
)
mock_reset_rate.assert_called_once_with("user@example.com")
@patch("controllers.web.forgot_password.AccountService.reset_forgot_password_error_rate_limit")
@patch("controllers.web.forgot_password.AccountService.generate_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.web.forgot_password.AccountService.is_forgot_password_error_rate_limit")
def test_should_preserve_token_email_case(
self,
mock_is_rate_limit,
mock_get_data,
mock_revoke_token,
mock_generate_token,
mock_reset_rate,
app: Flask,
):
mock_is_rate_limit.return_value = False
mock_get_data.return_value = {"email": "MixedCase@Example.com", "code": "5678"}
mock_generate_token.return_value = (None, "fresh-token")
with app.test_request_context(
"/web/forgot-password/validity",
method="POST",
json={"email": "mixedcase@example.com", "code": "5678", "token": "token-upper"},
):
response = ForgotPasswordCheckApi().post()
assert response == {"is_valid": True, "email": "mixedcase@example.com", "token": "fresh-token"}
mock_generate_token.assert_called_once_with(
"MixedCase@Example.com",
code="5678",
additional_data={"phase": "reset"},
)
mock_revoke_token.assert_called_once_with("token-upper")
mock_reset_rate.assert_called_once_with("mixedcase@example.com")
class TestForgotPasswordResetApi:
@patch("controllers.web.forgot_password.ForgotPasswordResetApi._update_existing_account")
@patch("controllers.web.forgot_password.AccountService.get_account_by_email_with_case_fallback")
@patch("controllers.web.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.get_reset_password_data")
def test_should_fetch_account_with_fallback(
self,
mock_get_reset_data,
mock_revoke_token,
mock_get_account,
mock_update_account,
database_app: Flask,
):
mock_get_reset_data.return_value = {"phase": "reset", "email": "User@Example.com", "code": "1234"}
account = Account(name="User", email="user@example.com")
db.session.add(account)
db.session.commit()
mock_get_account.return_value = account
with database_app.test_request_context(
"/web/forgot-password/resets",
method="POST",
json={
"token": "token-123",
"new_password": "ValidPass123!",
"password_confirm": "ValidPass123!",
},
):
response = ForgotPasswordResetApi().post()
assert response == {"result": "success"}
mock_get_account.assert_called_once_with("User@Example.com", session=ANY)
mock_update_account.assert_called_once()
mock_revoke_token.assert_called_once_with("token-123")
@patch("controllers.web.forgot_password.hash_password", return_value=b"hashed-value")
@patch("controllers.web.forgot_password.secrets.token_bytes", return_value=b"0123456789abcdef")
@patch("controllers.web.forgot_password.AccountService.revoke_reset_password_token")
@patch("controllers.web.forgot_password.AccountService.get_reset_password_data")
@patch("controllers.web.forgot_password.AccountService.get_account_by_email_with_case_fallback")
def test_should_update_password_and_commit(
self,
mock_get_account,
mock_get_reset_data,
mock_revoke_token,
mock_token_bytes,
mock_hash_password,
database_app: Flask,
):
mock_get_reset_data.return_value = {"phase": "reset", "email": "user@example.com"}
account = Account(name="User", email="user@example.com")
db.session.add(account)
db.session.commit()
mock_get_account.return_value = account
with database_app.test_request_context(
"/web/forgot-password/resets",
method="POST",
json={
"token": "reset-token",
"new_password": "StrongPass123!",
"password_confirm": "StrongPass123!",
},
):
response = ForgotPasswordResetApi().post()
assert response == {"result": "success"}
mock_get_reset_data.assert_called_once_with("reset-token")
mock_revoke_token.assert_called_once_with("reset-token")
mock_token_bytes.assert_called_once_with(16)
mock_hash_password.assert_called_once_with("StrongPass123!", b"0123456789abcdef")
db.session.expire_all()
stored_account = db.session.get(Account, account.id)
assert stored_account is not None
expected_password = base64.b64encode(b"hashed-value").decode()
assert stored_account.password == expected_password
expected_salt = base64.b64encode(b"0123456789abcdef").decode()
assert stored_account.password_salt == expected_salt
with pytest.raises(http_error):
invoke()
api/tests/unit_tests/controllers/web/test_web_login.py
+221 -296
View File
@@ -1,321 +1,246 @@
import base64
import logging
from types import SimpleNamespace
from unittest.mock import ANY, MagicMock, patch
"""Controller-boundary tests for Web login endpoints."""
from dataclasses import dataclass
from inspect import unwrap
from typing import override
import pytest
from flask import Flask
from jwt import InvalidTokenError
from sqlalchemy.engine import Engine
from sqlalchemy.orm import Session, scoped_session, sessionmaker
from werkzeug.exceptions import Unauthorized
from flask import Flask, Response
from flask_restx import Api
import services.errors.account
from controllers.console import wraps as console_wraps
from controllers.web.login import EmailCodeLoginApi, EmailCodeLoginSendEmailApi, LoginApi, LoginStatusApi, LogoutApi
from enums import DeploymentEdition
from models.account import Account
from models.model import DifySetup
from services.entities.auth_audit_entities import LoginFailureReason
pytestmark = [
pytest.mark.parametrize("sqlite_session", [(DifySetup,)], indirect=True),
pytest.mark.usefixtures("app_query_services"),
]
def encode_code(code: str) -> str:
return base64.b64encode(code.encode("utf-8")).decode()
def assert_login_failure_logged(caplog: pytest.LogCaptureFixture, email: str, reason: LoginFailureReason) -> None:
records = [record for record in caplog.records if record.name == "controllers.web.login"]
assert len(records) == 1
assert records[0].args[0] == email
assert records[0].args[1] == reason
from controllers.console.auth.error import AuthenticationFailedError, EmailCodeError
from controllers.console.error import AccountBannedError
from controllers.web import login
from controllers.web.login import (
EmailCodeLoginApi,
EmailCodeLoginSendEmailApi,
EmailCodeLoginSendPayload,
EmailCodeLoginVerifyPayload,
LoginApi,
LoginPayload,
LoginStatusApi,
LoginStatusQuery,
LogoutApi,
)
from machinery.context import RequestContext
from services.entities.authentication_entities import WebLoginStatus
from services.web_authentication_service import (
WebAccountBannedError,
WebAuthenticationFailedError,
WebInvalidCodeError,
)
from services.webapp_access_query_service import WebAppAccessAppNotFoundError, WebAppAccessUnavailableError
@pytest.fixture
def app():
flask_app = Flask(__name__)
flask_app.config["TESTING"] = True
return flask_app
def app() -> Flask:
return Flask(__name__)
@pytest.fixture(autouse=True)
def _patch_wraps(
@pytest.fixture
def context() -> RequestContext:
return RequestContext("request-1", "trace-1", "", "", "127.0.0.1")
@dataclass(frozen=True, slots=True)
class ApplicationServicesStub:
web_authentication: object
def bind_service(monkeypatch: pytest.MonkeyPatch, service: object) -> None:
monkeypatch.setattr(login, "application_services", lambda: ApplicationServicesStub(web_authentication=service))
class PasswordLoginStub:
def __init__(self, error: Exception | None = None) -> None:
self.error = error
self.call: tuple[RequestContext, str, str] | None = None
def login_with_password(self, context: RequestContext, *, email: str, password: str) -> str:
self.call = (context, email, password)
if self.error is not None:
raise self.error
return "access-token"
def test_password_login_delegates_to_application_service(
monkeypatch: pytest.MonkeyPatch,
sqlite_engine: Engine,
sqlite_session: Session,
):
wraps_features = SimpleNamespace(enable_email_password_login=True)
console_dify = SimpleNamespace(DEPLOYMENT_EDITION=DeploymentEdition.ENTERPRISE)
web_dify = SimpleNamespace(DEPLOYMENT_EDITION=DeploymentEdition.ENTERPRISE)
sqlite_session.add(DifySetup(version="test"))
sqlite_session.commit()
console_wraps._is_setup_completed.reset_success()
session_registry = scoped_session(sessionmaker(bind=sqlite_engine, expire_on_commit=False))
monkeypatch.setattr(console_wraps.db, "session", session_registry)
with (
patch("controllers.console.wraps.dify_config", console_dify),
patch(
"controllers.console.wraps.SystemFeatureService.is_email_password_login_enabled",
return_value=wraps_features.enable_email_password_login,
),
patch("controllers.web.login.dify_config", web_dify),
):
yield
session_registry.remove()
console_wraps._is_setup_completed.reset_success()
context: RequestContext,
) -> None:
service = PasswordLoginStub()
bind_service(monkeypatch, service)
class TestEmailCodeLoginSendEmailApi:
@patch("controllers.web.login.WebAppAuthService.send_email_code_login_email")
@patch("controllers.web.login.WebAppAuthService.get_user_through_email")
def test_should_fetch_account_with_original_email(
self,
mock_get_user,
mock_send_email,
app: Flask,
):
account = Account(name="Test User", email="user@example.com")
mock_get_user.return_value = account
mock_send_email.return_value = "token-123"
with app.test_request_context(
"/web/email-code-login",
method="POST",
json={"email": "User@Example.com", "language": "en-US"},
):
response = EmailCodeLoginSendEmailApi().post()
assert response == {"result": "success", "data": "token-123"}
mock_get_user.assert_called_once_with("User@Example.com", ANY)
mock_send_email.assert_called_once_with(account=account, language="en-US")
class TestEmailCodeLoginApi:
@patch("controllers.web.login.AccountService.reset_login_error_rate_limit")
@patch("controllers.web.login.WebAppAuthService.login", return_value="new-access-token")
@patch("controllers.web.login.WebAppAuthService.get_user_through_email")
@patch("controllers.web.login.WebAppAuthService.revoke_email_code_login_token")
@patch("controllers.web.login.WebAppAuthService.get_email_code_login_data")
def test_should_normalize_email_before_validating(
self,
mock_get_token_data,
mock_revoke_token,
mock_get_user,
mock_login,
mock_reset_login_rate,
app: Flask,
):
mock_get_token_data.return_value = {"email": "User@Example.com", "code": "123456"}
mock_get_user.return_value = Account(name="Test User", email="user@example.com")
with app.test_request_context(
"/web/email-code-login/validity",
method="POST",
json={"email": "User@Example.com", "code": encode_code("123456"), "token": "token-123"},
):
response = EmailCodeLoginApi().post()
assert response == {"result": "success", "data": {"access_token": "new-access-token"}}
mock_get_user.assert_called_once_with("User@Example.com", ANY)
mock_revoke_token.assert_called_once_with("token-123")
mock_login.assert_called_once()
mock_reset_login_rate.assert_called_once_with("user@example.com")
class TestLoginApi:
@patch("controllers.web.login.WebAppAuthService.login", return_value="access-tok")
@patch("controllers.web.login.WebAppAuthService.authenticate")
def test_login_success(self, mock_auth: MagicMock, mock_login: MagicMock, app: Flask) -> None:
mock_auth.return_value = Account(name="Test User", email="user@example.com")
with app.test_request_context(
"/web/login",
method="POST",
json={"email": "user@example.com", "password": base64.b64encode(b"Valid1234").decode()},
):
response = LoginApi().post()
assert response["data"]["access_token"] == "access-tok"
mock_auth.assert_called_once()
@patch(
"controllers.web.login.WebAppAuthService.authenticate",
side_effect=services.errors.account.AccountLoginError(),
result = unwrap(LoginApi.post)(
LoginApi(),
LoginPayload(email="User@Example.com", password="Valid1234"),
context,
)
def test_login_banned_account(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
from controllers.console.error import AccountBannedError
with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
with app.test_request_context(
"/web/login",
method="POST",
json={"email": "user@example.com", "password": base64.b64encode(b"Valid1234").decode()},
):
with pytest.raises(AccountBannedError):
LoginApi().post()
assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.ACCOUNT_BANNED)
@patch(
"controllers.web.login.WebAppAuthService.authenticate",
side_effect=services.errors.account.AccountPasswordError(),
)
def test_login_wrong_password(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
from controllers.console.auth.error import AuthenticationFailedError
with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
with app.test_request_context(
"/web/login",
method="POST",
json={"email": "user@example.com", "password": base64.b64encode(b"Valid1234").decode()},
):
with pytest.raises(AuthenticationFailedError):
LoginApi().post()
assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.INVALID_CREDENTIALS)
@patch(
"controllers.web.login.WebAppAuthService.authenticate",
side_effect=services.errors.account.AccountNotFoundError(),
)
def test_login_account_not_found(self, mock_auth: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture) -> None:
from controllers.console.auth.error import AuthenticationFailedError
with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
with app.test_request_context(
"/web/login",
method="POST",
json={"email": "missing@example.com", "password": base64.b64encode(b"Valid1234").decode()},
):
with pytest.raises(AuthenticationFailedError):
LoginApi().post()
assert_login_failure_logged(caplog, "missing@example.com", LoginFailureReason.ACCOUNT_NOT_FOUND)
@patch("controllers.web.login.WebAppAuthService.get_email_code_login_data", return_value=None)
def test_email_code_login_logs_invalid_token(
self, mock_get_token_data: MagicMock, app: Flask, caplog: pytest.LogCaptureFixture
) -> None:
with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
with app.test_request_context(
"/web/email-code-login/validity",
method="POST",
json={"email": "user@example.com", "code": encode_code("123456"), "token": "token-123"},
):
with pytest.raises(InvalidTokenError):
EmailCodeLoginApi().post()
mock_get_token_data.assert_called_once_with("token-123")
assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.INVALID_EMAIL_CODE_TOKEN)
@patch("controllers.web.login.WebAppAuthService.revoke_email_code_login_token")
@patch(
"controllers.web.login.WebAppAuthService.get_user_through_email",
side_effect=Unauthorized("Account is banned."),
)
@patch(
"controllers.web.login.WebAppAuthService.get_email_code_login_data",
return_value={"email": "User@Example.com", "code": "123456"},
)
def test_email_code_login_logs_banned_account(
self,
mock_get_token_data: MagicMock,
mock_get_user: MagicMock,
mock_revoke_token: MagicMock,
app: Flask,
caplog: pytest.LogCaptureFixture,
) -> None:
from controllers.console.error import AccountBannedError
with caplog.at_level(logging.WARNING, logger="controllers.web.login"):
with app.test_request_context(
"/web/email-code-login/validity",
method="POST",
json={"email": "User@Example.com", "code": encode_code("123456"), "token": "token-123"},
):
with pytest.raises(AccountBannedError):
EmailCodeLoginApi().post()
mock_get_token_data.assert_called_once_with("token-123")
mock_revoke_token.assert_called_once_with("token-123")
assert_login_failure_logged(caplog, "user@example.com", LoginFailureReason.ACCOUNT_BANNED)
assert result == {"result": "success", "data": {"access_token": "access-token"}}
assert service.call == (context, "User@Example.com", "Valid1234")
class TestLoginStatusApi:
@patch("controllers.web.login.extract_webapp_access_token", return_value=None)
def test_no_app_code_returns_logged_in_false(self, mock_extract: MagicMock, app: Flask) -> None:
with app.test_request_context("/web/login/status"):
result = LoginStatusApi().get()
@pytest.mark.parametrize(
("service_error", "http_error"),
[
pytest.param(WebAccountBannedError(), AccountBannedError, id="banned"),
pytest.param(WebAuthenticationFailedError(), AuthenticationFailedError, id="credentials"),
],
)
def test_password_login_translates_service_errors(
monkeypatch: pytest.MonkeyPatch,
context: RequestContext,
service_error: Exception,
http_error: type[Exception],
) -> None:
bind_service(monkeypatch, PasswordLoginStub(service_error))
assert result["logged_in"] is False
assert result["app_logged_in"] is False
@patch("controllers.web.login.decode_jwt_token")
@patch("controllers.web.login.PassportService")
@patch(
"services.webapp_access_query_service.WebAppAccessQueryService.requires_permission_check", return_value=False
)
@patch("services.webapp_access_query_service.WebAppAccessQueryService.get_app_id_by_code", return_value="app-1")
@patch("controllers.web.login.extract_webapp_access_token", return_value="tok")
def test_public_app_user_logged_in(
self,
mock_extract: MagicMock,
mock_app_id: MagicMock,
mock_perm: MagicMock,
mock_passport: MagicMock,
mock_decode: MagicMock,
app: Flask,
) -> None:
mock_decode.return_value = (MagicMock(), MagicMock())
with app.test_request_context("/web/login/status?app_code=code1"):
result = LoginStatusApi().get()
assert result["logged_in"] is True
assert result["app_logged_in"] is True
@patch("controllers.web.login.decode_jwt_token", side_effect=Exception("bad"))
@patch("controllers.web.login.PassportService")
@patch("services.webapp_access_query_service.WebAppAccessQueryService.requires_permission_check", return_value=True)
@patch("services.webapp_access_query_service.WebAppAccessQueryService.get_app_id_by_code", return_value="app-1")
@patch("controllers.web.login.extract_webapp_access_token", return_value="tok")
def test_private_app_passport_fails(
self,
mock_extract: MagicMock,
mock_app_id: MagicMock,
mock_perm: MagicMock,
mock_passport_cls: MagicMock,
mock_decode: MagicMock,
app: Flask,
) -> None:
mock_passport_cls.return_value.verify.side_effect = Exception("bad")
with app.test_request_context("/web/login/status?app_code=code1"):
result = LoginStatusApi().get()
assert result["logged_in"] is False
assert result["app_logged_in"] is False
with pytest.raises(http_error):
unwrap(LoginApi.post)(
LoginApi(),
LoginPayload(email="user@example.com", password="Valid1234"),
context,
)
class TestLogoutApi:
@patch("controllers.web.login.clear_webapp_access_token_from_cookie")
def test_logout_success(self, mock_clear: MagicMock, app: Flask) -> None:
with app.test_request_context("/web/logout", method="POST"):
response = LogoutApi().post()
class LoginStatusStub:
def __init__(self) -> None:
self.call: dict[str, str | None] | None = None
assert response.get_json() == {"result": "success"}
mock_clear.assert_called_once()
def get_login_status(self, **kwargs: str | None) -> WebLoginStatus:
self.call = kwargs
return WebLoginStatus(logged_in=True, app_logged_in=False)
def test_unknown_web_app_code_returns_http_not_found(app: Flask) -> None:
from flask_restx import Api
def test_login_status_passes_transport_tokens_to_service(monkeypatch: pytest.MonkeyPatch, app: Flask) -> None:
service = LoginStatusStub()
bind_service(monkeypatch, service)
monkeypatch.setattr(login, "extract_webapp_access_token", lambda _request: "account-token")
monkeypatch.setattr(login, "extract_webapp_passport", lambda app_code, _request: f"passport:{app_code}")
with app.test_request_context("/login/status?app_code=site-code"):
result = unwrap(LoginStatusApi.get)(
LoginStatusApi(),
LoginStatusQuery(app_code="site-code", user_id="session-1"),
RequestContext("request-1", None, "", "", "127.0.0.1"),
)
assert result == {"logged_in": True, "app_logged_in": False}
assert service.call == {
"app_code": "site-code",
"user_id": "session-1",
"access_token": "account-token",
"app_session_token": "passport:site-code",
}
@pytest.mark.parametrize(
("service_error", "status_code", "error_code"),
[
pytest.param(WebAppAccessAppNotFoundError(), 404, "app_not_found", id="unknown-app"),
pytest.param(WebAppAccessUnavailableError(), 503, "web_app_access_unavailable", id="access-unavailable"),
],
)
def test_login_status_translates_access_failures_to_http_errors(
monkeypatch: pytest.MonkeyPatch,
app: Flask,
service_error: Exception,
status_code: int,
error_code: str,
) -> None:
class UnavailableLoginStatusService(LoginStatusStub):
@override
def get_login_status(self, **kwargs: str | None) -> WebLoginStatus:
raise service_error
bind_service(monkeypatch, UnavailableLoginStatusService())
monkeypatch.setattr(console_wraps, "_is_setup_completed", lambda: True)
api = Api(app)
api.add_resource(LoginStatusApi, "/web/login/status")
response = app.test_client().get("/web/login/status?app_code=does-not-exist")
assert response.status_code == 404
assert response.get_json()["code"] == "app_not_found"
assert response.status_code == status_code
assert response.get_json()["code"] == error_code
class EmailLoginStub:
def send_email_login_code(self, *, email: str, language: str | None) -> str:
assert (email, language) == ("User@Example.com", "zh-Hans")
return "email-token"
def login_with_email_code(
self,
context: RequestContext,
*,
email: str,
code: str,
token: str,
) -> str:
assert context.remote_ip == "127.0.0.1"
assert (email, code, token) == ("User@Example.com", "123456", "email-token")
return "access-token"
def test_email_login_endpoints_delegate_to_application_service(
monkeypatch: pytest.MonkeyPatch,
context: RequestContext,
) -> None:
bind_service(monkeypatch, EmailLoginStub())
send_result = unwrap(EmailCodeLoginSendEmailApi.post)(
EmailCodeLoginSendEmailApi(),
EmailCodeLoginSendPayload(email="User@Example.com", language="zh-Hans"),
context,
)
login_result = unwrap(EmailCodeLoginApi.post)(
EmailCodeLoginApi(),
EmailCodeLoginVerifyPayload(email="User@Example.com", code="123456", token="email-token"),
context,
)
assert send_result == {"result": "success", "data": "email-token"}
assert login_result == {"result": "success", "data": {"access_token": "access-token"}}
def test_email_login_translates_invalid_code(monkeypatch: pytest.MonkeyPatch, context: RequestContext) -> None:
class InvalidCodeService(EmailLoginStub):
@override
def login_with_email_code(
self,
context: RequestContext,
*,
email: str,
code: str,
token: str,
) -> str:
del context, email, code, token
raise WebInvalidCodeError
bind_service(monkeypatch, InvalidCodeService())
with pytest.raises(EmailCodeError):
unwrap(EmailCodeLoginApi.post)(
EmailCodeLoginApi(),
EmailCodeLoginVerifyPayload(email="user@example.com", code="bad", token="email-token"),
context,
)
def test_logout_only_serializes_response_and_clears_cookie(
monkeypatch: pytest.MonkeyPatch,
app: Flask,
context: RequestContext,
) -> None:
cleared: list[tuple[str | None, str]] = []
def clear_cookie(response: Response, *, samesite: str) -> None:
cleared.append((response.get_json()["result"], samesite))
monkeypatch.setattr(login, "clear_webapp_access_token_from_cookie", clear_cookie)
with app.test_request_context("/logout", method="POST"):
response = unwrap(LogoutApi.post)(LogoutApi(), context)
assert response.get_json() == {"result": "success"}
assert cleared == [("success", "None")]