Чем вход может закончиться#

Успешный вход проходит через второй фактор. Остальные ответы должны иметь собственные имена: тогда приложение ловит ожидаемый отказ по типу, а не разбирает текст или адрес страницы.

Случай

Что получает вызывающий

Неверный пароль

WrongPassword или WrongPasswordError

Аккаунт не найден

AccountNotFound или AccountNotFoundError

Аккаунт заблокирован

AccountBlocked или AccountBlockedError

Нужен второй фактор

OtpStep

Неверный код

WrongCode или WrongCodeError

Нужна капча

CaptchaRequired или CaptchaRequiredError

→ Request · POST /login/password

{
  "login_id": "login:ada@mail.example",
  "password": "wrong"
}

← Response · 200 OK

{
  "status": "wrong_password",
  "body": {
    "code": "wrong_password",
    "message": "Password is incorrect"
  }
}

В JSON имя варианта повторяется во внешнем status и во внутреннем body.code. Const проверяет оба дискриминатора до создания исключения. Капча приходит с HTTP-статусом 403, остальные ожидаемые ответы этого сайта остаются внутри 200.

# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True)
class WrongPasswordDetails:
    code: Annotated[str, Const("wrong_password")]
    message: str


@dataclass(frozen=True, slots=True)
class AccountNotFoundDetails:
    code: Annotated[str, Const("account_not_found")]
    message: str


@dataclass(frozen=True, slots=True)
class AccountBlockedDetails:
    code: Annotated[str, Const("account_blocked")]
    message: str


@dataclass(frozen=True, slots=True)
class WrongCodeDetails:
    code: Annotated[str, Const("wrong_code")]
    message: str


@dataclass(frozen=True, slots=True)
class WrongPasswordResponse:
    status: Annotated[str, Const("wrong_password")]
    body: Payload[WrongPasswordDetails]


@dataclass(frozen=True, slots=True)
class AccountNotFoundResponse:
    status: Annotated[str, Const("account_not_found")]
    body: Payload[AccountNotFoundDetails]


@dataclass(frozen=True, slots=True)
class AccountBlockedResponse:
    status: Annotated[str, Const("account_blocked")]
    body: Payload[AccountBlockedDetails]


@dataclass(frozen=True, slots=True)
class WrongCodeResponse:
    status: Annotated[str, Const("wrong_code")]
    body: Payload[WrongCodeDetails]


@dataclass(frozen=True, slots=True)
class CaptchaRequiredResponse:
    kind: Annotated[str, Const("captcha")]
    site_key: str


class WrongPassword(ApiError[WrongPasswordResponse]):
    pass


class AccountNotFound(ApiError[AccountNotFoundResponse]):
    pass


class AccountBlocked(ApiError[AccountBlockedResponse]):
    pass


class WrongCode(ApiError[WrongCodeResponse]):
    pass


class CaptchaRequired(ApiError[CaptchaRequiredResponse]):
    pass


class LoginService:
    """Failures shared by all three operations of the login service."""

    base_url = BASE_URL
    errors = (
        Error(200, Json(WrongPasswordResponse), exception=WrongPassword),
        Error(200, Json(AccountNotFoundResponse), exception=AccountNotFound),
        Error(200, Json(AccountBlockedResponse), exception=AccountBlocked),
        Error(200, Json(WrongCodeResponse), exception=WrongCode),
        Error(403, Json(CaptchaRequiredResponse), exception=CaptchaRequired),
    )

Ответ сайта

Правило

Результат

status = wrong_password

Error(200, Json(WrongPasswordResponse), ...)

WrongPassword

status = account_not_found

Error(200, Json(AccountNotFoundResponse), ...)

AccountNotFound

status = account_blocked

Error(200, Json(AccountBlockedResponse), ...)

AccountBlocked

status = ok, модель OtpStep

успешный ответ SubmitPassword

OtpStep

status = wrong_code

Error(200, Json(WrongCodeResponse), ...)

WrongCode

HTTP 403, kind = captcha

Error(403, Json(CaptchaRequiredResponse), ...)

CaptchaRequired
Файл примера целиком
"""Probe a three-step login declaration and its six documented outcomes."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated

from eazy_sdk import Http, HttpOperation, JsonField, SyncApi, op
from eazy_sdk.response import ApiError, Const, Error, Json, Payload

BASE_URL = "https://mail.example"
WRONG_INPUT = "wrong"
VALID_INPUT = "correct"


@dataclass(frozen=True, slots=True)
class PasswordStep:
    login_id: str


@dataclass(frozen=True, slots=True)
class OtpStep:
    login_id: str
    destination: str


@dataclass(frozen=True, slots=True)
class Session:
    access_token: str


@dataclass(frozen=True, slots=True)
class PasswordStepResponse:
    status: Annotated[str, Const("ok")]
    body: Payload[PasswordStep]


@dataclass(frozen=True, slots=True)
class OtpStepResponse:
    status: Annotated[str, Const("ok")]
    body: Payload[OtpStep]


@dataclass(frozen=True, slots=True)
class SessionResponse:
    status: Annotated[str, Const("ok")]
    body: Payload[Session]


# region docs: http-login-failures
# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True)
class WrongPasswordDetails:
    code: Annotated[str, Const("wrong_password")]
    message: str


@dataclass(frozen=True, slots=True)
class AccountNotFoundDetails:
    code: Annotated[str, Const("account_not_found")]
    message: str


@dataclass(frozen=True, slots=True)
class AccountBlockedDetails:
    code: Annotated[str, Const("account_blocked")]
    message: str


@dataclass(frozen=True, slots=True)
class WrongCodeDetails:
    code: Annotated[str, Const("wrong_code")]
    message: str


@dataclass(frozen=True, slots=True)
class WrongPasswordResponse:
    status: Annotated[str, Const("wrong_password")]
    body: Payload[WrongPasswordDetails]


@dataclass(frozen=True, slots=True)
class AccountNotFoundResponse:
    status: Annotated[str, Const("account_not_found")]
    body: Payload[AccountNotFoundDetails]


@dataclass(frozen=True, slots=True)
class AccountBlockedResponse:
    status: Annotated[str, Const("account_blocked")]
    body: Payload[AccountBlockedDetails]


@dataclass(frozen=True, slots=True)
class WrongCodeResponse:
    status: Annotated[str, Const("wrong_code")]
    body: Payload[WrongCodeDetails]


@dataclass(frozen=True, slots=True)
class CaptchaRequiredResponse:
    kind: Annotated[str, Const("captcha")]
    site_key: str


class WrongPassword(ApiError[WrongPasswordResponse]):
    pass


class AccountNotFound(ApiError[AccountNotFoundResponse]):
    pass


class AccountBlocked(ApiError[AccountBlockedResponse]):
    pass


class WrongCode(ApiError[WrongCodeResponse]):
    pass


class CaptchaRequired(ApiError[CaptchaRequiredResponse]):
    pass


class LoginService:
    """Failures shared by all three operations of the login service."""

    base_url = BASE_URL
    errors = (
        Error(200, Json(WrongPasswordResponse), exception=WrongPassword),
        Error(200, Json(AccountNotFoundResponse), exception=AccountNotFound),
        Error(200, Json(AccountBlockedResponse), exception=AccountBlocked),
        Error(200, Json(WrongCodeResponse), exception=WrongCode),
        Error(403, Json(CaptchaRequiredResponse), exception=CaptchaRequired),
    )
# endregion docs: http-login-failures


# region docs: http-login-operations
# examples/mail/http/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(HttpOperation[PasswordStep]):
    __http__ = Http.post("/login/identify", success={200: Json(PasswordStepResponse)})

    email: JsonField[str]


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(HttpOperation[OtpStep]):
    __http__ = Http.post("/login/password", success={200: Json(OtpStepResponse)})

    login_id: JsonField[str]
    password: JsonField[str]


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(HttpOperation[Session]):
    __http__ = Http.post("/login/otp", success={200: Json(SessionResponse)})

    login_id: JsonField[str]
    code: JsonField[str]
# endregion docs: http-login-operations


# region docs: http-login-router
# examples/mail/http/login_probe.py
class LoginApi(LoginService, SyncApi):
    identify = op(Identify)
    password = op(SubmitPassword)
    otp = op(SubmitOtp)
# endregion docs: http-login-router

"""Run all six documented outcomes of the HTTP login."""

from __future__ import annotations

from collections.abc import Callable

from eazy_sdk.response import ApiError

from examples.mail.http._client import mail_client
from examples.mail.http.login_probe import VALID_INPUT, WRONG_INPUT, LoginApi


def _print_error(label: str, call: Callable[[], object]) -> None:
    try:
        call()
    except ApiError as error:
        print(f"{label}: {type(error).__name__}")


def main() -> None:
    with mail_client() as client:
        login = LoginApi(client)

        password_step = login.identify(email="ada@mail.example")
        _print_error(
            "wrong password",
            lambda: login.password(login_id=password_step.login_id, password=WRONG_INPUT),
        )
        _print_error("account not found", lambda: login.identify(email="missing@mail.example"))
        _print_error("account blocked", lambda: login.identify(email="blocked@mail.example"))

        otp_step = login.password(login_id=password_step.login_id, password=VALID_INPUT)
        print(f"second factor: {type(otp_step).__name__}")
        _print_error(
            "wrong code",
            lambda: login.otp(login_id=otp_step.login_id, code="000000"),
        )
        _print_error(
            "captcha",
            lambda: login.password(
                login_id=login.identify(email="captcha@mail.example").login_id,
                password=VALID_INPUT,
            ),
        )


if __name__ == "__main__":
    main()

Запуск: uv run python -m examples.mail.http.login_failures.

wrong password: WrongPassword
account not found: AccountNotFound
account blocked: AccountBlocked
second factor: OtpStep
wrong code: WrongCode
captcha: CaptchaRequired

Страница сообщает об отказах атрибутами data-code. Правила аккаунта принадлежат роутеру, а неверный пароль, капча и неверный код остаются у операций своих шагов.

При неверном пароле браузер видит этот фрагмент фактического ответа сайта:

<main data-page="failure" data-code="wrong_password">
  <p role="alert">Password is incorrect</p>
</main>

# examples/mail/browser/login_probe.py
ACCOUNT_FAILURES = (
    Failure(
        when=visible(css('[data-code="account_not_found"]')),
        exception=AccountNotFoundError,
    ),
    Failure(
        when=visible(css('[data-code="account_blocked"]')),
        exception=AccountBlockedError,
    ),
)

PASSWORD_FAILURES = (
    Failure(
        when=visible(css('[data-code="wrong_password"]')),
        exception=WrongPasswordError,
    ),
    Failure(
        when=visible(css('[data-page="captcha"]')),
        exception=CaptchaRequiredError,
    ),
)

OTP_FAILURES = (
    Failure(
        when=visible(css('[data-code="wrong_code"]')),
        exception=WrongCodeError,
    ),
)

Признак страницы

Где объявлен

Результат

data-code="wrong_password"

шаг пароля

WrongPasswordError

data-code="account_not_found"

LoginPortal.errors

AccountNotFoundError

data-code="account_blocked"

LoginPortal.errors

AccountBlockedError

поле code

исход шага пароля

OtpStep

data-code="wrong_code"

шаг кода

WrongCodeError

data-page="captcha"

шаг пароля

CaptchaRequiredError
Файл примера целиком
"""Type probe for a three-step browser login expressed with ``to=`` states."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Annotated, NoReturn

from eazy_sdk_browser import (
    AsyncBrowserApi,
    Browser,
    BrowserOperation,
    Element,
    Failure,
    PageError,
    css,
    outcomes,
    visible,
    when,
)

from eazy_sdk import op

VALID_LOGIN_INPUT = "correct"


class WrongPasswordError(PageError):
    """Пароль не подошёл."""


class AccountNotFoundError(PageError):
    """Аккаунт не найден."""


class AccountBlockedError(PageError):
    """Аккаунт заблокирован."""


class WrongCodeError(PageError):
    """Код подтверждения не подошёл."""


class CaptchaRequiredError(PageError):
    """Страница запросила капчу."""


# region docs: browser-login-failures
# examples/mail/browser/login_probe.py
ACCOUNT_FAILURES = (
    Failure(
        when=visible(css('[data-code="account_not_found"]')),
        exception=AccountNotFoundError,
    ),
    Failure(
        when=visible(css('[data-code="account_blocked"]')),
        exception=AccountBlockedError,
    ),
)

PASSWORD_FAILURES = (
    Failure(
        when=visible(css('[data-code="wrong_password"]')),
        exception=WrongPasswordError,
    ),
    Failure(
        when=visible(css('[data-page="captcha"]')),
        exception=CaptchaRequiredError,
    ),
)

OTP_FAILURES = (
    Failure(
        when=visible(css('[data-code="wrong_code"]')),
        exception=WrongCodeError,
    ),
)
# endregion docs: browser-login-failures


class IdentifyPage:
    email: Annotated[Element, css('input[name="email"]')]
    submit: Annotated[Element, css('button[type="submit"]')]


class PasswordPage:
    password: Annotated[Element, css('input[name="password"]')]
    submit: Annotated[Element, css('button[type="submit"]')]


class OtpPage:
    code: Annotated[Element, css('input[name="code"]')]
    submit: Annotated[Element, css('button[type="submit"]')]


@dataclass(frozen=True, slots=True)
class Mailbox:
    api: LoginPortal


@dataclass(frozen=True, slots=True)
class PasswordStep:
    api: LoginPortal

    async def submit(self, password: str) -> OtpStep:
        return await self.api.password(password=password)


@dataclass(frozen=True, slots=True)
class OtpStep:
    api: LoginPortal

    async def submit(self, code: str) -> Mailbox:
        return await self.api.otp(code=code)


async def _identify_stalled(content: IdentifyPage) -> NoReturn:
    _ = content
    raise TimeoutError("password step did not appear")


async def _password_stalled(content: PasswordPage) -> NoReturn:
    _ = content
    raise TimeoutError("OTP step did not appear")


async def _otp_stalled(content: OtpPage) -> NoReturn:
    _ = content
    raise TimeoutError("mailbox did not appear")


# region docs: browser-login-operations
# examples/mail/browser/login_probe.py
@dataclass(frozen=True, slots=True, kw_only=True)
class Identify(BrowserOperation[IdentifyPage, PasswordStep]):
    __browser__ = Browser.act(
        IdentifyPage,
        at=css('input[name="email"]'),
        outcomes=outcomes(
            when(visible(css('input[name="password"]')), to=PasswordStep),
            otherwise=_identify_stalled,
        ),
    )

    email: str

    async def act(self, content: IdentifyPage) -> None:
        await content.email.fill(self.email)
        await content.submit.click()


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitPassword(BrowserOperation[PasswordPage, OtpStep]):
    __browser__ = Browser.act(
        PasswordPage,
        at=css('input[name="password"]'),
        errors=PASSWORD_FAILURES,
        outcomes=outcomes(
            when(visible(css('input[name="code"]')), to=OtpStep),
            otherwise=_password_stalled,
        ),
    )

    password: str

    async def act(self, content: PasswordPage) -> None:
        await content.password.fill(self.password)
        await content.submit.click()


@dataclass(frozen=True, slots=True, kw_only=True)
class SubmitOtp(BrowserOperation[OtpPage, Mailbox]):
    __browser__ = Browser.act(
        OtpPage,
        at=css('input[name="code"]'),
        errors=OTP_FAILURES,
        outcomes=outcomes(
            when(visible(css('[data-page="mailbox"]')), to=Mailbox),
            otherwise=_otp_stalled,
        ),
    )

    code: str

    async def act(self, content: OtpPage) -> None:
        await content.code.fill(self.code)
        await content.submit.click()
# endregion docs: browser-login-operations


# region docs: browser-login-router
# examples/mail/browser/login_probe.py
class LoginPortal(AsyncBrowserApi):
    errors = ACCOUNT_FAILURES

    identify = op(Identify)
    password = op(SubmitPassword)
    otp = op(SubmitOtp)
# endregion docs: browser-login-router


# region docs: browser-login-flow
# examples/mail/browser/login_probe.py
async def typed_login(portal: LoginPortal) -> Mailbox:
    """Make loss of a step type fail the ordinary mypy gate."""

    password: PasswordStep = await portal.identify(email="ada@mail.example")
    otp: OtpStep = await password.submit(password=VALID_LOGIN_INPUT)
    mailbox: Mailbox = await otp.submit(code="123456")
    return mailbox
# endregion docs: browser-login-flow

"""Execute all six browser-login outcomes against the teaching mail site."""

from __future__ import annotations

import asyncio
from collections.abc import Awaitable

from eazy_sdk_browser import BrowserCallOptions, PageError

from examples.mail.browser.login_probe import (
    VALID_LOGIN_INPUT,
    LoginPortal,
    OtpStep,
    PasswordStep,
)
from examples.mail.site._playwright import MailBrowser, playwright_mail

OPTIONS = BrowserCallOptions(timeout=5.0)
REJECTED_INPUT = "wrong"


async def _portal(runtime: MailBrowser) -> LoginPortal:
    await runtime.reset()
    return LoginPortal(runtime.client)


async def _password(portal: LoginPortal, email: str = "ada@mail.example") -> PasswordStep:
    return await portal.identify(email=email, options=OPTIONS)


async def _otp(portal: LoginPortal, email: str = "ada@mail.example") -> OtpStep:
    password = await _password(portal, email)
    return await password.api.password(password=VALID_LOGIN_INPUT, options=OPTIONS)


async def _report(label: str, result: Awaitable[object]) -> None:
    try:
        value = await result
    except PageError as error:
        print(f"{label}: {type(error).__name__}")
    else:
        print(f"{label}: {type(value).__name__}")


async def run() -> None:
    async with playwright_mail() as runtime:
        portal = await _portal(runtime)
        password = await _password(portal)
        await _report(
            "wrong password",
            password.api.password(password=REJECTED_INPUT, options=OPTIONS),
        )
        await _report(
            "account not found",
            _password(await _portal(runtime), "missing@mail.example"),
        )
        await _report(
            "account blocked",
            _password(await _portal(runtime), "blocked@mail.example"),
        )
        await _report("second factor", _otp(await _portal(runtime)))

        otp = await _otp(await _portal(runtime))
        await _report("wrong code", otp.api.otp(code="000000", options=OPTIONS))

        captcha_password = await _password(
            await _portal(runtime),
            "captcha@mail.example",
        )
        await _report(
            "captcha",
            captcha_password.api.password(password=VALID_LOGIN_INPUT, options=OPTIONS),
        )


def main() -> None:
    asyncio.run(run())


if __name__ == "__main__":
    main()

Команда uv run python -m examples.mail.browser.login_failures запускает Chromium через PlaywrightDriver. Перехват запросов передаёт формы тому же MailSite, а операция читает DOM-признаки полученной страницы.

wrong password: WrongPasswordError
account not found: AccountNotFoundError
account blocked: AccountBlockedError
second factor: OtpStep
wrong code: WrongCodeError
captcha: CaptchaRequiredError

Примечание

OtpStep не является ошибкой. Это ожидаемое продолжение входа, поэтому оно объявлено исходом операции, а не исключением.

Дальше#