diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 814facbb..9b5ca33a 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -74,3 +74,11 @@ repos:
rev: 5.7.0
hooks:
- id: isort
+ - repo: local
+ hooks:
+ - id: pytest-check
+ name: pytest-check
+ entry: pytest
+ language: system
+ pass_filenames: false
+ always_run: true
diff --git a/.prospector.yml b/.prospector.yml
index 1cde132c..32fcf97a 100644
--- a/.prospector.yml
+++ b/.prospector.yml
@@ -10,7 +10,7 @@ ignore-paths:
- docs
- tests
autodetect: true
-max-line-length: 88
+max-line-length: 108
bandit:
run: false
@@ -37,7 +37,7 @@ pep8:
enable:
- W601
options:
- max-line-length: 79
+ max-line-length: 108
pep257:
disable:
diff --git a/README.md b/README.md
index e164e5e1..bb5950e7 100644
--- a/README.md
+++ b/README.md
@@ -14,12 +14,13 @@
[![Discord][discord-shield]][discord]
[![Community Forum][forum-shield]][forum]
-A fork of the [official Tesla integration](https://www.home-assistant.io/integrations/tesla/) in Home Assistant to use an oauth proxy for logins.
+A fork of the [official Tesla integration](https://www.home-assistant.io/integrations/tesla/) in Home Assistant.
-This fork uses an oauth proxy instead of screen scraping which was [rejected by HA](https://github.com/home-assistant/core/pull/46558#issuecomment-822858608). The oauth proxy sits as a middleman between Home Assistant and Tesla to intercept login credentials such as your account and password. Due to the way the HTTP server works in Home Assistant, the auth endpoint cannot be turned off although we protect access by requiring knowledge of a ongoing config flow id. However, for maximum security, restart Home Assistant to completely disable the proxy server.
-
-To the extent the official component adds features unrelated to the login, we will attempt to keep up to date. Users are welcome to port any fixes in this custom integration into HA. Please note that this component will not have the same quality or support as the official component. Do not report issues to Home Assistant.
+This is the successor to the core app which was removed due to Tesla login issues. Do not report issues to Home Assistant.
+To use the component, you will need an application to generate a Tesla refresh token:
+- [Tesla Tokens](https://play.google.com/store/apps/details?id=net.leveugle.teslatokens)
+- [Auth App for Tesla](https://apps.apple.com/us/app/auth-app-for-tesla/id1552058613)
## Installation
1. Use HACS after adding this `https://github.com/alandtse/tesla` as a custom repository. Skip to 7.
@@ -29,7 +30,7 @@ To the extent the official component adds features unrelated to the login, we wi
5. Download _all_ the files from the `custom_components/tesla_custom/` directory (folder) in this repository.
6. Place the files you downloaded in the new directory (folder) you created.
7. Restart Home Assistant.
-8. In the HA UI go to "Configuration" -> "Integrations" click "+" and search for "Tesla Custom Integration". If you are replacing core, remove the core integration before installing.
+8. In the HA UI go to "Configuration" -> "Integrations" click "+" and search for "Tesla Custom Integration".
diff --git a/custom_components/tesla_custom/__init__.py b/custom_components/tesla_custom/__init__.py
index 8bc602b5..740bdf47 100644
--- a/custom_components/tesla_custom/__init__.py
+++ b/custom_components/tesla_custom/__init__.py
@@ -5,19 +5,24 @@
import logging
import async_timeout
-from homeassistant.config_entries import SOURCE_IMPORT, ConfigEntry
+from homeassistant.config_entries import SOURCE_IMPORT
from homeassistant.const import (
CONF_ACCESS_TOKEN,
- CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_TOKEN,
CONF_USERNAME,
+ EVENT_HOMEASSISTANT_CLOSE,
HTTP_UNAUTHORIZED,
)
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import callback
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.httpx_client import SERVER_SOFTWARE, USER_AGENT
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+import httpx
from teslajsonpy import Controller as TeslaAPI
from teslajsonpy.exceptions import IncompleteCredentials, TeslaException
+import voluptuous as vol
from .config_flow import CannotConnect, InvalidAuth, validate_input
from .const import (
@@ -33,6 +38,21 @@
_LOGGER = logging.getLogger(__name__)
+CONFIG_SCHEMA = vol.Schema(
+ {
+ DOMAIN: vol.Schema(
+ {
+ vol.Required(CONF_USERNAME): cv.string,
+ vol.Required(CONF_TOKEN): cv.string,
+ vol.Optional(
+ CONF_SCAN_INTERVAL, default=DEFAULT_SCAN_INTERVAL
+ ): vol.All(cv.positive_int, vol.Clamp(min=MIN_SCAN_INTERVAL)),
+ }
+ )
+ },
+ extra=vol.ALLOW_EXTRA,
+)
+
@callback
def _async_save_tokens(hass, config_entry, access_token, refresh_token, expiration):
@@ -50,7 +70,11 @@ def _async_save_tokens(hass, config_entry, access_token, refresh_token, expirati
@callback
def _async_configured_emails(hass):
"""Return a set of configured Tesla emails."""
- return {entry.title for entry in hass.config_entries.async_entries(DOMAIN)}
+ return {
+ entry.data[CONF_USERNAME]
+ for entry in hass.config_entries.async_entries(DOMAIN)
+ if CONF_USERNAME in entry.data
+ }
async def async_setup(hass, base_config):
@@ -71,7 +95,7 @@ def _update_entry(email, data=None, options=None):
if not config:
return True
email = config[CONF_USERNAME]
- password = config[CONF_PASSWORD]
+ token = config[CONF_TOKEN]
scan_interval = config[CONF_SCAN_INTERVAL]
if email in _async_configured_emails(hass):
try:
@@ -81,6 +105,7 @@ def _update_entry(email, data=None, options=None):
_update_entry(
email,
data={
+ CONF_USERNAME: email,
CONF_ACCESS_TOKEN: info[CONF_ACCESS_TOKEN],
CONF_TOKEN: info[CONF_TOKEN],
CONF_EXPIRATION: info[CONF_EXPIRATION],
@@ -92,7 +117,7 @@ def _update_entry(email, data=None, options=None):
hass.config_entries.flow.async_init(
DOMAIN,
context={"source": SOURCE_IMPORT},
- data={CONF_USERNAME: email, CONF_PASSWORD: password},
+ data={CONF_USERNAME: email, CONF_TOKEN: token},
)
)
hass.data.setdefault(DOMAIN, {})
@@ -101,10 +126,12 @@ def _update_entry(email, data=None, options=None):
async def async_setup_entry(hass, config_entry):
- # pylint: disable=too-many-locals
"""Set up Tesla as config entry."""
+ # pylint: disable=too-many-locals
hass.data.setdefault(DOMAIN, {})
config = config_entry.data
+ # Because users can have multiple accounts, we always create a new session so they have separate cookies
+ async_client = httpx.AsyncClient(headers={USER_AGENT: SERVER_SOFTWARE}, timeout=60)
email = config_entry.title
if email in hass.data[DOMAIN] and CONF_SCAN_INTERVAL in hass.data[DOMAIN][email]:
scan_interval = hass.data[DOMAIN][email][CONF_SCAN_INTERVAL]
@@ -114,9 +141,8 @@ async def async_setup_entry(hass, config_entry):
hass.data[DOMAIN].pop(email)
try:
controller = TeslaAPI(
- websession=None,
+ async_client,
email=config.get(CONF_USERNAME),
- password=config.get(CONF_PASSWORD),
refresh_token=config[CONF_TOKEN],
access_token=config[CONF_ACCESS_TOKEN],
expiration=config.get(CONF_EXPIRATION, 0),
@@ -132,14 +158,40 @@ async def async_setup_entry(hass, config_entry):
refresh_token = result["refresh_token"]
access_token = result["access_token"]
expiration = result["expiration"]
- except IncompleteCredentials:
- _async_start_reauth(hass, config_entry)
- return False
+ except IncompleteCredentials as ex:
+ await async_client.aclose()
+ raise ConfigEntryAuthFailed from ex
+ except httpx.ConnectTimeout as ex:
+ await async_client.aclose()
+ raise ConfigEntryNotReady from ex
except TeslaException as ex:
+ await async_client.aclose()
if ex.code == HTTP_UNAUTHORIZED:
- _async_start_reauth(hass, config_entry)
+ raise ConfigEntryAuthFailed from ex
+ if ex.message in [
+ "VEHICLE_UNAVAILABLE",
+ "TOO_MANY_REQUESTS",
+ "SERVICE_MAINTENANCE",
+ "UPSTREAM_TIMEOUT",
+ ]:
+ raise ConfigEntryNotReady(
+ f"Temporarily unable to communicate with Tesla API: {ex.message}"
+ ) from ex
_LOGGER.error("Unable to communicate with Tesla API: %s", ex.message)
return False
+
+ async def _async_close_client(*_):
+ await async_client.aclose()
+
+ @callback
+ def _async_create_close_task():
+ asyncio.create_task(_async_close_client())
+
+ config_entry.async_on_unload(
+ hass.bus.async_listen_once(EVENT_HOMEASSISTANT_CLOSE, _async_close_client)
+ )
+ config_entry.async_on_unload(_async_create_close_task)
+
_async_save_tokens(hass, config_entry, access_token, refresh_token, expiration)
coordinator = TeslaDataUpdateCoordinator(
hass, config_entry=config_entry, controller=controller
@@ -162,23 +214,15 @@ async def async_setup_entry(hass, config_entry):
for device in all_devices:
entry_data["devices"][device.hass_type].append(device)
- for platform in PLATFORMS:
- _LOGGER.debug("Loading %s", platform)
- hass.async_create_task(
- hass.config_entries.async_forward_entry_setup(config_entry, platform)
- )
+ hass.config_entries.async_setup_platforms(config_entry, PLATFORMS)
+
return True
async def async_unload_entry(hass, config_entry) -> bool:
"""Unload a config entry."""
- unload_ok = all(
- await asyncio.gather(
- *[
- hass.config_entries.async_forward_entry_unload(config_entry, platform)
- for platform in PLATFORMS
- ]
- )
+ unload_ok = await hass.config_entries.async_unload_platforms(
+ config_entry, PLATFORMS
)
await hass.data[DOMAIN].get(config_entry.entry_id)[
"coordinator"
@@ -193,17 +237,6 @@ async def async_unload_entry(hass, config_entry) -> bool:
return False
-def _async_start_reauth(hass: HomeAssistant, entry: ConfigEntry):
- hass.async_create_task(
- hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": "reauth"},
- data=entry.data,
- )
- )
- _LOGGER.error("Credentials are no longer valid. Please reauthenticate")
-
-
async def update_listener(hass, config_entry):
"""Update when config_entry options update."""
controller = hass.data[DOMAIN][config_entry.entry_id]["coordinator"].controller
diff --git a/custom_components/tesla_custom/config_flow.py b/custom_components/tesla_custom/config_flow.py
index 50153d06..12d78a02 100644
--- a/custom_components/tesla_custom/config_flow.py
+++ b/custom_components/tesla_custom/config_flow.py
@@ -1,59 +1,43 @@
"""Tesla Config Flow."""
-import datetime
import logging
-from typing import Any, Dict, List, Optional
-from aiohttp import web, web_response
-from aiohttp.web_exceptions import HTTPBadRequest
from homeassistant import config_entries, core, exceptions
-from homeassistant.components.http.view import HomeAssistantView
from homeassistant.const import (
CONF_ACCESS_TOKEN,
- CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_TOKEN,
CONF_USERNAME,
HTTP_UNAUTHORIZED,
)
from homeassistant.core import callback
-from homeassistant.data_entry_flow import UnknownFlow
-from homeassistant.exceptions import Unauthorized
from homeassistant.helpers import config_validation as cv
-from homeassistant.helpers.network import NoURLAvailableError, get_url
-from teslajsonpy import Controller as TeslaAPI
-from teslajsonpy.exceptions import IncompleteCredentials, TeslaException
-from teslajsonpy.teslaproxy import TeslaProxy
+from homeassistant.helpers.httpx_client import SERVER_SOFTWARE, USER_AGENT
+import httpx
+from teslajsonpy import Controller as TeslaAPI, TeslaException
+from teslajsonpy.exceptions import IncompleteCredentials
import voluptuous as vol
-from yarl import URL
-from .const import ( # pylint: disable=unused-import
- AUTH_CALLBACK_NAME,
- AUTH_CALLBACK_PATH,
- AUTH_PROXY_NAME,
- AUTH_PROXY_PATH,
+from .const import (
CONF_EXPIRATION,
CONF_WAKE_ON_START,
DEFAULT_SCAN_INTERVAL,
DEFAULT_WAKE_ON_START,
- DOMAIN,
- ERROR_URL_NOT_DETECTED,
MIN_SCAN_INTERVAL,
)
+from .const import DOMAIN # pylint: disable=unused-import
_LOGGER = logging.getLogger(__name__)
-class TeslaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): # type: ignore
+class TeslaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a config flow for Tesla."""
VERSION = 1
- CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
- proxy: TeslaProxy = None
- proxy_view: Optional["TeslaAuthorizationProxyView"] = None
- data: Optional[Dict[str, Any]] = None
- warning_shown: bool = False
- callback_url: Optional[URL] = None
- controller: Optional[TeslaAPI] = None
+
+ def __init__(self) -> None:
+ """Initialize the tesla flow."""
+ self.username = None
+ self.reauth = False
async def async_step_import(self, import_config):
"""Import a config entry from configuration.yaml."""
@@ -61,104 +45,44 @@ async def async_step_import(self, import_config):
async def async_step_user(self, user_input=None):
"""Handle the start of the config flow."""
- if not self.warning_shown:
- self.warning_shown = True
- return self.async_show_form(
- step_id="user",
- data_schema=vol.Schema({}, extra=vol.ALLOW_EXTRA),
- errors={},
- description_placeholders={},
- )
- return await self.async_step_start_oauth()
-
- async def async_step_reauth(self, data):
- """Handle configuration by re-auth."""
- self.warning_shown = False
- return await self.async_step_user()
-
- async def async_step_start_oauth(self, user_input=None):
- """Start oauth step for login."""
- self.warning_shown = False
- self.controller = TeslaAPI(
- websession=None,
- update_interval=DEFAULT_SCAN_INTERVAL,
- )
- host_url: URL = self.controller.get_oauth_url()
- try:
- hass_proxy_url: URL = URL(
- get_url(self.hass, prefer_external=True)
- ).with_path(AUTH_PROXY_PATH)
-
- TeslaConfigFlow.proxy: TeslaProxy = TeslaProxy(
- proxy_url=hass_proxy_url,
- host_url=host_url,
- )
- TeslaConfigFlow.callback_url: URL = (
- URL(get_url(self.hass, prefer_external=True))
- .with_path(AUTH_CALLBACK_PATH)
- .with_query({"flow_id": self.flow_id})
- )
- except NoURLAvailableError:
- self.warning_shown = False
- return self.async_show_form(
- step_id="user",
- data_schema=vol.Schema({}, extra=vol.ALLOW_EXTRA),
- errors={"base": ERROR_URL_NOT_DETECTED},
- description_placeholders={},
- )
-
- proxy_url: URL = self.proxy.access_url().with_query(
- {"config_flow_id": self.flow_id, "callback_url": str(self.callback_url)}
- )
-
- if not self.proxy_view:
- TeslaConfigFlow.proxy_view = TeslaAuthorizationProxyView(
- self.proxy.all_handler
- )
- self.hass.http.register_view(TeslaAuthorizationCallbackView())
- self.hass.http.register_view(self.proxy_view)
- return self.async_external_step(step_id="check_proxy", url=str(proxy_url))
-
- async def async_step_check_proxy(self, user_input=None):
- """Check status of oauth response for login."""
- self.data = user_input
- self.proxy_view.reset()
- return self.async_external_step_done(next_step_id="finish_oauth")
-
- async def async_step_finish_oauth(self, user_input=None):
- """Finish auth."""
- info = {}
errors = {}
- self.controller.set_authorization_code(self.data.get("code", ""))
- self.controller.set_authorization_domain(self.data.get("domain", ""))
- try:
- info = await validate_input(self.hass, info, self.controller)
- except CannotConnect:
- errors["base"] = "cannot_connect"
- return self.async_abort(reason="cannot_connect")
- except InvalidAuth:
- errors["base"] = "invalid_auth"
- return self.async_abort(reason="invalid_auth")
- # convert from teslajsonpy to HA keys
- if info:
- info = {
- CONF_TOKEN: info["refresh_token"],
- CONF_ACCESS_TOKEN: info[CONF_ACCESS_TOKEN],
- CONF_EXPIRATION: info[CONF_EXPIRATION],
- }
- await self.proxy.reset_data()
- if info and not errors:
- existing_entry = self._async_entry_for_username(self.data[CONF_USERNAME])
- if existing_entry and existing_entry.data == info:
+
+ if user_input is not None:
+ existing_entry = self._async_entry_for_username(user_input[CONF_USERNAME])
+ if existing_entry and not self.reauth:
return self.async_abort(reason="already_configured")
- if existing_entry:
- self.hass.config_entries.async_update_entry(existing_entry, data=info)
- await self.hass.config_entries.async_reload(existing_entry.entry_id)
- return self.async_abort(reason="reauth_successful")
+ try:
+ info = await validate_input(self.hass, user_input)
+ except CannotConnect:
+ errors["base"] = "cannot_connect"
+ except InvalidAuth:
+ errors["base"] = "invalid_auth"
+
+ if not errors:
+ if existing_entry:
+ self.hass.config_entries.async_update_entry(
+ existing_entry, data=info
+ )
+ await self.hass.config_entries.async_reload(existing_entry.entry_id)
+ return self.async_abort(reason="reauth_successful")
+
+ return self.async_create_entry(
+ title=user_input[CONF_USERNAME], data=info
+ )
+
+ return self.async_show_form(
+ step_id="user",
+ data_schema=self._async_schema(),
+ errors=errors,
+ description_placeholders={},
+ )
- return self.async_create_entry(title=self.data[CONF_USERNAME], data=info)
- return self.async_abort(reason="login_failed")
+ async def async_step_reauth(self, data):
+ """Handle configuration by re-auth."""
+ self.username = data[CONF_USERNAME]
+ self.reauth = True
+ return await self.async_step_user()
@staticmethod
@callback
@@ -166,11 +90,21 @@ def async_get_options_flow(config_entry):
"""Get the options flow for this handler."""
return OptionsFlowHandler(config_entry)
+ @callback
+ def _async_schema(self):
+ """Fetch schema with defaults."""
+ return vol.Schema(
+ {
+ vol.Required(CONF_USERNAME, default=self.username): str,
+ vol.Required(CONF_TOKEN): str,
+ }
+ )
+
@callback
def _async_entry_for_username(self, username):
"""Find an existing entry for a username."""
for entry in self._async_current_entries():
- if entry.title == username:
+ if entry.data.get(CONF_USERNAME) == username:
return entry
return None
@@ -178,7 +112,7 @@ def _async_entry_for_username(self, username):
class OptionsFlowHandler(config_entries.OptionsFlow):
"""Handle a option flow for Tesla."""
- def __init__(self, config_entry: config_entries.ConfigEntry):
+ def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
"""Initialize options flow."""
self.config_entry = config_entry
@@ -206,35 +140,40 @@ async def async_step_init(self, user_input=None):
return self.async_show_form(step_id="init", data_schema=data_schema)
-async def validate_input(hass: core.HomeAssistant, data, controller: TeslaAPI = None):
+async def validate_input(hass: core.HomeAssistant, data):
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
"""
config = {}
+ async_client = httpx.AsyncClient(headers={USER_AGENT: SERVER_SOFTWARE}, timeout=60)
try:
- if not controller:
- controller = TeslaAPI(
- websession=None,
- email=data.get(CONF_USERNAME),
- password=data.get(CONF_PASSWORD),
- refresh_token=data.get(CONF_TOKEN),
- access_token=data.get(CONF_ACCESS_TOKEN),
- expiration=data.get(CONF_EXPIRATION, 0),
- update_interval=data.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL),
- )
- config = await controller.connect(
- wake_if_asleep=data.get(CONF_WAKE_ON_START, DEFAULT_WAKE_ON_START),
- test_login=True,
+ controller = TeslaAPI(
+ async_client,
+ email=data[CONF_USERNAME],
+ refresh_token=data[CONF_TOKEN],
+ update_interval=DEFAULT_SCAN_INTERVAL,
+ expiration=config.get(CONF_EXPIRATION, 0),
)
+ result = await controller.connect(test_login=True)
+ config[CONF_TOKEN] = result["refresh_token"]
+ config[CONF_ACCESS_TOKEN] = result["access_token"]
+ config[CONF_EXPIRATION] = result[CONF_EXPIRATION]
+ config[CONF_USERNAME] = data[CONF_USERNAME]
+
+ except IncompleteCredentials as ex:
+ _LOGGER.error("Authentication error: %s %s", ex.message, ex)
+ raise InvalidAuth() from ex
except TeslaException as ex:
if ex.code == HTTP_UNAUTHORIZED or isinstance(ex, IncompleteCredentials):
_LOGGER.error("Invalid credentials: %s", ex.message)
raise InvalidAuth() from ex
_LOGGER.error("Unable to communicate with Tesla API: %s", ex.message)
raise CannotConnect() from ex
+ finally:
+ await async_client.aclose()
_LOGGER.debug("Credentials successfully connected to the Tesla API")
return config
@@ -245,88 +184,3 @@ class CannotConnect(exceptions.HomeAssistantError):
class InvalidAuth(exceptions.HomeAssistantError):
"""Error to indicate there is invalid auth."""
-
-
-class TeslaAuthorizationCallbackView(HomeAssistantView):
- """Handle callback from external auth."""
-
- url = AUTH_CALLBACK_PATH
- name = AUTH_CALLBACK_NAME
- requires_auth = False
-
- async def get(self, request: web.Request):
- """Receive authorization confirmation."""
- hass = request.app["hass"]
- try:
- await hass.config_entries.flow.async_configure(
- flow_id=request.query["flow_id"],
- user_input=request.query,
- )
- except (KeyError, UnknownFlow) as ex:
- _LOGGER.debug("Callback flow_id is invalid")
- raise HTTPBadRequest() from ex
- return web_response.Response(
- headers={"content-type": "text/html"},
- text="Success! This window can be closed",
- )
-
-
-class TeslaAuthorizationProxyView(HomeAssistantView):
- """Handle proxy connections."""
-
- url: str = AUTH_PROXY_PATH
- extra_urls: List[str] = [f"{AUTH_PROXY_PATH}/{{tail:.*}}"]
- name: str = AUTH_PROXY_NAME
- requires_auth: bool = False
- handler: web.RequestHandler = None
- known_ips: Dict[str, datetime.datetime] = {}
- auth_seconds: int = 300
- cors_allowed = False
-
- def __init__(self, handler: web.RequestHandler):
- """Initialize routes for view.
-
- Args:
- handler (web.RequestHandler): Handler to apply to all method types
-
- """
- TeslaAuthorizationProxyView.handler = handler
- for method in ("get", "post", "delete", "put", "patch", "head", "options"):
- setattr(self, method, self.check_auth())
-
- @classmethod
- def check_auth(cls):
- """Wrap access control into the handler."""
-
- async def wrapped(request, **kwargs):
- """Wrap the handler to require knowledge of config_flow_id."""
- hass = request.app["hass"]
- success = False
- if (
- request.remote not in cls.known_ips
- or (datetime.datetime.now() - cls.known_ips.get(request.remote)).seconds
- > cls.auth_seconds
- ):
- try:
- flow_id = request.url.query["config_flow_id"]
- except KeyError as ex:
- raise Unauthorized() from ex
- for flow in hass.config_entries.flow.async_progress():
- if flow["flow_id"] == flow_id:
- _LOGGER.debug(
- "Found flow_id; adding %s to known_ips for %s seconds",
- request.remote,
- cls.auth_seconds,
- )
- success = True
- if not success:
- raise Unauthorized()
- cls.known_ips[request.remote] = datetime.datetime.now()
- return await cls.handler(request, **kwargs)
-
- return wrapped
-
- @classmethod
- def reset(cls) -> None:
- """Reset the view."""
- cls.known_ips = {}
diff --git a/custom_components/tesla_custom/const.py b/custom_components/tesla_custom/const.py
index 0883d0b7..fe23bc81 100644
--- a/custom_components/tesla_custom/const.py
+++ b/custom_components/tesla_custom/const.py
@@ -25,7 +25,7 @@
"parking brake sensor": "mdi:car-brake-parking",
"charger sensor": "mdi:ev-station",
"charger switch": "mdi:battery-charging",
- "update switch": "mdi:update",
+ "update switch": "mdi:car-connected",
"maxrange switch": "mdi:gauge-full",
"temperature sensor": "mdi:thermometer",
"location tracker": "mdi:crosshairs-gps",
diff --git a/custom_components/tesla_custom/manifest.json b/custom_components/tesla_custom/manifest.json
index e09afa07..6dd25d45 100644
--- a/custom_components/tesla_custom/manifest.json
+++ b/custom_components/tesla_custom/manifest.json
@@ -4,15 +4,9 @@
"config_flow": true,
"documentation": "https://github.com/alandtse/tesla",
"issue_tracker": "https://github.com/alandtse/tesla/issues",
- "requirements": [
- "teslajsonpy~=0.19.1"
- ],
- "codeowners": [
- "@alandtse"
- ],
- "dependencies": [
- "http"
- ],
+ "requirements": ["teslajsonpy~=0.20.0"],
+ "codeowners": ["@alandtse"],
+ "dependencies": ["http"],
"dhcp": [
{
"hostname": "tesla_*",
@@ -29,4 +23,4 @@
],
"iot_class": "cloud_polling",
"version": "0.2.1"
-}
\ No newline at end of file
+}
diff --git a/custom_components/tesla_custom/strings.json b/custom_components/tesla_custom/strings.json
index 71f47b44..0b760bfd 100644
--- a/custom_components/tesla_custom/strings.json
+++ b/custom_components/tesla_custom/strings.json
@@ -1,18 +1,23 @@
{
"config": {
- "error": {
- "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
- "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
- "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
- "url_not_detected": "Home Assistant External URL not detected. Please set in Configuration -> General"
- },
"abort": {
- "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
- "already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
+ "already_configured": "Account is already configured",
+ "reauth_successful": "Re-authentication was successful"
+ },
+ "error": {
+ "already_configured": "Account is already configured",
+ "cannot_connect": "Failed to connect",
+ "invalid_auth": "Invalid authentication"
},
"step": {
"user": {
- "description": "The next step will forward to the proxy for Tesla authentication. Tesla protects the login with a web application firewall (WAF) which may require a refresh to get or receive data. Please be cautious with multiple refreshes in a short period of time.",
+ "data": {
+ "mfa": "MFA Code (optional)",
+ "password": "Password",
+ "username": "Email",
+ "token": "Refresh Token"
+ },
+ "description": "Use 'Auth App for Tesla' on iOS or 'Tesla Tokens' on Android\r\n to create a refresh token and enter it below.",
"title": "Tesla - Configuration"
}
}
@@ -21,8 +26,8 @@
"step": {
"init": {
"data": {
- "scan_interval": "Seconds between scans",
- "enable_wake_on_start": "Force cars awake on startup"
+ "enable_wake_on_start": "Force cars awake on startup",
+ "scan_interval": "Seconds between polling"
}
}
}
diff --git a/custom_components/tesla_custom/switch.py b/custom_components/tesla_custom/switch.py
index 8ea865fb..bf1937d6 100644
--- a/custom_components/tesla_custom/switch.py
+++ b/custom_components/tesla_custom/switch.py
@@ -1,4 +1,5 @@
"""Support for Tesla charger switches."""
+from custom_components.tesla_custom.const import ICONS
import logging
from homeassistant.components.switch import SwitchEntity
@@ -75,7 +76,7 @@ def is_on(self):
class UpdateSwitch(TeslaDevice, SwitchEntity):
- """Representation of a Tesla update switch."""
+ """Representation of a Tesla update switch. Described in UI as polling."""
def __init__(self, tesla_device, coordinator):
"""Initialise the switch."""
@@ -85,7 +86,12 @@ def __init__(self, tesla_device, coordinator):
@property
def name(self):
"""Return the name of the device."""
- return super().name.replace("charger", "update")
+ return super().name.replace("charger", "polling")
+
+ @property
+ def icon(self):
+ """Return the icon of the sensor."""
+ return ICONS.get("update switch")
@property
def unique_id(self) -> str:
@@ -94,13 +100,13 @@ def unique_id(self) -> str:
async def async_turn_on(self, **kwargs):
"""Send the on command."""
- _LOGGER.debug("Enable updates: %s %s", self.name, self.tesla_device.id())
+ _LOGGER.debug("Enable polling: %s %s", self.name, self.tesla_device.id())
self.controller.set_updates(self.tesla_device.id(), True)
self.async_write_ha_state()
async def async_turn_off(self, **kwargs):
"""Send the off command."""
- _LOGGER.debug("Disable updates: %s %s", self.name, self.tesla_device.id())
+ _LOGGER.debug("Disable polling: %s %s", self.name, self.tesla_device.id())
self.controller.set_updates(self.tesla_device.id(), False)
self.async_write_ha_state()
diff --git a/custom_components/tesla_custom/translations/en.json b/custom_components/tesla_custom/translations/en.json
index 71f47b44..0b760bfd 100644
--- a/custom_components/tesla_custom/translations/en.json
+++ b/custom_components/tesla_custom/translations/en.json
@@ -1,18 +1,23 @@
{
"config": {
- "error": {
- "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
- "already_configured": "[%key:common::config_flow::abort::already_configured_account%]",
- "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
- "url_not_detected": "Home Assistant External URL not detected. Please set in Configuration -> General"
- },
"abort": {
- "reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]",
- "already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
+ "already_configured": "Account is already configured",
+ "reauth_successful": "Re-authentication was successful"
+ },
+ "error": {
+ "already_configured": "Account is already configured",
+ "cannot_connect": "Failed to connect",
+ "invalid_auth": "Invalid authentication"
},
"step": {
"user": {
- "description": "The next step will forward to the proxy for Tesla authentication. Tesla protects the login with a web application firewall (WAF) which may require a refresh to get or receive data. Please be cautious with multiple refreshes in a short period of time.",
+ "data": {
+ "mfa": "MFA Code (optional)",
+ "password": "Password",
+ "username": "Email",
+ "token": "Refresh Token"
+ },
+ "description": "Use 'Auth App for Tesla' on iOS or 'Tesla Tokens' on Android\r\n to create a refresh token and enter it below.",
"title": "Tesla - Configuration"
}
}
@@ -21,8 +26,8 @@
"step": {
"init": {
"data": {
- "scan_interval": "Seconds between scans",
- "enable_wake_on_start": "Force cars awake on startup"
+ "enable_wake_on_start": "Force cars awake on startup",
+ "scan_interval": "Seconds between polling"
}
}
}
diff --git a/custom_components/tesla_custom/translations/no.json b/custom_components/tesla_custom/translations/no.json
index 38826c86..d6069404 100644
--- a/custom_components/tesla_custom/translations/no.json
+++ b/custom_components/tesla_custom/translations/no.json
@@ -15,7 +15,7 @@
"password": "Passord",
"username": "E-post"
},
- "description": "Vennligst fyll inn din informasjonen.",
+ "description": "Bruk 'Auth App for Tesla' på iOS eller 'Tesla Tokens' på Android for å generere et 'refresh token' du kan bruke her.",
"title": "Tesla - Konfigurasjon"
}
}
diff --git a/info.md b/info.md
index 8b04eb18..e215b2d3 100644
--- a/info.md
+++ b/info.md
@@ -12,11 +12,13 @@
[![Discord][discord-shield]][discord]
[![Community Forum][forum-shield]][forum]
-A fork of the [official Tesla integration](https://www.home-assistant.io/integrations/tesla/) in Home Assistant to use an oauth proxy for logins.
+A fork of the [official Tesla integration](https://www.home-assistant.io/integrations/tesla/) in Home Assistant.
-This fork uses an oauth proxy instead of screen scraping which was [rejected by HA](https://github.com/home-assistant/core/pull/46558#issuecomment-822858608). The oauth proxy sits as a middleman between Home Assistant and Tesla to intercept login credentials such as your account and password. Due to the way the HTTP server works in Home Assistant, the auth endpoint cannot be turned off although we protect access by requiring knowledge of a ongoing config flow id. However, for maximum security, restart Home Assistant to completely disable the proxy server.
+This is the successor to the core app which was removed due to Tesla login issues. Do not report issues to Home Assistant.
-To the extent the official component adds features unrelated to the login, we will attempt to keep up to date. Users are welcome to port any fixes in this custom integration into HA. Please note that this component will not have the same quality or support as the official component. Do not report issues to Home Assistant.
+To use the component, you will need an application to generate a Tesla refresh token:
+- [Tesla Tokens](https://play.google.com/store/apps/details?id=net.leveugle.teslatokens)
+- [Auth App for Tesla](https://apps.apple.com/us/app/auth-app-for-tesla/id1552058613)
{% if not installed %}
diff --git a/tests/test_config_flow.py b/tests/test_config_flow.py
index ef47ee9f..25ee9759 100644
--- a/tests/test_config_flow.py
+++ b/tests/test_config_flow.py
@@ -1,185 +1,43 @@
"""Test the Tesla config flow."""
import datetime
-from unittest.mock import AsyncMock, patch
+from unittest.mock import patch
-from aiohttp import web
-from homeassistant import config_entries, data_entry_flow
-from homeassistant.components import http
+from homeassistant import config_entries, data_entry_flow, setup
from homeassistant.const import (
CONF_ACCESS_TOKEN,
- CONF_PASSWORD,
CONF_SCAN_INTERVAL,
CONF_TOKEN,
CONF_USERNAME,
HTTP_NOT_FOUND,
- HTTP_UNAUTHORIZED,
)
-from homeassistant.data_entry_flow import UnknownFlow
-from homeassistant.helpers.network import NoURLAvailableError
-from homeassistant.setup import async_setup_component
-import pytest
from pytest_homeassistant_custom_component.common import MockConfigEntry
-from teslajsonpy import TeslaException
-import voluptuous as vol
-from yarl import URL
-
-from custom_components.tesla_custom.config_flow import (
- TeslaAuthorizationCallbackView,
- TeslaAuthorizationProxyView,
- validate_input,
-)
+from teslajsonpy.exceptions import IncompleteCredentials, TeslaException
+
from custom_components.tesla_custom.const import (
- AUTH_CALLBACK_PATH,
- AUTH_PROXY_PATH,
CONF_EXPIRATION,
CONF_WAKE_ON_START,
DEFAULT_SCAN_INTERVAL,
DEFAULT_WAKE_ON_START,
DOMAIN,
- ERROR_URL_NOT_DETECTED,
MIN_SCAN_INTERVAL,
)
-HA_URL = "https://homeassistant.com"
TEST_USERNAME = "test-username"
TEST_TOKEN = "test-token"
+TEST_PASSWORD = "test-password"
TEST_ACCESS_TOKEN = "test-access-token"
TEST_VALID_EXPIRATION = datetime.datetime.now().timestamp() * 2
-TEST_INVALID_EXPIRATION = 0
-async def test_warning_form(hass):
- """Test we get the warning form."""
+async def test_form(hass):
+ """Test we get the form."""
+ await setup.async_setup_component(hass, "persistent_notification", {})
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
- # "type": RESULT_TYPE_FORM,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": step_id,
- # "data_schema": data_schema,
- # "errors": errors,
- # "description_placeholders": description_placeholders,
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["handler"] == DOMAIN
- assert result["step_id"] == "user"
- assert result["data_schema"] == vol.Schema({})
- assert result["errors"] == {}
- assert result["description_placeholders"] == {}
- return result
-
-
-async def test_reauth_warning_form(hass):
- """Test we get the warning form on reauth."""
- result = await hass.config_entries.flow.async_init(
- DOMAIN, context={"source": config_entries.SOURCE_REAUTH}
- )
- # "type": RESULT_TYPE_FORM,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": step_id,
- # "data_schema": data_schema,
- # "errors": errors,
- # "description_placeholders": description_placeholders,
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["handler"] == DOMAIN
- assert result["step_id"] == "user"
- assert result["data_schema"] == vol.Schema({})
+ assert result["type"] == "form"
assert result["errors"] == {}
- assert result["description_placeholders"] == {}
- return result
-
-async def test_external_url(hass, callback_view):
- """Test we get the external url after submitting once."""
- result = await test_warning_form(hass)
- flow_id = result["flow_id"]
- with patch(
- "custom_components.tesla_custom.config_flow.get_url",
- return_value=HA_URL,
- ):
- result = await hass.config_entries.flow.async_configure(
- flow_id,
- user_input={},
- )
- # "type": RESULT_TYPE_EXTERNAL_STEP,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": step_id,
- # "url": url,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["step_id"] == "check_proxy"
- callback_url: str = str(
- URL(HA_URL).with_path(AUTH_CALLBACK_PATH).with_query({"flow_id": flow_id})
- )
- assert result["url"] == str(
- URL(HA_URL)
- .with_path(AUTH_PROXY_PATH)
- .with_query({"config_flow_id": flow_id, "callback_url": callback_url})
- )
- assert result["description_placeholders"] is None
- return result
-
-
-async def test_external_url_no_hass_url_exception(hass):
- """Test we handle case with no detectable hass external url."""
- result = await test_warning_form(hass)
- flow_id = result["flow_id"]
- with patch(
- "custom_components.tesla_custom.config_flow.get_url",
- side_effect=NoURLAvailableError,
- ):
- result = await hass.config_entries.flow.async_configure(
- flow_id,
- user_input={},
- )
- # "type": RESULT_TYPE_EXTERNAL_STEP,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": step_id,
- # "url": url,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["handler"] == DOMAIN
- assert result["step_id"] == "user"
- assert result["data_schema"] == vol.Schema({})
- assert result["errors"] == {"base": ERROR_URL_NOT_DETECTED}
- assert result["description_placeholders"] == {}
-
-
-async def test_external_url_callback(hass, callback_view):
- """Test we get the processing of callback_url."""
- result = await test_external_url(hass, callback_view)
- flow_id = result["flow_id"]
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={
- CONF_USERNAME: TEST_USERNAME,
- CONF_TOKEN: TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- },
- )
- # "type": RESULT_TYPE_EXTERNAL_STEP_DONE,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": next_step_id,
- assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP_DONE
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["step_id"] == "finish_oauth"
- return result
-
-
-async def test_finish_oauth(hass, callback_view):
- """Test config entry after finishing oauth."""
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
return_value={
@@ -187,126 +45,99 @@ async def test_finish_oauth(hass, callback_view):
CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
CONF_EXPIRATION: TEST_VALID_EXPIRATION,
},
- ):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={},
+ ), patch(
+ "custom_components.tesla_custom.async_setup", return_value=True
+ ) as mock_setup, patch(
+ "custom_components.tesla_custom.async_setup_entry", return_value=True
+ ) as mock_setup_entry:
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_TOKEN: TEST_TOKEN, CONF_USERNAME: "test@email.com"}
)
- # "version": self.VERSION,
- # "type": RESULT_TYPE_CREATE_ENTRY,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "title": title,
- # "data": data,
- # "description": description,
- # "description_placeholders": description_placeholders,
- assert result["version"] == 1
- assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["title"] == TEST_USERNAME
- assert result["data"] == {
+ await hass.async_block_till_done()
+
+ assert result2["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result2["title"] == "test@email.com"
+ assert result2["data"] == {
+ CONF_USERNAME: "test@email.com",
+ CONF_TOKEN: TEST_TOKEN,
CONF_TOKEN: TEST_TOKEN,
CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
CONF_EXPIRATION: TEST_VALID_EXPIRATION,
}
- assert result["description"] is None
- assert result["description_placeholders"] is None
- return result
+ assert len(mock_setup.mock_calls) == 1
+ assert len(mock_setup_entry.mock_calls) == 1
+
+async def test_form_invalid_auth(hass):
+ """Test we handle invalid auth."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
-async def test_form_invalid_auth(hass, callback_view):
- """Test we handle invalid auth error."""
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
- side_effect=TeslaException(code=HTTP_UNAUTHORIZED),
+ side_effect=TeslaException(401),
):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={},
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: TEST_TOKEN},
)
- # "type": RESULT_TYPE_ABORT,
- # "flow_id": flow_id,
- # "handler": handler,
- # "reason": reason,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["reason"] == "invalid_auth"
- assert result["description_placeholders"] is None
+
+ assert result2["type"] == "form"
+ assert result2["errors"] == {"base": "invalid_auth"}
-async def test_form_login_failed(hass, callback_view):
- """Test we handle invalid auth error."""
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
+async def test_form_invalid_auth_incomplete_credentials(hass):
+ """Test we handle invalid auth with incomplete credentials."""
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
- return_value={},
+ side_effect=IncompleteCredentials(401),
):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={},
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: TEST_TOKEN},
)
- # "type": RESULT_TYPE_ABORT,
- # "flow_id": flow_id,
- # "handler": handler,
- # "reason": reason,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["reason"] == "login_failed"
- assert result["description_placeholders"] is None
+
+ assert result2["type"] == "form"
+ assert result2["errors"] == {"base": "invalid_auth"}
-async def test_form_cannot_connect(hass, callback_view):
+async def test_form_cannot_connect(hass):
"""Test we handle cannot connect error."""
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
+
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
side_effect=TeslaException(code=HTTP_NOT_FOUND),
):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={},
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {CONF_TOKEN: TEST_TOKEN, CONF_USERNAME: TEST_USERNAME},
)
- # "type": RESULT_TYPE_ABORT,
- # "flow_id": flow_id,
- # "handler": handler,
- # "reason": reason,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["reason"] == "cannot_connect"
- assert result["description_placeholders"] is None
+ assert result2["type"] == "form"
+ assert result2["errors"] == {"base": "cannot_connect"}
-async def test_form_repeat_identifier(hass, callback_view):
- """Test we handle repeat identifiers.
- Repeats are identified if the title and tokens are identical. Otherwise they are
- replaced.
- """
+async def test_form_repeat_identifier(hass):
+ """Test we handle repeat identifiers."""
entry = MockConfigEntry(
domain=DOMAIN,
title=TEST_USERNAME,
- data={
- CONF_TOKEN: TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- },
+ data={CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: TEST_TOKEN},
options=None,
)
entry.add_to_hass(hass)
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": config_entries.SOURCE_USER}
+ )
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
return_value={
@@ -315,59 +146,30 @@ async def test_form_repeat_identifier(hass, callback_view):
CONF_EXPIRATION: TEST_VALID_EXPIRATION,
},
):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={},
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: TEST_TOKEN},
)
- # "type": RESULT_TYPE_ABORT,
- # "flow_id": flow_id,
- # "handler": handler,
- # "reason": reason,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["reason"] == "already_configured"
- assert result["description_placeholders"] is None
+ assert result2["type"] == "abort"
+ assert result2["reason"] == "already_configured"
-async def test_form_second_identifier(hass, callback_view):
- """Test we can create another entry with a different name.
- Repeats are identified if the title and tokens are identical. Otherwise they are
- replaced.
- """
- entry = MockConfigEntry(
- domain=DOMAIN,
- title="OTHER_USERNAME",
- data={
- CONF_TOKEN: TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- },
- options=None,
- )
- entry.add_to_hass(hass)
- await test_finish_oauth(hass, callback_view)
- assert len(hass.config_entries.async_entries(DOMAIN)) == 2
-
-
-async def test_form_reauth(hass, callback_view):
+async def test_form_reauth(hass):
"""Test we handle reauth."""
entry = MockConfigEntry(
domain=DOMAIN,
title=TEST_USERNAME,
- data={
- CONF_TOKEN: TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_INVALID_EXPIRATION,
- },
+ data={CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: TEST_TOKEN},
options=None,
)
entry.add_to_hass(hass)
- result = await test_external_url_callback(hass, callback_view)
- flow_id = result["flow_id"]
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN,
+ context={"source": config_entries.SOURCE_REAUTH},
+ data={CONF_USERNAME: TEST_USERNAME},
+ )
with patch(
"custom_components.tesla_custom.config_flow.TeslaAPI.connect",
return_value={
@@ -376,46 +178,36 @@ async def test_form_reauth(hass, callback_view):
CONF_EXPIRATION: TEST_VALID_EXPIRATION,
},
):
- result = await hass.config_entries.flow.async_configure(
- flow_id=flow_id,
- user_input={
- # CONF_TOKEN: TEST_TOKEN,
- # CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- # CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- },
+ result2 = await hass.config_entries.flow.async_configure(
+ result["flow_id"],
+ {CONF_USERNAME: TEST_USERNAME, CONF_TOKEN: "new-password"},
)
- # "type": RESULT_TYPE_ABORT,
- # "flow_id": flow_id,
- # "handler": handler,
- # "reason": reason,
- # "description_placeholders": description_placeholders,
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["flow_id"] == flow_id
- assert result["handler"] == DOMAIN
- assert result["reason"] == "reauth_successful"
- assert result["description_placeholders"] is None
+ assert result2["type"] == "abort"
+ assert result2["reason"] == "reauth_successful"
-async def test_import(hass):
- """Test import step results in warning form."""
- result = await hass.config_entries.flow.async_init(
- DOMAIN,
- context={"source": config_entries.SOURCE_IMPORT},
- data={CONF_PASSWORD: "test-password", CONF_USERNAME: "test-username"},
- )
- # "type": RESULT_TYPE_FORM,
- # "flow_id": self.flow_id,
- # "handler": self.handler,
- # "step_id": step_id,
- # "data_schema": data_schema,
- # "errors": errors,
- # "description_placeholders": description_placeholders,
+async def test_import(hass):
+ """Test import step."""
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "user"
- assert result["data_schema"] == vol.Schema({})
- assert result["description_placeholders"] == {}
+ with patch(
+ "custom_components.tesla_custom.config_flow.TeslaAPI.connect",
+ return_value={
+ "refresh_token": TEST_TOKEN,
+ CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
+ CONF_EXPIRATION: TEST_VALID_EXPIRATION,
+ },
+ ):
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN,
+ context={"source": config_entries.SOURCE_IMPORT},
+ data={CONF_TOKEN: TEST_TOKEN, CONF_USERNAME: TEST_USERNAME},
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["title"] == TEST_USERNAME
+ assert result["data"][CONF_ACCESS_TOKEN] == TEST_ACCESS_TOKEN
+ assert result["data"][CONF_TOKEN] == TEST_TOKEN
+ assert result["description_placeholders"] is None
async def test_option_flow(hass):
@@ -474,239 +266,3 @@ async def test_option_flow_input_floor(hass):
CONF_SCAN_INTERVAL: MIN_SCAN_INTERVAL,
CONF_WAKE_ON_START: DEFAULT_WAKE_ON_START,
}
-
-
-@pytest.fixture
-async def callback_view(hass, aiohttp_unused_port):
- """Generate registered callback_view fixture."""
- await async_setup_component(
- hass, http.DOMAIN, {http.DOMAIN: {http.CONF_SERVER_PORT: aiohttp_unused_port()}}
- )
- await async_setup_component(hass, DOMAIN, {})
- await hass.async_start()
-
- hass.http.register_view(TeslaAuthorizationCallbackView)
- return callback_view
-
-
-async def test_callback_view_invalid_query(hass, aiohttp_client, callback_view):
- """Test callback view with invalid query."""
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.get(AUTH_CALLBACK_PATH)
- assert resp.status == 400
-
- resp = await client.get(
- AUTH_CALLBACK_PATH, params={"api_password": "test-password"}
- )
- assert resp.status == 400
-
- # https://alandtse-test.duckdns.org/auth/tesla/callback?flow_id=7c0bdd32efca42c9bc8ce9c27f431f12&code=67443912fda4a307767a47081c55085650db40069aabd293da57185719c2&username=alandtse@gmail.com&domain=auth.tesla.com
- resp = await client.get(AUTH_CALLBACK_PATH, params={"flow_id": 1234})
- assert resp.status == 400
-
- with patch(
- "custom_components.tesla_custom.async_setup_entry", side_effect=KeyError
- ):
- resp = await client.get(AUTH_CALLBACK_PATH, params={"flow_id": 1234})
- assert resp.status == 400
-
-
-async def test_callback_view_keyerror(hass, aiohttp_client, callback_view):
- """Test callback view with keyerror."""
- client = await aiohttp_client(hass.http.app)
-
- with patch(
- "custom_components.tesla_custom.async_setup_entry", side_effect=KeyError
- ):
- resp = await client.get(AUTH_CALLBACK_PATH, params={"flow_id": 1234})
- assert resp.status == 400
-
-
-async def test_callback_view_unknownflow(hass, aiohttp_client, callback_view):
- """Test callback view with unknownflow."""
- client = await aiohttp_client(hass.http.app)
-
- with patch(
- "custom_components.tesla_custom.async_setup_entry", side_effect=UnknownFlow
- ):
- resp = await client.get(AUTH_CALLBACK_PATH, params={"flow_id": 1234})
- assert resp.status == 400
-
-
-async def test_callback_view_success(hass, aiohttp_client, callback_view):
- """Test callback view with success response."""
- result = await test_external_url(hass, callback_view)
- flow_id = result["flow_id"]
-
- client = await aiohttp_client(hass.http.app)
-
- with patch("custom_components.tesla_custom.async_setup_entry", return_value=True):
- resp = await client.get(AUTH_CALLBACK_PATH, params={"flow_id": flow_id})
- assert resp.status == 200
- assert (
- "Success! This window can be closed"
- in await resp.text()
- )
-
-
-@pytest.fixture
-async def proxy_view(hass, aiohttp_unused_port):
- """Generate registered proxy_view fixture."""
- await async_setup_component(
- hass, http.DOMAIN, {http.DOMAIN: {http.CONF_SERVER_PORT: aiohttp_unused_port()}}
- )
- await async_setup_component(hass, DOMAIN, {})
- await hass.async_start()
-
- mock_handler = AsyncMock(return_value=web.Response(text="Success"))
- proxy_view = TeslaAuthorizationProxyView(mock_handler)
- hass.http.register_view(proxy_view)
- return proxy_view
-
-
-@pytest.fixture
-async def proxy_view_with_flow(hass, proxy_view):
- """Generate registered proxy_view fixture with running flow."""
- result = await hass.config_entries.flow.async_init(
- DOMAIN, context={"source": config_entries.SOURCE_USER}
- )
- flow_id = result["flow_id"]
- return flow_id
-
-
-async def test_proxy_view_invalid_auth(hass, aiohttp_client, proxy_view):
- """Test proxy view request results in auth error."""
-
- client = await aiohttp_client(hass.http.app)
-
- for method in ("get", "post", "delete", "put", "patch", "head", "options"):
- resp = await getattr(client, method)(AUTH_PROXY_PATH)
- assert resp.status in [403, 401]
-
-
-async def test_proxy_view_valid_auth_get(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view get request results in valid response."""
- flow_id = proxy_view_with_flow
-
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.get(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_post(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view post request results in valid response."""
- flow_id = proxy_view_with_flow
-
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.post(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_delete(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view delete request results in valid response."""
- flow_id = proxy_view_with_flow
-
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.delete(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_put(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view put request results in valid response."""
- flow_id = proxy_view_with_flow
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.put(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_patch(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view patch request results in valid response."""
- flow_id = proxy_view_with_flow
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.patch(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_head(hass, aiohttp_client, proxy_view_with_flow):
- """Test proxy view head request results in valid response."""
- flow_id = proxy_view_with_flow
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.head(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
-
-async def test_proxy_view_valid_auth_options(
- hass, aiohttp_client, proxy_view_with_flow
-):
- """Test proxy view options request results in valid response."""
- flow_id = proxy_view_with_flow
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.options(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 403
-
-
-async def test_proxy_view_invalid_auth_after_reset(
- hass, aiohttp_client, proxy_view, proxy_view_with_flow
-):
- """Test proxy view request results in invalid auth response after reset."""
- flow_id = proxy_view_with_flow
- client = await aiohttp_client(hass.http.app)
-
- resp = await client.get(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 200
-
- proxy_view.reset()
- hass.config_entries.flow.async_abort(flow_id)
- resp = await client.get(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.post(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.delete(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.put(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.patch(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.head(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 401
-
- resp = await client.options(AUTH_PROXY_PATH, params={"config_flow_id": flow_id})
- assert resp.status == 403
-
-
-async def test_validate_input_no_controller(
- hass,
-):
- """Test validate input."""
- user_input = {
- CONF_USERNAME: TEST_USERNAME,
- CONF_TOKEN: TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- }
- with patch(
- "custom_components.tesla_custom.config_flow.TeslaAPI.connect",
- return_value={
- "refresh_token": TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- },
- ):
- assert await validate_input(hass, user_input) == {
- "refresh_token": TEST_TOKEN,
- CONF_ACCESS_TOKEN: TEST_ACCESS_TOKEN,
- CONF_EXPIRATION: TEST_VALID_EXPIRATION,
- }