1
0
mirror of https://github.com/home-assistant/core.git synced 2026-05-08 09:38:58 +01:00

Pull out Dropbox integration (#166986)

This commit is contained in:
Joost Lekkerkerker
2026-03-31 20:40:04 +02:00
committed by Franck Nijhof
parent 12dc33eabc
commit 35826dfd14
22 changed files with 0 additions and 1644 deletions
-1
View File
@@ -1 +0,0 @@
"""Tests for the Dropbox integration."""
-114
View File
@@ -1,114 +0,0 @@
"""Shared fixtures for Dropbox integration tests."""
from __future__ import annotations
from collections.abc import Generator
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from homeassistant.components.application_credentials import (
ClientCredential,
async_import_client_credential,
)
from homeassistant.components.dropbox.const import DOMAIN, OAUTH2_SCOPES
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from tests.common import MockConfigEntry
CLIENT_ID = "1234"
CLIENT_SECRET = "5678"
ACCOUNT_ID = "dbid:1234567890abcdef"
ACCOUNT_EMAIL = "user@example.com"
CONFIG_ENTRY_TITLE = "Dropbox test account"
TEST_AGENT_ID = f"{DOMAIN}.{ACCOUNT_ID}"
@pytest.fixture(autouse=True)
async def setup_credentials(hass: HomeAssistant) -> None:
"""Set up application credentials for Dropbox."""
assert await async_setup_component(hass, "application_credentials", {})
await async_import_client_credential(
hass,
DOMAIN,
ClientCredential(CLIENT_ID, CLIENT_SECRET),
)
@pytest.fixture
def account_info() -> SimpleNamespace:
"""Return mocked Dropbox account information."""
return SimpleNamespace(account_id=ACCOUNT_ID, email=ACCOUNT_EMAIL)
@pytest.fixture
def mock_config_entry() -> MockConfigEntry:
"""Return a default Dropbox config entry."""
return MockConfigEntry(
domain=DOMAIN,
unique_id=ACCOUNT_ID,
title=CONFIG_ENTRY_TITLE,
data={
"auth_implementation": DOMAIN,
"token": {
"access_token": "mock-access-token",
"refresh_token": "mock-refresh-token",
"expires_at": 9_999_999_999,
"scope": " ".join(OAUTH2_SCOPES),
},
},
)
@pytest.fixture
def mock_setup_entry() -> Generator[AsyncMock]:
"""Override async_setup_entry."""
with patch(
"homeassistant.components.dropbox.async_setup_entry", return_value=True
) as mock_setup_entry:
yield mock_setup_entry
@pytest.fixture
def mock_dropbox_client(account_info: SimpleNamespace) -> Generator[MagicMock]:
"""Patch DropboxAPIClient to exercise auth while mocking API calls."""
client = MagicMock()
client.list_folder = AsyncMock(return_value=[])
client.download_file = MagicMock()
client.upload_file = AsyncMock()
client.delete_file = AsyncMock()
captured_auth = None
def capture_auth(auth):
nonlocal captured_auth
captured_auth = auth
return client
async def get_account_info_with_auth():
await captured_auth.async_get_access_token()
return client.get_account_info.return_value
client.get_account_info = AsyncMock(
side_effect=get_account_info_with_auth,
return_value=account_info,
)
with (
patch(
"homeassistant.components.dropbox.config_flow.DropboxAPIClient",
side_effect=capture_auth,
),
patch(
"homeassistant.components.dropbox.DropboxAPIClient",
side_effect=capture_auth,
),
):
yield client
-577
View File
@@ -1,577 +0,0 @@
"""Test the Dropbox backup platform."""
from __future__ import annotations
from collections.abc import AsyncIterator
from io import StringIO
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch
import pytest
from python_dropbox_api import DropboxAuthException
from homeassistant.components.backup import (
DOMAIN as BACKUP_DOMAIN,
AddonInfo,
AgentBackup,
suggested_filename,
)
from homeassistant.components.dropbox.backup import (
DropboxFileOrFolderNotFoundException,
DropboxUnknownException,
async_register_backup_agents_listener,
)
from homeassistant.components.dropbox.const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
from homeassistant.config_entries import SOURCE_REAUTH
from homeassistant.core import HomeAssistant
from homeassistant.setup import async_setup_component
from .conftest import CONFIG_ENTRY_TITLE, TEST_AGENT_ID
from tests.common import MockConfigEntry
from tests.test_util.aiohttp import mock_stream
from tests.typing import ClientSessionGenerator, WebSocketGenerator
TEST_AGENT_BACKUP = AgentBackup(
addons=[AddonInfo(name="Test", slug="test", version="1.0.0")],
backup_id="dropbox-backup",
database_included=True,
date="2025-01-01T00:00:00.000Z",
extra_metadata={"with_automatic_settings": False},
folders=[],
homeassistant_included=True,
homeassistant_version="2024.12.0",
name="Dropbox backup",
protected=False,
size=2048,
)
TEST_AGENT_BACKUP_RESULT = {
"addons": [{"name": "Test", "slug": "test", "version": "1.0.0"}],
"agents": {TEST_AGENT_ID: {"protected": False, "size": 2048}},
"backup_id": TEST_AGENT_BACKUP.backup_id,
"database_included": True,
"date": TEST_AGENT_BACKUP.date,
"extra_metadata": {"with_automatic_settings": False},
"failed_addons": [],
"failed_agent_ids": [],
"failed_folders": [],
"folders": [],
"homeassistant_included": True,
"homeassistant_version": TEST_AGENT_BACKUP.homeassistant_version,
"name": TEST_AGENT_BACKUP.name,
"with_automatic_settings": None,
}
def _suggested_filenames(backup: AgentBackup) -> tuple[str, str]:
"""Return the suggested filenames for the backup and metadata."""
base_name = suggested_filename(backup).rsplit(".", 1)[0]
return f"{base_name}.tar", f"{base_name}.metadata.json"
async def _mock_metadata_stream(backup: AgentBackup) -> AsyncIterator[bytes]:
"""Create a mock metadata download stream."""
yield json.dumps(backup.as_dict()).encode()
def _setup_list_folder_with_backup(
mock_dropbox_client: Mock,
backup: AgentBackup,
) -> None:
"""Set up mock to return a backup in list_folder and download_file."""
tar_name, metadata_name = _suggested_filenames(backup)
mock_dropbox_client.list_folder = AsyncMock(
return_value=[
SimpleNamespace(name=tar_name),
SimpleNamespace(name=metadata_name),
]
)
mock_dropbox_client.download_file = Mock(return_value=_mock_metadata_stream(backup))
@pytest.fixture(autouse=True)
async def setup_integration(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_dropbox_client,
) -> None:
"""Set up the Dropbox and Backup integrations for testing."""
mock_config_entry.add_to_hass(hass)
assert await async_setup_component(hass, BACKUP_DOMAIN, {BACKUP_DOMAIN: {}})
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
mock_dropbox_client.reset_mock()
async def test_agents_info(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test listing available backup agents."""
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/agents/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agents": [
{"agent_id": "backup.local", "name": "local"},
{"agent_id": TEST_AGENT_ID, "name": CONFIG_ENTRY_TITLE},
]
}
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
await client.send_json_auto_id({"type": "backup/agents/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agents": [{"agent_id": "backup.local", "name": "local"}]
}
async def test_agents_list_backups(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test listing backups via the Dropbox agent."""
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
assert response["result"]["backups"] == [TEST_AGENT_BACKUP_RESULT]
mock_dropbox_client.list_folder.assert_awaited()
async def test_agents_list_backups_metadata_without_tar(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that orphaned metadata files are skipped with a warning."""
mock_dropbox_client.list_folder = AsyncMock(
return_value=[SimpleNamespace(name="orphan.metadata.json")]
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
assert response["result"]["backups"] == []
assert "without matching backup file" in caplog.text
async def test_agents_list_backups_invalid_metadata(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test that invalid metadata files are skipped with a warning."""
async def _invalid_stream() -> AsyncIterator[bytes]:
yield b"not valid json"
mock_dropbox_client.list_folder = AsyncMock(
return_value=[
SimpleNamespace(name="backup.tar"),
SimpleNamespace(name="backup.metadata.json"),
]
)
mock_dropbox_client.download_file = Mock(return_value=_invalid_stream())
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
assert response["result"]["backups"] == []
assert "Skipping invalid metadata file" in caplog.text
async def test_agents_list_backups_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test handling list backups failures."""
mock_dropbox_client.list_folder = AsyncMock(
side_effect=DropboxUnknownException("boom")
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["backups"] == []
assert response["result"]["agent_errors"] == {
TEST_AGENT_ID: "Failed to list backups"
}
async def test_agents_list_backups_reauth(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test reauthentication is triggered on auth error."""
mock_dropbox_client.list_folder = AsyncMock(
side_effect=DropboxAuthException("auth failed")
)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/info"})
response = await client.receive_json()
assert response["success"]
assert response["result"]["backups"] == []
assert response["result"]["agent_errors"] == {TEST_AGENT_ID: "Authentication error"}
await hass.async_block_till_done()
flows = hass.config_entries.flow.async_progress()
assert len(flows) == 1
flow = flows[0]
assert flow["step_id"] == "reauth_confirm"
assert flow["handler"] == DOMAIN
assert flow["context"]["source"] == SOURCE_REAUTH
assert flow["context"]["entry_id"] == mock_config_entry.entry_id
@pytest.mark.parametrize(
"backup_id",
[TEST_AGENT_BACKUP.backup_id, "other-backup"],
ids=["found", "not_found"],
)
async def test_agents_get_backup(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
backup_id: str,
) -> None:
"""Test retrieving a backup's metadata."""
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
client = await hass_ws_client(hass)
await client.send_json_auto_id({"type": "backup/details", "backup_id": backup_id})
response = await client.receive_json()
assert response["success"]
assert response["result"]["agent_errors"] == {}
if backup_id == TEST_AGENT_BACKUP.backup_id:
assert response["result"]["backup"] == TEST_AGENT_BACKUP_RESULT
else:
assert response["result"]["backup"] is None
async def test_agents_download(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test downloading a backup file."""
tar_name, metadata_name = _suggested_filenames(TEST_AGENT_BACKUP)
mock_dropbox_client.list_folder = AsyncMock(
return_value=[
SimpleNamespace(name=tar_name),
SimpleNamespace(name=metadata_name),
]
)
def download_side_effect(path: str) -> AsyncIterator[bytes]:
if path == f"/{tar_name}":
return mock_stream(b"backup data")
return _mock_metadata_stream(TEST_AGENT_BACKUP)
mock_dropbox_client.download_file = Mock(side_effect=download_side_effect)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 200
assert await resp.content.read() == b"backup data"
async def test_agents_download_fail(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test handling download failures."""
mock_dropbox_client.list_folder = AsyncMock(
side_effect=DropboxUnknownException("boom")
)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 500
body = await resp.content.read()
assert b"Failed to get backup" in body
async def test_agents_download_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test download when backup disappears between get and download."""
tar_name, metadata_name = _suggested_filenames(TEST_AGENT_BACKUP)
files = [
SimpleNamespace(name=tar_name),
SimpleNamespace(name=metadata_name),
]
# First list_folder call (async_get_backup) returns the backup;
# second call (async_download_backup) returns empty, simulating deletion.
mock_dropbox_client.list_folder = AsyncMock(side_effect=[files, []])
mock_dropbox_client.download_file = Mock(
return_value=_mock_metadata_stream(TEST_AGENT_BACKUP)
)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 404
assert await resp.content.read() == b""
async def test_agents_download_file_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test download when Dropbox file is not found returns 404."""
mock_dropbox_client.list_folder = AsyncMock(
side_effect=DropboxFileOrFolderNotFoundException("not found")
)
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 404
async def test_agents_download_metadata_not_found(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test download when metadata lookup fails."""
mock_dropbox_client.list_folder = AsyncMock(return_value=[])
client = await hass_client()
resp = await client.get(
f"/api/backup/download/{TEST_AGENT_BACKUP.backup_id}?agent_id={TEST_AGENT_ID}"
)
assert resp.status == 404
assert await resp.content.read() == b""
async def test_agents_upload(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_dropbox_client: Mock,
) -> None:
"""Test uploading a backup to Dropbox."""
mock_dropbox_client.upload_file = AsyncMock(return_value=None)
client = await hass_client()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
return_value=TEST_AGENT_BACKUP,
),
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_AGENT_BACKUP,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
resp = await client.post(
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
data={"file": StringIO("test")},
)
assert resp.status == 201
assert f"Uploading backup {TEST_AGENT_BACKUP.backup_id} to agents" in caplog.text
assert mock_dropbox_client.upload_file.await_count == 2
async def test_agents_upload_fail(
hass: HomeAssistant,
hass_client: ClientSessionGenerator,
caplog: pytest.LogCaptureFixture,
mock_dropbox_client: Mock,
) -> None:
"""Test that backup tar is cleaned up when metadata upload fails."""
call_count = 0
async def upload_side_effect(path: str, stream: AsyncIterator[bytes]) -> None:
nonlocal call_count
call_count += 1
async for _ in stream:
pass
if call_count == 2:
raise DropboxUnknownException("metadata upload failed")
mock_dropbox_client.upload_file = AsyncMock(side_effect=upload_side_effect)
mock_dropbox_client.delete_file = AsyncMock()
client = await hass_client()
with (
patch(
"homeassistant.components.backup.manager.BackupManager.async_get_backup",
return_value=TEST_AGENT_BACKUP,
),
patch(
"homeassistant.components.backup.manager.read_backup",
return_value=TEST_AGENT_BACKUP,
),
patch("pathlib.Path.open") as mocked_open,
):
mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
resp = await client.post(
f"/api/backup/upload?agent_id={TEST_AGENT_ID}",
data={"file": StringIO("test")},
)
await hass.async_block_till_done()
assert resp.status == 201
assert "Failed to upload backup" in caplog.text
mock_dropbox_client.delete_file.assert_awaited_once()
async def test_agents_delete(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test deleting a backup."""
_setup_list_folder_with_backup(mock_dropbox_client, TEST_AGENT_BACKUP)
mock_dropbox_client.delete_file = AsyncMock(return_value=None)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": TEST_AGENT_BACKUP.backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
assert mock_dropbox_client.delete_file.await_count == 2
async def test_agents_delete_fail(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test error handling when delete fails."""
mock_dropbox_client.list_folder = AsyncMock(
side_effect=DropboxUnknownException("boom")
)
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": TEST_AGENT_BACKUP.backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {
"agent_errors": {TEST_AGENT_ID: "Failed to delete backup"}
}
async def test_agents_delete_not_found(
hass: HomeAssistant,
hass_ws_client: WebSocketGenerator,
mock_dropbox_client: Mock,
) -> None:
"""Test deleting a backup that does not exist."""
mock_dropbox_client.list_folder = AsyncMock(return_value=[])
client = await hass_ws_client(hass)
await client.send_json_auto_id(
{
"type": "backup/delete",
"backup_id": TEST_AGENT_BACKUP.backup_id,
}
)
response = await client.receive_json()
assert response["success"]
assert response["result"] == {"agent_errors": {}}
async def test_remove_backup_agents_listener(
hass: HomeAssistant,
) -> None:
"""Test removing a backup agent listener."""
listener = Mock()
remove = async_register_backup_agents_listener(hass, listener=listener)
assert DATA_BACKUP_AGENT_LISTENERS in hass.data
assert listener in hass.data[DATA_BACKUP_AGENT_LISTENERS]
# Remove all other listeners to test the cleanup path
hass.data[DATA_BACKUP_AGENT_LISTENERS] = [listener]
remove()
assert DATA_BACKUP_AGENT_LISTENERS not in hass.data
@@ -1,210 +0,0 @@
"""Test the Dropbox config flow."""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from yarl import URL
from homeassistant import config_entries
from homeassistant.components.dropbox.const import (
DOMAIN,
OAUTH2_AUTHORIZE,
OAUTH2_SCOPES,
OAUTH2_TOKEN,
)
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers import config_entry_oauth2_flow
from .conftest import ACCOUNT_EMAIL, ACCOUNT_ID, CLIENT_ID
from tests.test_util.aiohttp import AiohttpClientMocker
from tests.typing import ClientSessionGenerator
@pytest.mark.usefixtures("current_request_with_host")
async def test_full_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_dropbox_client,
mock_setup_entry: AsyncMock,
) -> None:
"""Test creating a new config entry through the OAuth flow."""
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
result_url = URL(result["url"])
assert f"{result_url.origin()}{result_url.path}" == OAUTH2_AUTHORIZE
assert result_url.query["response_type"] == "code"
assert result_url.query["client_id"] == CLIENT_ID
assert (
result_url.query["redirect_uri"] == "https://example.com/auth/external/callback"
)
assert result_url.query["state"] == state
assert result_url.query["scope"] == " ".join(OAUTH2_SCOPES)
assert result_url.query["token_access_type"] == "offline"
assert result_url.query["code_challenge"]
assert result_url.query["code_challenge_method"] == "S256"
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
assert resp.headers["content-type"] == "text/html; charset=utf-8"
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"token_type": "Bearer",
"expires_in": 60,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] is FlowResultType.CREATE_ENTRY
assert result["title"] == ACCOUNT_EMAIL
assert result["data"]["token"]["access_token"] == "mock-access-token"
assert result["result"].unique_id == ACCOUNT_ID
assert len(mock_setup_entry.mock_calls) == 1
assert len(hass.config_entries.async_entries(DOMAIN)) == 1
@pytest.mark.usefixtures("current_request_with_host")
async def test_already_configured(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry,
mock_dropbox_client,
) -> None:
"""Test aborting when the account is already configured."""
mock_config_entry.add_to_hass(hass)
result = await hass.config_entries.flow.async_init(
DOMAIN, context={"source": config_entries.SOURCE_USER}
)
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "mock-access-token",
"token_type": "Bearer",
"expires_in": 60,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == "already_configured"
@pytest.mark.usefixtures("current_request_with_host")
@pytest.mark.parametrize(
(
"new_account_info",
"expected_reason",
"expected_setup_calls",
"expected_access_token",
),
[
(
SimpleNamespace(account_id=ACCOUNT_ID, email=ACCOUNT_EMAIL),
"reauth_successful",
1,
"updated-access-token",
),
(
SimpleNamespace(account_id="dbid:different", email="other@example.com"),
"wrong_account",
0,
"mock-access-token",
),
],
ids=["success", "wrong_account"],
)
async def test_reauth_flow(
hass: HomeAssistant,
hass_client_no_auth: ClientSessionGenerator,
aioclient_mock: AiohttpClientMocker,
mock_config_entry,
mock_dropbox_client,
mock_setup_entry: AsyncMock,
new_account_info: SimpleNamespace,
expected_reason: str,
expected_setup_calls: int,
expected_access_token: str,
) -> None:
"""Test reauthentication flow outcomes."""
mock_config_entry.add_to_hass(hass)
mock_dropbox_client.get_account_info.return_value = new_account_info
result = await mock_config_entry.start_reauth_flow(hass)
assert result["type"] is FlowResultType.FORM
assert result["step_id"] == "reauth_confirm"
result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
state = config_entry_oauth2_flow._encode_jwt(
hass,
{
"flow_id": result["flow_id"],
"redirect_uri": "https://example.com/auth/external/callback",
},
)
client = await hass_client_no_auth()
resp = await client.get(f"/auth/external/callback?code=abcd&state={state}")
assert resp.status == 200
aioclient_mock.post(
OAUTH2_TOKEN,
json={
"refresh_token": "mock-refresh-token",
"access_token": "updated-access-token",
"token_type": "Bearer",
"expires_in": 120,
},
)
result = await hass.config_entries.flow.async_configure(result["flow_id"])
await hass.async_block_till_done()
assert result["type"] is FlowResultType.ABORT
assert result["reason"] == expected_reason
assert mock_setup_entry.await_count == expected_setup_calls
assert mock_config_entry.data["token"]["access_token"] == expected_access_token
-100
View File
@@ -1,100 +0,0 @@
"""Test the Dropbox integration setup."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
import pytest
from python_dropbox_api import DropboxAuthException, DropboxUnknownException
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
from homeassistant.helpers.config_entry_oauth2_flow import (
ImplementationUnavailableError,
)
from tests.common import MockConfigEntry
@pytest.mark.usefixtures("mock_dropbox_client")
async def test_setup_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test successful setup of a config entry."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
async def test_setup_entry_auth_failed(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_dropbox_client: AsyncMock,
) -> None:
"""Test setup failure when authentication fails."""
mock_dropbox_client.get_account_info.side_effect = DropboxAuthException(
"Invalid token"
)
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_ERROR
@pytest.mark.parametrize(
"side_effect",
[DropboxUnknownException("Unknown error"), TimeoutError("Connection timed out")],
ids=["unknown_exception", "timeout_error"],
)
async def test_setup_entry_not_ready(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_dropbox_client: AsyncMock,
side_effect: Exception,
) -> None:
"""Test setup retry when the service is temporarily unavailable."""
mock_dropbox_client.get_account_info.side_effect = side_effect
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
async def test_setup_entry_implementation_unavailable(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test setup retry when OAuth implementation is unavailable."""
mock_config_entry.add_to_hass(hass)
with patch(
"homeassistant.components.dropbox.async_get_config_entry_implementation",
side_effect=ImplementationUnavailableError,
):
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.usefixtures("mock_dropbox_client")
async def test_unload_entry(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test unloading a config entry."""
mock_config_entry.add_to_hass(hass)
await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
await hass.config_entries.async_unload(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED