Add missing type hints to websocket_api (#50915)

This commit is contained in:
Ruslan Sayfutdinov 2021-05-21 17:39:18 +01:00 committed by GitHub
parent dc65f279a7
commit 42ff687c32
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 251 additions and 159 deletions

View file

@ -1,22 +1,31 @@
"""Handle the auth of a connection."""
from __future__ import annotations
from collections.abc import Callable
from typing import TYPE_CHECKING, Any, Final
from aiohttp.web import Request
import voluptuous as vol
from voluptuous.humanize import humanize_error
from homeassistant.auth.models import RefreshToken, User
from homeassistant.components.http.ban import process_success_login, process_wrong_login
from homeassistant.const import __version__
from homeassistant.core import HomeAssistant
from .connection import ActiveConnection
from .error import Disconnect
# mypy: allow-untyped-calls, allow-untyped-defs
if TYPE_CHECKING:
from .http import WebSocketAdapter
TYPE_AUTH = "auth"
TYPE_AUTH_INVALID = "auth_invalid"
TYPE_AUTH_OK = "auth_ok"
TYPE_AUTH_REQUIRED = "auth_required"
AUTH_MESSAGE_SCHEMA = vol.Schema(
TYPE_AUTH: Final = "auth"
TYPE_AUTH_INVALID: Final = "auth_invalid"
TYPE_AUTH_OK: Final = "auth_ok"
TYPE_AUTH_REQUIRED: Final = "auth_required"
AUTH_MESSAGE_SCHEMA: Final = vol.Schema(
{
vol.Required("type"): TYPE_AUTH,
vol.Exclusive("api_password", "auth"): str,
@ -25,17 +34,17 @@ AUTH_MESSAGE_SCHEMA = vol.Schema(
)
def auth_ok_message():
def auth_ok_message() -> dict[str, str]:
"""Return an auth_ok message."""
return {"type": TYPE_AUTH_OK, "ha_version": __version__}
def auth_required_message():
def auth_required_message() -> dict[str, str]:
"""Return an auth_required message."""
return {"type": TYPE_AUTH_REQUIRED, "ha_version": __version__}
def auth_invalid_message(message):
def auth_invalid_message(message: str) -> dict[str, str]:
"""Return an auth_invalid message."""
return {"type": TYPE_AUTH_INVALID, "message": message}
@ -43,16 +52,20 @@ def auth_invalid_message(message):
class AuthPhase:
"""Connection that requires client to authenticate first."""
def __init__(self, logger, hass, send_message, request):
def __init__(
self,
logger: WebSocketAdapter,
hass: HomeAssistant,
send_message: Callable[[str | dict[str, Any]], None],
request: Request,
) -> None:
"""Initialize the authentiated connection."""
self._hass = hass
self._send_message = send_message
self._logger = logger
self._request = request
self._authenticated = False
self._connection = None
async def async_handle(self, msg):
async def async_handle(self, msg: dict[str, str]) -> ActiveConnection:
"""Handle authentication."""
try:
msg = AUTH_MESSAGE_SCHEMA(msg)