diff --git a/homeassistant/components/smartthings/__init__.py b/homeassistant/components/smartthings/__init__.py
index 97a7d32a9c1..e4d720c94e5 100644
--- a/homeassistant/components/smartthings/__init__.py
+++ b/homeassistant/components/smartthings/__init__.py
@@ -37,6 +37,7 @@ from .const import (
TOKEN_REFRESH_INTERVAL,
)
from .smartapp import (
+ format_unique_id,
setup_smartapp,
setup_smartapp_endpoint,
smartapp_sync_subscriptions,
@@ -76,6 +77,15 @@ async def async_migrate_entry(hass: HomeAssistantType, entry: ConfigEntry):
async def async_setup_entry(hass: HomeAssistantType, entry: ConfigEntry):
"""Initialize config entry which represents an installed SmartApp."""
+ # For backwards compat
+ if entry.unique_id is None:
+ hass.config_entries.async_update_entry(
+ entry,
+ unique_id=format_unique_id(
+ entry.data[CONF_APP_ID], entry.data[CONF_LOCATION_ID]
+ ),
+ )
+
if not validate_webhook_requirements(hass):
_LOGGER.warning(
"The 'base_url' of the 'http' integration must be configured and start with 'https://'"
diff --git a/homeassistant/components/smartthings/config_flow.py b/homeassistant/components/smartthings/config_flow.py
index cb4623cea1c..c03ade4d8b1 100644
--- a/homeassistant/components/smartthings/config_flow.py
+++ b/homeassistant/components/smartthings/config_flow.py
@@ -7,7 +7,7 @@ from pysmartthings.installedapp import format_install_url
import voluptuous as vol
from homeassistant import config_entries
-from homeassistant.const import CONF_ACCESS_TOKEN, HTTP_FORBIDDEN
+from homeassistant.const import CONF_ACCESS_TOKEN, HTTP_FORBIDDEN, HTTP_UNAUTHORIZED
from homeassistant.helpers.aiohttp_client import async_get_clientsession
# pylint: disable=unused-import
@@ -26,6 +26,7 @@ from .const import (
from .smartapp import (
create_app,
find_app,
+ format_unique_id,
get_webhook_url,
setup_smartapp,
setup_smartapp_endpoint,
@@ -138,7 +139,7 @@ class SmartThingsFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
return self._show_step_pat(errors)
except ClientResponseError as ex:
- if ex.status == 401:
+ if ex.status == HTTP_UNAUTHORIZED:
errors[CONF_ACCESS_TOKEN] = "token_unauthorized"
_LOGGER.debug(
"Unauthorized error received setting up SmartApp", exc_info=True
@@ -183,6 +184,7 @@ class SmartThingsFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
self.location_id = user_input[CONF_LOCATION_ID]
+ await self.async_set_unique_id(format_unique_id(self.app_id, self.location_id))
return await self.async_step_authorize()
async def async_step_authorize(self, user_input=None):
diff --git a/homeassistant/components/smartthings/smartapp.py b/homeassistant/components/smartthings/smartapp.py
index 0b86a430d89..7d02a04d2ff 100644
--- a/homeassistant/components/smartthings/smartapp.py
+++ b/homeassistant/components/smartthings/smartapp.py
@@ -39,7 +39,6 @@ from .const import (
CONF_CLOUDHOOK_URL,
CONF_INSTALLED_APP_ID,
CONF_INSTANCE_ID,
- CONF_LOCATION_ID,
CONF_REFRESH_TOKEN,
DATA_BROKERS,
DATA_MANAGER,
@@ -53,6 +52,11 @@ from .const import (
_LOGGER = logging.getLogger(__name__)
+def format_unique_id(app_id: str, location_id: str) -> str:
+ """Format the unique id for a config entry."""
+ return f"{app_id}_{location_id}"
+
+
async def find_app(hass: HomeAssistantType, api):
"""Find an existing SmartApp for this installation of hass."""
apps = await api.apps()
@@ -366,13 +370,20 @@ async def smartapp_sync_subscriptions(
_LOGGER.debug("Subscriptions for app '%s' are up-to-date", installed_app_id)
-async def smartapp_install(hass: HomeAssistantType, req, resp, app):
- """Handle a SmartApp installation and continue the config flow."""
+async def _continue_flow(
+ hass: HomeAssistantType,
+ app_id: str,
+ location_id: str,
+ installed_app_id: str,
+ refresh_token: str,
+):
+ """Continue a config flow if one is in progress for the specific installed app."""
+ unique_id = format_unique_id(app_id, location_id)
flow = next(
(
flow
for flow in hass.config_entries.flow.async_progress()
- if flow["handler"] == DOMAIN
+ if flow["handler"] == DOMAIN and flow["context"]["unique_id"] == unique_id
),
None,
)
@@ -380,18 +391,23 @@ async def smartapp_install(hass: HomeAssistantType, req, resp, app):
await hass.config_entries.flow.async_configure(
flow["flow_id"],
{
- CONF_INSTALLED_APP_ID: req.installed_app_id,
- CONF_LOCATION_ID: req.location_id,
- CONF_REFRESH_TOKEN: req.refresh_token,
+ CONF_INSTALLED_APP_ID: installed_app_id,
+ CONF_REFRESH_TOKEN: refresh_token,
},
)
_LOGGER.debug(
"Continued config flow '%s' for SmartApp '%s' under parent app '%s'",
flow["flow_id"],
- req.installed_app_id,
- app.app_id,
+ installed_app_id,
+ app_id,
)
+
+async def smartapp_install(hass: HomeAssistantType, req, resp, app):
+ """Handle a SmartApp installation and continue the config flow."""
+ await _continue_flow(
+ hass, app.app_id, req.location_id, req.installed_app_id, req.refresh_token
+ )
_LOGGER.debug(
"Installed SmartApp '%s' under parent app '%s'",
req.installed_app_id,
@@ -420,30 +436,9 @@ async def smartapp_update(hass: HomeAssistantType, req, resp, app):
app.app_id,
)
- flow = next(
- (
- flow
- for flow in hass.config_entries.flow.async_progress()
- if flow["handler"] == DOMAIN
- ),
- None,
+ await _continue_flow(
+ hass, app.app_id, req.location_id, req.installed_app_id, req.refresh_token
)
- if flow is not None:
- await hass.config_entries.flow.async_configure(
- flow["flow_id"],
- {
- CONF_INSTALLED_APP_ID: req.installed_app_id,
- CONF_LOCATION_ID: req.location_id,
- CONF_REFRESH_TOKEN: req.refresh_token,
- },
- )
- _LOGGER.debug(
- "Continued config flow '%s' for SmartApp '%s' under parent app '%s'",
- flow["flow_id"],
- req.installed_app_id,
- app.app_id,
- )
-
_LOGGER.debug(
"Updated SmartApp '%s' under parent app '%s'", req.installed_app_id, app.app_id
)
diff --git a/tests/components/smartthings/test_config_flow.py b/tests/components/smartthings/test_config_flow.py
index dc046f718a8..81dbab917a3 100644
--- a/tests/components/smartthings/test_config_flow.py
+++ b/tests/components/smartthings/test_config_flow.py
@@ -8,28 +8,30 @@ from pysmartthings.installedapp import format_install_url
from homeassistant import data_entry_flow
from homeassistant.components.smartthings import smartapp
-from homeassistant.components.smartthings.config_flow import SmartThingsFlowHandler
from homeassistant.components.smartthings.const import (
CONF_APP_ID,
CONF_INSTALLED_APP_ID,
CONF_LOCATION_ID,
CONF_OAUTH_CLIENT_ID,
CONF_OAUTH_CLIENT_SECRET,
- CONF_REFRESH_TOKEN,
DOMAIN,
)
-from homeassistant.const import CONF_ACCESS_TOKEN, HTTP_FORBIDDEN, HTTP_NOT_FOUND
+from homeassistant.const import (
+ CONF_ACCESS_TOKEN,
+ HTTP_FORBIDDEN,
+ HTTP_NOT_FOUND,
+ HTTP_UNAUTHORIZED,
+)
from tests.common import MockConfigEntry, mock_coro
-async def test_step_import(hass):
- """Test import returns user."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_import()
-
+async def test_import_shows_user_step(hass):
+ """Test import source shows the user form."""
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "import"}
+ )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "user"
assert result["description_placeholders"][
@@ -37,237 +39,316 @@ async def test_step_import(hass):
] == smartapp.get_webhook_url(hass)
-async def test_step_user(hass):
- """Test the webhook confirmation is shown."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_user()
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "user"
- assert result["description_placeholders"][
- "webhook_url"
- ] == smartapp.get_webhook_url(hass)
-
-
-async def test_step_user_aborts_invalid_webhook(hass):
- """Test flow aborts if webhook is invalid."""
- hass.config.api.base_url = "http://0.0.0.0"
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_user()
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
- assert result["reason"] == "invalid_webhook_url"
- assert result["description_placeholders"][
- "webhook_url"
- ] == smartapp.get_webhook_url(hass)
- assert "component_url" in result["description_placeholders"]
-
-
-async def test_step_user_advances_to_pat(hass):
- """Test user step advances to the pat step."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_user({})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
-
-
-async def test_step_pat(hass):
- """Test pat step shows the input form."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_pat()
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {}
- assert result["data_schema"]({CONF_ACCESS_TOKEN: ""}) == {CONF_ACCESS_TOKEN: ""}
- assert "token_url" in result["description_placeholders"]
- assert "component_url" in result["description_placeholders"]
-
-
-async def test_step_pat_defaults_token(hass):
- """Test pat form defaults the token from another entry."""
+async def test_entry_created(hass, app, app_oauth_client, location, smartthings_mock):
+ """Test local webhook, new app, install event creates entry."""
token = str(uuid4())
- entry = MockConfigEntry(domain=DOMAIN, data={CONF_ACCESS_TOKEN: token})
- entry.add_to_hass(hass)
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_pat()
-
- assert flow.access_token == token
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
- assert "token_url" in result["description_placeholders"]
- assert "component_url" in result["description_placeholders"]
-
-
-async def test_step_pat_invalid_token(hass):
- """Test an error is shown for invalid token formats."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- token = "123456789"
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
- assert result["errors"] == {"access_token": "token_invalid_format"}
- assert "token_url" in result["description_placeholders"]
- assert "component_url" in result["description_placeholders"]
-
-
-async def test_step_pat_unauthorized(hass, smartthings_mock):
- """Test an error is shown when the token is not authorized."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- request_info = Mock(real_url="http://example.com")
- smartthings_mock.apps.side_effect = ClientResponseError(
- request_info=request_info, history=None, status=401
- )
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {CONF_ACCESS_TOKEN: "token_unauthorized"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_forbidden(hass, smartthings_mock):
- """Test an error is shown when the token is forbidden."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- request_info = Mock(real_url="http://example.com")
- smartthings_mock.apps.side_effect = ClientResponseError(
- request_info=request_info, history=None, status=HTTP_FORBIDDEN
- )
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {CONF_ACCESS_TOKEN: "token_forbidden"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_webhook_error(hass, smartthings_mock):
- """Test an error is shown when there's an problem with the webhook endpoint."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- data = {"error": {}}
- request_info = Mock(real_url="http://example.com")
- error = APIResponseError(
- request_info=request_info, history=None, data=data, status=422
- )
- error.is_target_error = Mock(return_value=True)
- smartthings_mock.apps.side_effect = error
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {"base": "webhook_error"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_api_error(hass, smartthings_mock):
- """Test an error is shown when other API errors occur."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- data = {"error": {}}
- request_info = Mock(real_url="http://example.com")
- error = APIResponseError(
- request_info=request_info, history=None, data=data, status=400
- )
- smartthings_mock.apps.side_effect = error
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {"base": "app_setup_error"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_unknown_api_error(hass, smartthings_mock):
- """Test an error is shown when there is an unknown API error."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- request_info = Mock(real_url="http://example.com")
- smartthings_mock.apps.side_effect = ClientResponseError(
- request_info=request_info, history=None, status=HTTP_NOT_FOUND
- )
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {"base": "app_setup_error"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_unknown_error(hass, smartthings_mock):
- """Test an error is shown when there is an unknown API error."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- smartthings_mock.apps.side_effect = Exception("Unknown error")
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "pat"
- assert result["errors"] == {"base": "app_setup_error"}
- assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
-
-
-async def test_step_pat_app_created_webhook(
- hass, app, app_oauth_client, location, smartthings_mock
-):
- """Test SmartApp is created when one does not exist and shows location form."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
+ installed_app_id = str(uuid4())
+ refresh_token = str(uuid4())
smartthings_mock.apps.return_value = []
smartthings_mock.create_app.return_value = (app, app_oauth_client)
smartthings_mock.locations.return_value = [location]
- token = str(uuid4())
+ request = Mock()
+ request.installed_app_id = installed_app_id
+ request.auth_token = token
+ request.location_id = location.location_id
+ request.refresh_token = refresh_token
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
- assert flow.access_token == token
- assert flow.app_id == app.app_id
- assert flow.oauth_client_secret == app_oauth_client.client_secret
- assert flow.oauth_client_id == app_oauth_client.client_id
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
assert result["step_id"] == "select_location"
+ # Select location and advance to external auth
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_LOCATION_ID: location.location_id}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
+ assert result["step_id"] == "authorize"
+ assert result["url"] == format_install_url(app.app_id, location.location_id)
-async def test_step_pat_app_created_cloudhook(
+ # Complete external auth and advance to install
+ await smartapp.smartapp_install(hass, request, None, app)
+
+ # Finish
+ result = await hass.config_entries.flow.async_configure(result["flow_id"])
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["data"]["app_id"] == app.app_id
+ assert result["data"]["installed_app_id"] == installed_app_id
+ assert result["data"]["location_id"] == location.location_id
+ assert result["data"]["access_token"] == token
+ assert result["data"]["refresh_token"] == request.refresh_token
+ assert result["data"]["client_secret"] == app_oauth_client.client_secret
+ assert result["data"]["client_id"] == app_oauth_client.client_id
+ assert result["title"] == location.name
+ entry = next((entry for entry in hass.config_entries.async_entries(DOMAIN)), None,)
+ assert entry.unique_id == smartapp.format_unique_id(
+ app.app_id, location.location_id
+ )
+
+
+async def test_entry_created_from_update_event(
hass, app, app_oauth_client, location, smartthings_mock
):
- """Test SmartApp is created with a cloudhook and shows location form."""
- hass.config.components.add("cloud")
+ """Test local webhook, new app, update event creates entry."""
+ token = str(uuid4())
+ installed_app_id = str(uuid4())
+ refresh_token = str(uuid4())
+ smartthings_mock.apps.return_value = []
+ smartthings_mock.create_app.return_value = (app, app_oauth_client)
+ smartthings_mock.locations.return_value = [location]
+ request = Mock()
+ request.installed_app_id = installed_app_id
+ request.auth_token = token
+ request.location_id = location.location_id
+ request.refresh_token = refresh_token
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "select_location"
+
+ # Select location and advance to external auth
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_LOCATION_ID: location.location_id}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
+ assert result["step_id"] == "authorize"
+ assert result["url"] == format_install_url(app.app_id, location.location_id)
+
+ # Complete external auth and advance to install
+ await smartapp.smartapp_update(hass, request, None, app)
+
+ # Finish
+ result = await hass.config_entries.flow.async_configure(result["flow_id"])
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["data"]["app_id"] == app.app_id
+ assert result["data"]["installed_app_id"] == installed_app_id
+ assert result["data"]["location_id"] == location.location_id
+ assert result["data"]["access_token"] == token
+ assert result["data"]["refresh_token"] == request.refresh_token
+ assert result["data"]["client_secret"] == app_oauth_client.client_secret
+ assert result["data"]["client_id"] == app_oauth_client.client_id
+ assert result["title"] == location.name
+ entry = next((entry for entry in hass.config_entries.async_entries(DOMAIN)), None,)
+ assert entry.unique_id == smartapp.format_unique_id(
+ app.app_id, location.location_id
+ )
+
+
+async def test_entry_created_existing_app_new_oauth_client(
+ hass, app, app_oauth_client, location, smartthings_mock
+):
+ """Test entry is created with an existing app and generation of a new oauth client."""
+ token = str(uuid4())
+ installed_app_id = str(uuid4())
+ refresh_token = str(uuid4())
+ smartthings_mock.apps.return_value = [app]
+ smartthings_mock.generate_app_oauth.return_value = app_oauth_client
+ smartthings_mock.locations.return_value = [location]
+ request = Mock()
+ request.installed_app_id = installed_app_id
+ request.auth_token = token
+ request.location_id = location.location_id
+ request.refresh_token = refresh_token
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "select_location"
+
+ # Select location and advance to external auth
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_LOCATION_ID: location.location_id}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
+ assert result["step_id"] == "authorize"
+ assert result["url"] == format_install_url(app.app_id, location.location_id)
+
+ # Complete external auth and advance to install
+ await smartapp.smartapp_install(hass, request, None, app)
+
+ # Finish
+ result = await hass.config_entries.flow.async_configure(result["flow_id"])
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["data"]["app_id"] == app.app_id
+ assert result["data"]["installed_app_id"] == installed_app_id
+ assert result["data"]["location_id"] == location.location_id
+ assert result["data"]["access_token"] == token
+ assert result["data"]["refresh_token"] == request.refresh_token
+ assert result["data"]["client_secret"] == app_oauth_client.client_secret
+ assert result["data"]["client_id"] == app_oauth_client.client_id
+ assert result["title"] == location.name
+ entry = next((entry for entry in hass.config_entries.async_entries(DOMAIN)), None,)
+ assert entry.unique_id == smartapp.format_unique_id(
+ app.app_id, location.location_id
+ )
+
+
+async def test_entry_created_existing_app_copies_oauth_client(
+ hass, app, location, smartthings_mock
+):
+ """Test entry is created with an existing app and copies the oauth client from another entry."""
+ token = str(uuid4())
+ installed_app_id = str(uuid4())
+ refresh_token = str(uuid4())
+ oauth_client_id = str(uuid4())
+ oauth_client_secret = str(uuid4())
+ smartthings_mock.apps.return_value = [app]
+ smartthings_mock.locations.return_value = [location]
+ request = Mock()
+ request.installed_app_id = installed_app_id
+ request.auth_token = token
+ request.location_id = location.location_id
+ request.refresh_token = refresh_token
+ entry = MockConfigEntry(
+ domain=DOMAIN,
+ data={
+ CONF_APP_ID: app.app_id,
+ CONF_OAUTH_CLIENT_ID: oauth_client_id,
+ CONF_OAUTH_CLIENT_SECRET: oauth_client_secret,
+ CONF_LOCATION_ID: str(uuid4()),
+ CONF_INSTALLED_APP_ID: str(uuid4()),
+ CONF_ACCESS_TOKEN: token,
+ },
+ )
+ entry.add_to_hass(hass)
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+ # Assert access token is defaulted to an existing entry for convenience.
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "select_location"
+
+ # Select location and advance to external auth
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_LOCATION_ID: location.location_id}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
+ assert result["step_id"] == "authorize"
+ assert result["url"] == format_install_url(app.app_id, location.location_id)
+
+ # Complete external auth and advance to install
+ await smartapp.smartapp_install(hass, request, None, app)
+
+ # Finish
+ result = await hass.config_entries.flow.async_configure(result["flow_id"])
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["data"]["app_id"] == app.app_id
+ assert result["data"]["installed_app_id"] == installed_app_id
+ assert result["data"]["location_id"] == location.location_id
+ assert result["data"]["access_token"] == token
+ assert result["data"]["refresh_token"] == request.refresh_token
+ assert result["data"]["client_secret"] == oauth_client_secret
+ assert result["data"]["client_id"] == oauth_client_id
+ assert result["title"] == location.name
+ entry = next(
+ (
+ entry
+ for entry in hass.config_entries.async_entries(DOMAIN)
+ if entry.data[CONF_INSTALLED_APP_ID] == installed_app_id
+ ),
+ None,
+ )
+ assert entry.unique_id == smartapp.format_unique_id(
+ app.app_id, location.location_id
+ )
+
+
+async def test_entry_created_with_cloudhook(
+ hass, app, app_oauth_client, location, smartthings_mock
+):
+ """Test cloud, new app, install event creates entry."""
+ hass.config.components.add("cloud")
# Unload the endpoint so we can reload it under the cloud.
await smartapp.unload_smartapp_endpoint(hass)
+ token = str(uuid4())
+ installed_app_id = str(uuid4())
+ refresh_token = str(uuid4())
+ smartthings_mock.apps.return_value = []
+ smartthings_mock.create_app.return_value = (app, app_oauth_client)
+ smartthings_mock.locations.return_value = [location]
+ request = Mock()
+ request.installed_app_id = installed_app_id
+ request.auth_token = token
+ request.location_id = location.location_id
+ request.refresh_token = refresh_token
with patch.object(
hass.components.cloud, "async_active_subscription", return_value=True
@@ -279,163 +360,368 @@ async def test_step_pat_app_created_cloudhook(
await smartapp.setup_smartapp_endpoint(hass)
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- smartthings_mock.apps.return_value = []
- smartthings_mock.create_app.return_value = (app, app_oauth_client)
- smartthings_mock.locations.return_value = [location]
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert flow.access_token == token
- assert flow.app_id == app.app_id
- assert flow.oauth_client_secret == app_oauth_client.client_secret
- assert flow.oauth_client_id == app_oauth_client.client_id
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "select_location"
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
assert mock_create_cloudhook.call_count == 1
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
-async def test_step_pat_app_updated_webhook(
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "select_location"
+
+ # Select location and advance to external auth
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_LOCATION_ID: location.location_id}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
+ assert result["step_id"] == "authorize"
+ assert result["url"] == format_install_url(app.app_id, location.location_id)
+
+ # Complete external auth and advance to install
+ await smartapp.smartapp_install(hass, request, None, app)
+
+ # Finish
+ result = await hass.config_entries.flow.async_configure(result["flow_id"])
+ assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
+ assert result["data"]["app_id"] == app.app_id
+ assert result["data"]["installed_app_id"] == installed_app_id
+ assert result["data"]["location_id"] == location.location_id
+ assert result["data"]["access_token"] == token
+ assert result["data"]["refresh_token"] == request.refresh_token
+ assert result["data"]["client_secret"] == app_oauth_client.client_secret
+ assert result["data"]["client_id"] == app_oauth_client.client_id
+ assert result["title"] == location.name
+ entry = next(
+ (entry for entry in hass.config_entries.async_entries(DOMAIN)), None,
+ )
+ assert entry.unique_id == smartapp.format_unique_id(
+ app.app_id, location.location_id
+ )
+
+
+async def test_invalid_webhook_aborts(hass):
+ """Test flow aborts if webhook is invalid."""
+ hass.config.api.base_url = "http://0.0.0.0"
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
+ assert result["reason"] == "invalid_webhook_url"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_invalid_token_shows_error(hass):
+ """Test an error is shown for invalid token formats."""
+ token = "123456789"
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {CONF_ACCESS_TOKEN: "token_invalid_format"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_unauthorized_token_shows_error(hass, smartthings_mock):
+ """Test an error is shown for unauthorized token formats."""
+ token = str(uuid4())
+ request_info = Mock(real_url="http://example.com")
+ smartthings_mock.apps.side_effect = ClientResponseError(
+ request_info=request_info, history=None, status=HTTP_UNAUTHORIZED
+ )
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {CONF_ACCESS_TOKEN: "token_unauthorized"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_forbidden_token_shows_error(hass, smartthings_mock):
+ """Test an error is shown for forbidden token formats."""
+ token = str(uuid4())
+ request_info = Mock(real_url="http://example.com")
+ smartthings_mock.apps.side_effect = ClientResponseError(
+ request_info=request_info, history=None, status=HTTP_FORBIDDEN
+ )
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {CONF_ACCESS_TOKEN: "token_forbidden"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_webhook_problem_shows_error(hass, smartthings_mock):
+ """Test an error is shown when there's an problem with the webhook endpoint."""
+ token = str(uuid4())
+ data = {"error": {}}
+ request_info = Mock(real_url="http://example.com")
+ error = APIResponseError(
+ request_info=request_info, history=None, data=data, status=422
+ )
+ error.is_target_error = Mock(return_value=True)
+ smartthings_mock.apps.side_effect = error
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {"base": "webhook_error"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_api_error_shows_error(hass, smartthings_mock):
+ """Test an error is shown when other API errors occur."""
+ token = str(uuid4())
+ data = {"error": {}}
+ request_info = Mock(real_url="http://example.com")
+ error = APIResponseError(
+ request_info=request_info, history=None, data=data, status=400
+ )
+ smartthings_mock.apps.side_effect = error
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {"base": "app_setup_error"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_unknown_response_error_shows_error(hass, smartthings_mock):
+ """Test an error is shown when there is an unknown API error."""
+ token = str(uuid4())
+ request_info = Mock(real_url="http://example.com")
+ error = ClientResponseError(
+ request_info=request_info, history=None, status=HTTP_NOT_FOUND
+ )
+ smartthings_mock.apps.side_effect = error
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {"base": "app_setup_error"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_unknown_error_shows_error(hass, smartthings_mock):
+ """Test an error is shown when there is an unknown API error."""
+ token = str(uuid4())
+ smartthings_mock.apps.side_effect = Exception("Unknown error")
+
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert result["data_schema"]({}) == {CONF_ACCESS_TOKEN: token}
+ assert result["errors"] == {"base": "app_setup_error"}
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+
+async def test_no_available_locations_aborts(
hass, app, app_oauth_client, location, smartthings_mock
):
- """Test SmartApp is updated then show location form."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- smartthings_mock.apps.return_value = [app]
- smartthings_mock.generate_app_oauth.return_value = app_oauth_client
- smartthings_mock.locations.return_value = [location]
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert flow.access_token == token
- assert flow.app_id == app.app_id
- assert flow.oauth_client_secret == app_oauth_client.client_secret
- assert flow.oauth_client_id == app_oauth_client.client_id
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "select_location"
-
-
-async def test_step_pat_app_updated_webhook_from_existing_oauth_client(
- hass, app, location, smartthings_mock
-):
- """Test SmartApp is updated from existing then show location form."""
- oauth_client_id = str(uuid4())
- oauth_client_secret = str(uuid4())
- entry = MockConfigEntry(
- domain=DOMAIN,
- data={
- CONF_APP_ID: app.app_id,
- CONF_OAUTH_CLIENT_ID: oauth_client_id,
- CONF_OAUTH_CLIENT_SECRET: oauth_client_secret,
- CONF_LOCATION_ID: str(uuid4()),
- },
- )
- entry.add_to_hass(hass)
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- smartthings_mock.apps.return_value = [app]
- smartthings_mock.locations.return_value = [location]
- token = str(uuid4())
-
- result = await flow.async_step_pat({CONF_ACCESS_TOKEN: token})
-
- assert flow.access_token == token
- assert flow.app_id == app.app_id
- assert flow.oauth_client_secret == oauth_client_secret
- assert flow.oauth_client_id == oauth_client_id
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "select_location"
-
-
-async def test_step_select_location(hass, location, smartthings_mock):
- """Test select location shows form with available locations."""
- smartthings_mock.locations.return_value = [location]
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- flow.api = smartthings_mock
-
- result = await flow.async_step_select_location()
- assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
- assert result["step_id"] == "select_location"
- assert result["data_schema"]({CONF_LOCATION_ID: location.location_id}) == {
- CONF_LOCATION_ID: location.location_id
- }
-
-
-async def test_step_select_location_aborts(hass, location, smartthings_mock):
"""Test select location aborts if no available locations."""
+ token = str(uuid4())
+ smartthings_mock.apps.return_value = []
+ smartthings_mock.create_app.return_value = (app, app_oauth_client)
smartthings_mock.locations.return_value = [location]
entry = MockConfigEntry(
domain=DOMAIN, data={CONF_LOCATION_ID: location.location_id}
)
entry.add_to_hass(hass)
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- flow.api = smartthings_mock
- result = await flow.async_step_select_location()
+ # Webhook confirmation shown
+ result = await hass.config_entries.flow.async_init(
+ DOMAIN, context={"source": "user"}
+ )
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "user"
+ assert result["description_placeholders"][
+ "webhook_url"
+ ] == smartapp.get_webhook_url(hass)
+
+ # Advance to PAT screen
+ result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+ assert result["type"] == data_entry_flow.RESULT_TYPE_FORM
+ assert result["step_id"] == "pat"
+ assert "token_url" in result["description_placeholders"]
+ assert "component_url" in result["description_placeholders"]
+
+ # Enter token and advance to location screen
+ result = await hass.config_entries.flow.async_configure(
+ result["flow_id"], {CONF_ACCESS_TOKEN: token}
+ )
assert result["type"] == data_entry_flow.RESULT_TYPE_ABORT
assert result["reason"] == "no_available_locations"
-
-
-async def test_step_select_location_advances(hass):
- """Test select location aborts if no available locations."""
- location_id = str(uuid4())
- app_id = str(uuid4())
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- flow.app_id = app_id
-
- result = await flow.async_step_select_location({CONF_LOCATION_ID: location_id})
-
- assert flow.location_id == location_id
- assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP
- assert result["step_id"] == "authorize"
- assert result["url"] == format_install_url(app_id, location_id)
-
-
-async def test_step_authorize_advances(hass):
- """Test authorize step advances when completed."""
- installed_app_id = str(uuid4())
- refresh_token = str(uuid4())
- flow = SmartThingsFlowHandler()
- flow.hass = hass
-
- result = await flow.async_step_authorize(
- {CONF_INSTALLED_APP_ID: installed_app_id, CONF_REFRESH_TOKEN: refresh_token}
- )
-
- assert flow.installed_app_id == installed_app_id
- assert flow.refresh_token == refresh_token
- assert result["type"] == data_entry_flow.RESULT_TYPE_EXTERNAL_STEP_DONE
- assert result["step_id"] == "install"
-
-
-async def test_step_install_creates_entry(hass, location, smartthings_mock):
- """Test a config entry is created once the app is installed."""
- flow = SmartThingsFlowHandler()
- flow.hass = hass
- flow.api = smartthings_mock
- flow.access_token = str(uuid4())
- flow.app_id = str(uuid4())
- flow.installed_app_id = str(uuid4())
- flow.location_id = location.location_id
- flow.oauth_client_id = str(uuid4())
- flow.oauth_client_secret = str(uuid4())
- flow.refresh_token = str(uuid4())
-
- result = await flow.async_step_install()
-
- assert result["type"] == data_entry_flow.RESULT_TYPE_CREATE_ENTRY
- assert result["data"]["app_id"] == flow.app_id
- assert result["data"]["installed_app_id"] == flow.installed_app_id
- assert result["data"]["location_id"] == flow.location_id
- assert result["data"]["access_token"] == flow.access_token
- assert result["data"]["refresh_token"] == flow.refresh_token
- assert result["data"]["client_secret"] == flow.oauth_client_secret
- assert result["data"]["client_id"] == flow.oauth_client_id
- assert result["title"] == location.name
diff --git a/tests/components/smartthings/test_smartapp.py b/tests/components/smartthings/test_smartapp.py
index 4d7280a6a9e..efc4844cef2 100644
--- a/tests/components/smartthings/test_smartapp.py
+++ b/tests/components/smartthings/test_smartapp.py
@@ -6,8 +6,6 @@ from pysmartthings import AppEntity, Capability
from homeassistant.components.smartthings import smartapp
from homeassistant.components.smartthings.const import (
- CONF_INSTALLED_APP_ID,
- CONF_LOCATION_ID,
CONF_REFRESH_TOKEN,
DATA_MANAGER,
DOMAIN,
@@ -39,36 +37,6 @@ async def test_update_app_updated_needed(hass, app):
assert mock_app.classifications == app.classifications
-async def test_smartapp_install_configures_flow(hass):
- """Test install event continues an existing flow."""
- # Arrange
- flow_id = str(uuid4())
- flows = [{"flow_id": flow_id, "handler": DOMAIN}]
- app = Mock()
- app.app_id = uuid4()
- request = Mock()
- request.installed_app_id = str(uuid4())
- request.auth_token = str(uuid4())
- request.location_id = str(uuid4())
- request.refresh_token = str(uuid4())
-
- # Act
- with patch.object(
- hass.config_entries.flow, "async_progress", return_value=flows
- ), patch.object(hass.config_entries.flow, "async_configure") as configure_mock:
-
- await smartapp.smartapp_install(hass, request, None, app)
-
- configure_mock.assert_called_once_with(
- flow_id,
- {
- CONF_INSTALLED_APP_ID: request.installed_app_id,
- CONF_LOCATION_ID: request.location_id,
- CONF_REFRESH_TOKEN: request.refresh_token,
- },
- )
-
-
async def test_smartapp_update_saves_token(
hass, smartthings_mock, location, device_factory
):
@@ -92,36 +60,6 @@ async def test_smartapp_update_saves_token(
assert entry.data[CONF_REFRESH_TOKEN] == request.refresh_token
-async def test_smartapp_update_configures_flow(hass):
- """Test update event continues an existing flow."""
- # Arrange
- flow_id = str(uuid4())
- flows = [{"flow_id": flow_id, "handler": DOMAIN}]
- app = Mock()
- app.app_id = uuid4()
- request = Mock()
- request.installed_app_id = str(uuid4())
- request.auth_token = str(uuid4())
- request.location_id = str(uuid4())
- request.refresh_token = str(uuid4())
-
- # Act
- with patch.object(
- hass.config_entries.flow, "async_progress", return_value=flows
- ), patch.object(hass.config_entries.flow, "async_configure") as configure_mock:
-
- await smartapp.smartapp_update(hass, request, None, app)
-
- configure_mock.assert_called_once_with(
- flow_id,
- {
- CONF_INSTALLED_APP_ID: request.installed_app_id,
- CONF_LOCATION_ID: request.location_id,
- CONF_REFRESH_TOKEN: request.refresh_token,
- },
- )
-
-
async def test_smartapp_uninstall(hass, config_entry):
"""Test the config entry is unloaded when the app is uninstalled."""
config_entry.add_to_hass(hass)