Add Backblaze B2 integration for backups

CODEOWNERS

@@ -180,6 +180,8 @@ build.json @home-assistant/supervisor
 /homeassistant/components/azure_event_hub/ @eavanvalkenburg
 /tests/components/azure_event_hub/ @eavanvalkenburg
 /homeassistant/components/azure_service_bus/ @hfurubotten
+/homeassistant/components/backblaze/ @frenck
+/tests/components/backblaze/ @frenck
 /homeassistant/components/backup/ @home-assistant/core
 /tests/components/backup/ @home-assistant/core
 /homeassistant/components/baf/ @bdraco @jfroy

homeassistant/components/backblaze/__init__.py

@@ -0,0 +1,72 @@
+"""Integration for Backblaze B2 Cloud Storage."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from b2sdk.v2 import AuthInfoCache, B2Api, Bucket, InMemoryAccountInfo
+from b2sdk.v2.exception import InvalidAuthToken, NonExistentBucket
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
+
+from .const import (
+    CONF_APPLICATION_KEY,
+    CONF_APPLICATION_KEY_ID,
+    CONF_BUCKET,
+    DATA_BACKUP_AGENT_LISTENERS,
+)
+
+type BackblazeConfigEntry = ConfigEntry[BackblazeonfigEntryData]
+
+
+@dataclass(kw_only=True)
+class BackblazeonfigEntryData:
+    """Dataclass holding all config entry data for a Backblaze entry."""
+
+    api: B2Api
+    bucket: Bucket
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: BackblazeConfigEntry) -> bool:
+    """Set up Backblaze from a config entry."""
+
+    info = InMemoryAccountInfo()
+    backblaze = B2Api(info, cache=AuthInfoCache(info))
+    try:
+        await hass.async_add_executor_job(
+            backblaze.authorize_account,
+            "production",
+            entry.data[CONF_APPLICATION_KEY_ID],
+            entry.data[CONF_APPLICATION_KEY],
+        )
+        bucket = await hass.async_add_executor_job(
+            backblaze.get_bucket_by_id, entry.data[CONF_BUCKET]
+        )
+    except InvalidAuthToken as err:
+        raise ConfigEntryAuthFailed(
+            f"Invalid authentication token for Backblaze account: {err}"
+        ) from err
+    except NonExistentBucket as err:
+        raise ConfigEntryNotReady(
+            f"Non-existent bucket for Backblaze account: {err}"
+        ) from err
+
+    entry.runtime_data = BackblazeonfigEntryData(api=backblaze, bucket=bucket)
+
+    # Notify backup listeners
+    hass.async_create_task(_notify_backup_listeners(hass), eager_start=False)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: BackblazeConfigEntry) -> bool:
+    """Unload Backblaze config entry."""
+    hass.async_create_task(_notify_backup_listeners(hass), eager_start=False)
+    return True
+
+
+async def _notify_backup_listeners(hass: HomeAssistant) -> None:
+    for listener in hass.data.get(DATA_BACKUP_AGENT_LISTENERS, []):
+        listener()

homeassistant/components/backblaze/backup.py

@@ -0,0 +1,228 @@
+"""Backup platform for the Backblaze integration."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Coroutine
+from typing import Any
+
+from b2sdk.v2.exception import B2Error
+
+from homeassistant.components.backup import (
+    AddonInfo,
+    AgentBackup,
+    BackupAgent,
+    BackupAgentError,
+    Folder,
+)
+from homeassistant.core import HomeAssistant, callback
+
+from . import BackblazeConfigEntry
+from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN, SEPARATOR
+
+
+async def async_get_backup_agents(
+    hass: HomeAssistant,
+) -> list[BackupAgent]:
+    """Register the backup agents."""
+    entries: list[BackblazeConfigEntry] = hass.config_entries.async_entries(DOMAIN)
+    return [BackblazeBackupAgent(hass, entry) for entry in entries]
+
+
+@callback
+def async_register_backup_agents_listener(
+    hass: HomeAssistant,
+    *,
+    listener: Callable[[], None],
+    **kwargs: Any,
+) -> Callable[[], None]:
+    """Register a listener to be called when agents are added or removed."""
+    hass.data.setdefault(DATA_BACKUP_AGENT_LISTENERS, []).append(listener)
+
+    @callback
+    def remove_listener() -> None:
+        """Remove the listener."""
+        hass.data[DATA_BACKUP_AGENT_LISTENERS].remove(listener)
+
+    return remove_listener
+
+
+class BackblazeBackupAgent(BackupAgent):
+    """Backblaze backup agent."""
+
+    domain = DOMAIN
+
+    def __init__(self, hass: HomeAssistant, entry: BackblazeConfigEntry) -> None:
+        """Initialize the Backblaze backup sync agent."""
+        super().__init__()
+        self._bucket = entry.runtime_data.bucket
+        self._api = entry.runtime_data.api
+        self._hass = hass
+        self.name = entry.title
+
+    async def async_download_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> AsyncIterator[bytes]:
+        """Download a backup file from Backblaze."""
+        if not await self.async_get_backup(backup_id):
+            raise BackupAgentError("Backup not found")
+
+        try:
+            downloaded_file = await self._hass.async_add_executor_job(
+                self._bucket.download_file_by_name, f"{backup_id}.tar"
+            )
+        except B2Error as err:
+            raise BackupAgentError(
+                f"Failed to download backup {backup_id}: {err}"
+            ) from err
+
+        if not downloaded_file.response.ok:
+            raise BackupAgentError(
+                f"Failed to download backup {backup_id}: HTTP {downloaded_file.response.status_code}"
+            )
+
+        # Use an executor to avoid blocking the event loop
+        for chunk in await self._hass.async_add_executor_job(
+            downloaded_file.response.iter_content, 1024
+        ):
+            yield chunk
+
+    async def async_upload_backup(
+        self,
+        *,
+        open_stream: Callable[[], Coroutine[Any, Any, AsyncIterator[bytes]]],
+        backup: AgentBackup,
+        **kwargs: Any,
+    ) -> None:
+        """Upload a backup."""
+
+        # Prepare file info metadata to store with the backup in Backblaze
+        file_info = {
+            "backup_id": backup.backup_id,
+            "database_included": str(backup.database_included).lower(),
+            "date": backup.date,
+            "extra_metadata": "###META###".join(
+                f"{key}{SEPARATOR}{val}" for key, val in backup.extra_metadata.items()
+            ),
+            "homeassistant_included": str(backup.homeassistant_included).lower(),
+            "homeassistant_version": backup.homeassistant_version,
+            "name": backup.name,
+            "protected": str(backup.protected).lower(),
+            "size": str(backup.size),
+        }
+        if backup.addons:
+            file_info["addons"] = "###ADDON###".join(
+                f"{addon.slug}{SEPARATOR}{addon.version}{SEPARATOR}{addon.name}"
+                for addon in backup.addons
+            )
+        if backup.folders:
+            file_info["folders"] = ",".join(folder.value for folder in backup.folders)
+
+        stream: AsyncIterator[bytes] = await open_stream()
+
+        try:
+            await self._hass.async_add_executor_job(
+                self._bucket.upload_bytes,
+                b"".join([chunk async for chunk in stream]),
+                f"{backup.backup_id}.tar",
+                "application/octet-stream",
+                file_info,
+            )
+        except B2Error as err:
+            raise BackupAgentError(
+                f"Failed to upload backup {backup.backup_id}: {err}"
+            ) from err
+
+    def _delete_backup(
+        self,
+        backup_id: str,
+    ) -> None:
+        """Delete file from Backblaze."""
+        try:
+            file_info = self._bucket.get_file_info_by_name(f"{backup_id}.tar")
+            self._api.delete_file_version(
+                file_info.id_,
+                file_info.file_name,
+            )
+        except B2Error as err:
+            raise BackupAgentError(
+                f"Failed to delete backup {backup_id}: {err}"
+            ) from err
+
+    async def async_delete_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> None:
+        """Delete a backup file from Backblaze."""
+        if not await self.async_get_backup(backup_id):
+            return
+
+        await self._hass.async_add_executor_job(self._delete_backup, backup_id)
+
+    def _list_backups(self) -> list[AgentBackup]:
+        """List backups stored on Backblaze."""
+        backups = []
+        try:
+            for file_version, _ in self._bucket.ls(latest_only=True):
+                file_info = file_version.file_info
+
+                if "homeassistant_version" not in file_info:
+                    continue
+
+                addons: list[AddonInfo] = []
+                if addons_string := file_version.file_info.get("addons"):
+                    for addon in addons_string.split("###ADDON###"):
+                        slug, version, name = addon.split(SEPARATOR)
+                        addons.append(AddonInfo(slug=slug, version=version, name=name))
+
+                extra_metadata = {}
+                if extra_metadata_string := file_info.get("extra_metadata"):
+                    for meta in extra_metadata_string.split("###META###"):
+                        key, val = meta.split(SEPARATOR)
+                        extra_metadata[key] = val
+
+                folders: list[Folder] = []
+                if folder_string := file_version.file_info.get("folders"):
+                    folders = [
+                        Folder(folder) for folder in folder_string.split(SEPARATOR)
+                    ]
+
+                backups.append(
+                    AgentBackup(
+                        backup_id=file_info["backup_id"],
+                        name=file_info["name"],
+                        date=file_info["date"],
+                        size=int(file_info["size"]),
+                        homeassistant_version=file_info["homeassistant_version"],
+                        protected=file_info["protected"] == "true",
+                        addons=addons,
+                        folders=folders,
+                        database_included=file_info["database_included"] == "true",
+                        homeassistant_included=file_info["database_included"] == "true",
+                        extra_metadata=extra_metadata,
+                    )
+                )
+        except B2Error as err:
+            raise BackupAgentError(f"Failed to list backups: {err}") from err
+
+        return backups
+
+    async def async_list_backups(self, **kwargs: Any) -> list[AgentBackup]:
+        """List backups stored on Backblaze."""
+        return await self._hass.async_add_executor_job(self._list_backups)
+
+    async def async_get_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> AgentBackup | None:
+        """Return a backup."""
+        backups = await self.async_list_backups()
+
+        for backup in backups:
+            if backup.backup_id == backup_id:
+                return backup
+
+        return None

homeassistant/components/backblaze/config_flow.py

@@ -0,0 +1,132 @@
+"""Config flow to configure the Backblaze B2 Cloud Storage integration."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+from b2sdk.v2 import AuthInfoCache, B2Api, Bucket, InMemoryAccountInfo
+from b2sdk.v2.exception import InvalidAuthToken
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers.selector import (
+    SelectOptionDict,
+    SelectSelector,
+    SelectSelectorConfig,
+    SelectSelectorMode,
+    TextSelector,
+    TextSelectorConfig,
+    TextSelectorType,
+)
+
+from .const import (
+    CONF_APPLICATION_KEY,
+    CONF_APPLICATION_KEY_ID,
+    CONF_BUCKET,
+    DOMAIN,
+    LOGGER,
+)
+
+
+class BackblazeFlowHandler(ConfigFlow, domain=DOMAIN):
+    """Handle a Backblaze config flow."""
+
+    VERSION = 1
+
+    _buckets: Sequence[Bucket]
+    _authorization: dict[str, Any]
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a flow initiated by the user."""
+        errors = {}
+
+        if user_input is not None:
+            info = InMemoryAccountInfo()
+            backblaze = B2Api(info, cache=AuthInfoCache(info))
+            try:
+                await self.hass.async_add_executor_job(
+                    backblaze.authorize_account,
+                    "production",
+                    user_input[CONF_APPLICATION_KEY_ID],
+                    user_input[CONF_APPLICATION_KEY],
+                )
+                self._buckets = await self.hass.async_add_executor_job(
+                    backblaze.list_buckets,
+                )
+            except InvalidAuthToken:
+                errors["base"] = "invalid_auth"
+            except Exception:  # noqa: BLE001
+                LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                self._authorization = user_input
+                return await self.async_step_bucket()
+        else:
+            user_input = {}
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_APPLICATION_KEY_ID,
+                        default=user_input.get(CONF_APPLICATION_KEY_ID),
+                    ): TextSelector(
+                        config=TextSelectorConfig(
+                            autocomplete="off",
+                        ),
+                    ),
+                    vol.Required(CONF_APPLICATION_KEY): TextSelector(
+                        config=TextSelectorConfig(
+                            type=TextSelectorType.PASSWORD,
+                        ),
+                    ),
+                }
+            ),
+            errors=errors,
+        )
+
+    async def async_step_bucket(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle a bucket selection."""
+        if user_input is not None:
+            await self.async_set_unique_id(user_input[CONF_BUCKET])
+            self._abort_if_unique_id_configured()
+            return self.async_create_entry(
+                title=(
+                    next(
+                        bucket.name
+                        for bucket in self._buckets
+                        if bucket.id_ == user_input[CONF_BUCKET]
+                    )
+                    or "Backblaze"
+                ),
+                data=self._authorization | user_input,
+            )
+
+        return self.async_show_form(
+            step_id="bucket",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(
+                        CONF_BUCKET,
+                    ): SelectSelector(
+                        config=SelectSelectorConfig(
+                            mode=SelectSelectorMode.DROPDOWN,
+                            options=[
+                                SelectOptionDict(
+                                    value=bucket.id_,
+                                    label=bucket.name,
+                                )
+                                for bucket in self._buckets
+                            ],
+                            sort=True,
+                        )
+                    ),
+                }
+            ),
+        )

homeassistant/components/backblaze/const.py

@@ -0,0 +1,23 @@
+"""Constants for the Backblaze integration."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+import logging
+from typing import Final
+
+from homeassistant.util.hass_dict import HassKey
+
+DOMAIN: Final = "backblaze"
+
+LOGGER = logging.getLogger(__package__)
+
+SEPARATOR: Final = "#!#!#"
+
+CONF_APPLICATION_KEY_ID: Final = "application_key_id"
+CONF_APPLICATION_KEY: Final = "application_key"
+CONF_BUCKET: Final = "bucket"
+
+DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
+    f"{DOMAIN}.backup_agent_listeners"
+)

homeassistant/components/backblaze/manifest.json

@@ -0,0 +1,10 @@
+{
+  "domain": "backblaze",
+  "name": "Backblaze B2",
+  "codeowners": ["@frenck"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/backblaze",
+  "integration_type": "service",
+  "iot_class": "local_polling",
+  "requirements": ["b2sdk==2.7.0"]
+}

homeassistant/components/backblaze/quality_scale.yaml

@@ -0,0 +1,85 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: Integration does not register custom actions.
+  appropriate-polling:
+    status: exempt
+    comment: |
+      This integration does not poll.
+  brands: done
+  common-modules: done
+  config-flow-test-coverage: done
+  config-flow: done
+  dependency-transparency: done
+  docs-actions:
+    status: exempt
+    comment: |
+      This integration does not have any custom actions.
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup:
+    status: exempt
+    comment: |
+      Entities of this integration does not explicitly subscribe to events.
+  entity-unique-id:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  has-entity-name:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  runtime-data: done
+  test-before-configure: done
+  test-before-setup: done
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions: todo
+  config-entry-unloading: todo
+  docs-configuration-parameters:
+    status: exempt
+    comment: |
+      This integration does not have any configuration parameters.
+  docs-installation-parameters: todo
+  entity-unavailable: todo
+  integration-owner: todo
+  log-when-unavailable: todo
+  parallel-updates: todo
+  reauthentication-flow: todo
+  test-coverage: todo
+  # Gold
+  devices: todo
+  diagnostics: todo
+  discovery-update-info: todo
+  discovery: todo
+  docs-data-update: todo
+  docs-examples: todo
+  docs-known-limitations: todo
+  docs-supported-devices: todo
+  docs-supported-functions: todo
+  docs-troubleshooting: todo
+  docs-use-cases: todo
+  dynamic-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single service.
+  entity-category: todo
+  entity-device-class: todo
+  entity-disabled-by-default: todo
+  entity-translations: todo
+  exception-translations: todo
+  icon-translations: todo
+  reconfiguration-flow: todo
+  repair-issues: todo
+  stale-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single service.
+
+  # Platinum
+  async-dependency: todo
+  inject-websession: todo
+  strict-typing: todo

homeassistant/components/backblaze/strings.json

@@ -0,0 +1,33 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    },
+    "error": {
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "step": {
+      "bucket": {
+        "data": {
+          "bucket": "Bucket"
+        },
+        "data_description": {
+          "bucket": "Select the bucked to store backups in."
+        },
+        "description": "Pick the bucket you want to use for backups. Please note, it is best to create a dedicated bucket for Home Assistant backups."
+      },
+      "user": {
+        "data": {
+          "application_key": "Application key",
+          "application_key_id": "Application key ID"
+        },
+        "data_description": {
+          "application_key": "The application key Backblaze B2 generated for you.",
+          "application_key_id": "The ID of the application key you created in the Backblaze B2 web interface. Not: This is not the same as the key name!"
+        },
+        "description": "Set up your Backblaze B2 bucket to store backups.\n\nTo do so, you will need to create an application key in the Backblaze B2 account. This key will be used to access the bucket you want to use for backups."
+      }
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -78,6 +78,7 @@ FLOWS = {
         "azure_data_explorer",
         "azure_devops",
         "azure_event_hub",
+        "backblaze",
         "baf",
         "balboa",
         "bang_olufsen",

homeassistant/generated/integrations.json

@@ -612,6 +612,12 @@
       "config_flow": true,
       "iot_class": "cloud_push"
     },
+    "backblaze": {
+      "name": "Backblaze B2",
+      "integration_type": "service",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "baf": {
       "name": "Big Ass Fans",
       "integration_type": "hub",

requirements_all.txt

@@ -556,6 +556,9 @@ azure-kusto-ingest==4.5.1
 # homeassistant.components.azure_service_bus
 azure-servicebus==7.10.0
 
+# homeassistant.components.backblaze
+b2sdk==2.7.0
+
 # homeassistant.components.holiday
 babel==2.15.0
 

requirements_test_all.txt

@@ -502,6 +502,9 @@ azure-kusto-data[aio]==4.5.1
 # homeassistant.components.azure_data_explorer
 azure-kusto-ingest==4.5.1
 
+# homeassistant.components.backblaze
+b2sdk==2.7.0
+
 # homeassistant.components.holiday
 babel==2.15.0
 

tests/components/backblaze/__init__.py

@@ -0,0 +1 @@
+"""Integration tests for the Backblaze B2 integration."""

tests/components/backblaze/conftest.py

@@ -0,0 +1,63 @@
+"""Fixtures for the Backblaze integration tests."""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from homeassistant.components.backblaze.const import (
+    CONF_APPLICATION_KEY,
+    CONF_APPLICATION_KEY_ID,
+    CONF_BUCKET,
+    DOMAIN,
+)
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture
+def mock_config_entry() -> MockConfigEntry:
+    """Return the default mocked config entry."""
+    return MockConfigEntry(
+        title="Backblaze B2",
+        domain=DOMAIN,
+        data={
+            CONF_APPLICATION_KEY: "secret",
+            CONF_APPLICATION_KEY_ID: "keyid",
+            CONF_BUCKET: "bucket",
+        },
+        unique_id="bucket",
+    )
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[None]:
+    """Mock setting up a config entry."""
+    with patch(
+        "homeassistant.components.backblaze.async_setup_entry", return_value=True
+    ):
+        yield
+
+
+@pytest.fixture
+def mock_backblaze() -> Generator[MagicMock]:
+    """Return a mocked Backblaze client."""
+
+    bucket1 = MagicMock()
+    bucket1.name = "my-bucket"
+    bucket1.id_ = "bucket"
+    bucket2 = MagicMock()
+    bucket2.name = "my-otherbucket"
+    bucket2.id_ = "bucket2"
+
+    with (
+        patch(
+            "homeassistant.components.backblaze.config_flow.B2Api", autospec=True
+        ) as backblaze_mock,
+    ):
+        backblaze = backblaze_mock.return_value
+        backblaze.list_buckets.return_value = [bucket1, bucket2]
+
+        yield backblaze

tests/components/backblaze/test_config_flow.py

@@ -0,0 +1,147 @@
+"""Configuration flow tests for the Backblaze B2 integration."""
+
+from unittest.mock import MagicMock
+
+from b2sdk.v2.exception import InvalidAuthToken
+import pytest
+
+from homeassistant.components.backblaze.const import (
+    CONF_APPLICATION_KEY,
+    CONF_APPLICATION_KEY_ID,
+    CONF_BUCKET,
+    DOMAIN,
+)
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from tests.common import MockConfigEntry
+
+pytestmark = pytest.mark.usefixtures("mock_setup_entry")
+
+
+@pytest.mark.usefixtures("mock_backblaze")
+async def test_user_flow(hass: HomeAssistant) -> None:
+    """Test the full happy path user flow from start to finish."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_APPLICATION_KEY: "secret",
+            CONF_APPLICATION_KEY_ID: "keyid",
+        },
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "bucket"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={CONF_BUCKET: "bucket"},
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+    config_entry = result["result"]
+    assert config_entry.unique_id == "bucket"
+    assert config_entry.data == {
+        CONF_APPLICATION_KEY_ID: "keyid",
+        CONF_APPLICATION_KEY: "secret",
+        CONF_BUCKET: "bucket",
+    }
+    assert not config_entry.options
+
+
+@pytest.mark.parametrize(
+    ("side_effect", "expected_error"),
+    [
+        (InvalidAuthToken("back", "up"), {"base": "invalid_auth"}),
+        (Exception, {"base": "unknown"}),
+    ],
+)
+async def test_user_flow_errors(
+    hass: HomeAssistant,
+    mock_backblaze: MagicMock,
+    side_effect: Exception,
+    expected_error: dict[str, str],
+) -> None:
+    """Test we show user form on an error."""
+    mock_backblaze.list_buckets.side_effect = side_effect
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={
+            CONF_APPLICATION_KEY: "secret",
+            CONF_APPLICATION_KEY_ID: "keyid",
+        },
+    )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == expected_error
+
+    mock_backblaze.list_buckets.side_effect = None
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_APPLICATION_KEY: "secret",
+            CONF_APPLICATION_KEY_ID: "keyid",
+        },
+    )
+
+    assert result["type"] is FlowResultType.FORM
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_BUCKET: "bucket",
+        },
+    )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+    config_entry = result["result"]
+    assert config_entry.unique_id == "bucket"
+    assert config_entry.data == {
+        CONF_APPLICATION_KEY_ID: "keyid",
+        CONF_APPLICATION_KEY: "secret",
+        CONF_BUCKET: "bucket",
+    }
+    assert not config_entry.options
+
+
+@pytest.mark.usefixtures("mock_backblaze")
+async def test_user_flow_already_configured(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Test configuration flow aborts when the bucket is already configured."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN,
+        context={"source": SOURCE_USER},
+        data={
+            CONF_APPLICATION_KEY: "secret",
+            CONF_APPLICATION_KEY_ID: "keyid",
+        },
+    )
+
+    assert result["type"] is FlowResultType.FORM
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        user_input={
+            CONF_BUCKET: "bucket",
+        },
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"