diff --git a/.strict-typing b/.strict-typing
index 95eb2abb4b4..1df49300b1e 100644
--- a/.strict-typing
+++ b/.strict-typing
@@ -103,6 +103,7 @@ homeassistant.components.auth.*
 homeassistant.components.automation.*
 homeassistant.components.awair.*
 homeassistant.components.axis.*
+homeassistant.components.azure_storage.*
 homeassistant.components.backup.*
 homeassistant.components.baf.*
 homeassistant.components.bang_olufsen.*
diff --git a/CODEOWNERS b/CODEOWNERS
index bb8545c46b7..87f170009f0 100644
--- a/CODEOWNERS
+++ b/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/azure_storage/ @zweckj
+/tests/components/azure_storage/ @zweckj
 /homeassistant/components/backup/ @home-assistant/core
 /tests/components/backup/ @home-assistant/core
 /homeassistant/components/baf/ @bdraco @jfroy
diff --git a/homeassistant/brands/microsoft.json b/homeassistant/brands/microsoft.json
index 0e00c4a7bc3..918f67f06dd 100644
--- a/homeassistant/brands/microsoft.json
+++ b/homeassistant/brands/microsoft.json
@@ -6,6 +6,7 @@
     "azure_devops",
     "azure_event_hub",
     "azure_service_bus",
+    "azure_storage",
     "microsoft_face_detect",
     "microsoft_face_identify",
     "microsoft_face",
diff --git a/homeassistant/components/azure_storage/__init__.py b/homeassistant/components/azure_storage/__init__.py
new file mode 100644
index 00000000000..873a9ab90ca
--- /dev/null
+++ b/homeassistant/components/azure_storage/__init__.py
@@ -0,0 +1,82 @@
+"""The Azure Storage integration."""
+
+from aiohttp import ClientTimeout
+from azure.core.exceptions import (
+    ClientAuthenticationError,
+    HttpResponseError,
+    ResourceNotFoundError,
+)
+from azure.core.pipeline.transport._aiohttp import (
+    AioHttpTransport,
+)  # need to import from private file, as it is not properly imported in the init
+from azure.storage.blob.aio import ContainerClient
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_create_clientsession
+
+from .const import (
+    CONF_ACCOUNT_NAME,
+    CONF_CONTAINER_NAME,
+    CONF_STORAGE_ACCOUNT_KEY,
+    DATA_BACKUP_AGENT_LISTENERS,
+    DOMAIN,
+)
+
+type AzureStorageConfigEntry = ConfigEntry[ContainerClient]
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: AzureStorageConfigEntry
+) -> bool:
+    """Set up Azure Storage integration."""
+    # set increase aiohttp timeout for long running operations (up/download)
+    session = async_create_clientsession(
+        hass, timeout=ClientTimeout(connect=10, total=12 * 60 * 60)
+    )
+    container_client = ContainerClient(
+        account_url=f"https://{entry.data[CONF_ACCOUNT_NAME]}.blob.core.windows.net/",
+        container_name=entry.data[CONF_CONTAINER_NAME],
+        credential=entry.data[CONF_STORAGE_ACCOUNT_KEY],
+        transport=AioHttpTransport(session=session),
+    )
+
+    try:
+        if not await container_client.exists():
+            await container_client.create_container()
+    except ResourceNotFoundError as err:
+        raise ConfigEntryError(
+            translation_domain=DOMAIN,
+            translation_key="account_not_found",
+            translation_placeholders={CONF_ACCOUNT_NAME: entry.data[CONF_ACCOUNT_NAME]},
+        ) from err
+    except ClientAuthenticationError as err:
+        raise ConfigEntryError(
+            translation_domain=DOMAIN,
+            translation_key="invalid_auth",
+            translation_placeholders={CONF_ACCOUNT_NAME: entry.data[CONF_ACCOUNT_NAME]},
+        ) from err
+    except HttpResponseError as err:
+        raise ConfigEntryNotReady(
+            translation_domain=DOMAIN,
+            translation_key="cannot_connect",
+            translation_placeholders={CONF_ACCOUNT_NAME: entry.data[CONF_ACCOUNT_NAME]},
+        ) from err
+
+    entry.runtime_data = container_client
+
+    def _async_notify_backup_listeners() -> None:
+        for listener in hass.data.get(DATA_BACKUP_AGENT_LISTENERS, []):
+            listener()
+
+    entry.async_on_unload(entry.async_on_state_change(_async_notify_backup_listeners))
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: AzureStorageConfigEntry
+) -> bool:
+    """Unload an Azure Storage config entry."""
+    return True
diff --git a/homeassistant/components/azure_storage/backup.py b/homeassistant/components/azure_storage/backup.py
new file mode 100644
index 00000000000..6f39295761d
--- /dev/null
+++ b/homeassistant/components/azure_storage/backup.py
@@ -0,0 +1,182 @@
+"""Support for Azure Storage backup."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable, Coroutine
+from functools import wraps
+import json
+import logging
+from typing import Any, Concatenate
+
+from azure.core.exceptions import HttpResponseError
+from azure.storage.blob import BlobProperties
+
+from homeassistant.components.backup import (
+    AgentBackup,
+    BackupAgent,
+    BackupAgentError,
+    BackupNotFound,
+    suggested_filename,
+)
+from homeassistant.core import HomeAssistant, callback
+
+from . import AzureStorageConfigEntry
+from .const import DATA_BACKUP_AGENT_LISTENERS, DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+METADATA_VERSION = "1"
+
+
+async def async_get_backup_agents(
+    hass: HomeAssistant,
+) -> list[BackupAgent]:
+    """Return a list of backup agents."""
+    entries: list[AzureStorageConfigEntry] = hass.config_entries.async_loaded_entries(
+        DOMAIN
+    )
+    return [AzureStorageBackupAgent(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)
+        if not hass.data[DATA_BACKUP_AGENT_LISTENERS]:
+            hass.data.pop(DATA_BACKUP_AGENT_LISTENERS)
+
+    return remove_listener
+
+
+def handle_backup_errors[_R, **P](
+    func: Callable[Concatenate[AzureStorageBackupAgent, P], Coroutine[Any, Any, _R]],
+) -> Callable[Concatenate[AzureStorageBackupAgent, P], Coroutine[Any, Any, _R]]:
+    """Handle backup errors."""
+
+    @wraps(func)
+    async def wrapper(
+        self: AzureStorageBackupAgent, *args: P.args, **kwargs: P.kwargs
+    ) -> _R:
+        try:
+            return await func(self, *args, **kwargs)
+        except HttpResponseError as err:
+            _LOGGER.debug(
+                "Error during backup in %s: Status %s, message %s",
+                func.__name__,
+                err.status_code,
+                err.message,
+                exc_info=True,
+            )
+            raise BackupAgentError(
+                f"Error during backup operation in {func.__name__}:"
+                f" Status {err.status_code}, message: {err.message}"
+            ) from err
+
+    return wrapper
+
+
+class AzureStorageBackupAgent(BackupAgent):
+    """Azure storage backup agent."""
+
+    domain = DOMAIN
+
+    def __init__(self, hass: HomeAssistant, entry: AzureStorageConfigEntry) -> None:
+        """Initialize the Azure storage backup agent."""
+        super().__init__()
+        self._client = entry.runtime_data
+        self.name = entry.title
+        self.unique_id = entry.entry_id
+
+    @handle_backup_errors
+    async def async_download_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> AsyncIterator[bytes]:
+        """Download a backup file."""
+        blob = await self._find_blob_by_backup_id(backup_id)
+        if blob is None:
+            raise BackupNotFound(f"Backup {backup_id} not found")
+        download_stream = await self._client.download_blob(blob.name)
+        return download_stream.chunks()
+
+    @handle_backup_errors
+    async def async_upload_backup(
+        self,
+        *,
+        open_stream: Callable[[], Coroutine[Any, Any, AsyncIterator[bytes]]],
+        backup: AgentBackup,
+        **kwargs: Any,
+    ) -> None:
+        """Upload a backup."""
+
+        metadata = {
+            "metadata_version": METADATA_VERSION,
+            "backup_id": backup.backup_id,
+            "backup_metadata": json.dumps(backup.as_dict()),
+        }
+
+        await self._client.upload_blob(
+            name=suggested_filename(backup),
+            metadata=metadata,
+            data=await open_stream(),
+            length=backup.size,
+        )
+
+    @handle_backup_errors
+    async def async_delete_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> None:
+        """Delete a backup file."""
+        blob = await self._find_blob_by_backup_id(backup_id)
+        if blob is None:
+            return
+        await self._client.delete_blob(blob.name)
+
+    @handle_backup_errors
+    async def async_list_backups(self, **kwargs: Any) -> list[AgentBackup]:
+        """List backups."""
+        backups: list[AgentBackup] = []
+        async for blob in self._client.list_blobs(include="metadata"):
+            metadata = blob.metadata
+
+            if metadata.get("metadata_version") == METADATA_VERSION:
+                backups.append(
+                    AgentBackup.from_dict(json.loads(metadata["backup_metadata"]))
+                )
+
+        return backups
+
+    @handle_backup_errors
+    async def async_get_backup(
+        self,
+        backup_id: str,
+        **kwargs: Any,
+    ) -> AgentBackup | None:
+        """Return a backup."""
+        blob = await self._find_blob_by_backup_id(backup_id)
+        if blob is None:
+            return None
+
+        return AgentBackup.from_dict(json.loads(blob.metadata["backup_metadata"]))
+
+    async def _find_blob_by_backup_id(self, backup_id: str) -> BlobProperties | None:
+        """Find a blob by backup id."""
+        async for blob in self._client.list_blobs(include="metadata"):
+            if (
+                backup_id == blob.metadata.get("backup_id", "")
+                and blob.metadata.get("metadata_version") == METADATA_VERSION
+            ):
+                return blob
+        return None
diff --git a/homeassistant/components/azure_storage/config_flow.py b/homeassistant/components/azure_storage/config_flow.py
new file mode 100644
index 00000000000..e5b1214fa5b
--- /dev/null
+++ b/homeassistant/components/azure_storage/config_flow.py
@@ -0,0 +1,72 @@
+"""Config flow for Azure Storage integration."""
+
+import logging
+from typing import Any
+
+from azure.core.exceptions import ClientAuthenticationError, ResourceNotFoundError
+from azure.core.pipeline.transport._aiohttp import (
+    AioHttpTransport,
+)  # need to import from private file, as it is not properly imported in the init
+from azure.storage.blob.aio import ContainerClient
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+
+from .const import (
+    CONF_ACCOUNT_NAME,
+    CONF_CONTAINER_NAME,
+    CONF_STORAGE_ACCOUNT_KEY,
+    DOMAIN,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AzureStorageConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for azure storage."""
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """User step for Azure Storage."""
+
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            self._async_abort_entries_match(
+                {CONF_ACCOUNT_NAME: user_input[CONF_ACCOUNT_NAME]}
+            )
+            container_client = ContainerClient(
+                account_url=f"https://{user_input[CONF_ACCOUNT_NAME]}.blob.core.windows.net/",
+                container_name=user_input[CONF_CONTAINER_NAME],
+                credential=user_input[CONF_STORAGE_ACCOUNT_KEY],
+                transport=AioHttpTransport(session=async_get_clientsession(self.hass)),
+            )
+            try:
+                await container_client.exists()
+            except ResourceNotFoundError:
+                errors["base"] = "cannot_connect"
+            except ClientAuthenticationError:
+                errors[CONF_STORAGE_ACCOUNT_KEY] = "invalid_auth"
+            except Exception:
+                _LOGGER.exception("Unknown exception occurred")
+                errors["base"] = "unknown"
+            if not errors:
+                return self.async_create_entry(
+                    title=f"{user_input[CONF_ACCOUNT_NAME]}/{user_input[CONF_CONTAINER_NAME]}",
+                    data=user_input,
+                )
+
+        return self.async_show_form(
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_ACCOUNT_NAME): str,
+                    vol.Required(
+                        CONF_CONTAINER_NAME, default="home-assistant-backups"
+                    ): str,
+                    vol.Required(CONF_STORAGE_ACCOUNT_KEY): str,
+                }
+            ),
+            errors=errors,
+        )
diff --git a/homeassistant/components/azure_storage/const.py b/homeassistant/components/azure_storage/const.py
new file mode 100644
index 00000000000..efcb338a096
--- /dev/null
+++ b/homeassistant/components/azure_storage/const.py
@@ -0,0 +1,16 @@
+"""Constants for the Azure Storage integration."""
+
+from collections.abc import Callable
+from typing import Final
+
+from homeassistant.util.hass_dict import HassKey
+
+DOMAIN: Final = "azure_storage"
+
+CONF_STORAGE_ACCOUNT_KEY: Final = "storage_account_key"
+CONF_ACCOUNT_NAME: Final = "account_name"
+CONF_CONTAINER_NAME: Final = "container_name"
+
+DATA_BACKUP_AGENT_LISTENERS: HassKey[list[Callable[[], None]]] = HassKey(
+    f"{DOMAIN}.backup_agent_listeners"
+)
diff --git a/homeassistant/components/azure_storage/manifest.json b/homeassistant/components/azure_storage/manifest.json
new file mode 100644
index 00000000000..8f2d8aeaca7
--- /dev/null
+++ b/homeassistant/components/azure_storage/manifest.json
@@ -0,0 +1,12 @@
+{
+  "domain": "azure_storage",
+  "name": "Azure Storage",
+  "codeowners": ["@zweckj"],
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/azure_storage",
+  "integration_type": "service",
+  "iot_class": "cloud_polling",
+  "loggers": ["azure-storage-blob"],
+  "quality_scale": "bronze",
+  "requirements": ["azure-storage-blob==12.24.0"]
+}
diff --git a/homeassistant/components/azure_storage/quality_scale.yaml b/homeassistant/components/azure_storage/quality_scale.yaml
new file mode 100644
index 00000000000..6b6f90de494
--- /dev/null
+++ b/homeassistant/components/azure_storage/quality_scale.yaml
@@ -0,0 +1,133 @@
+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: done
+  config-entry-unloading: done
+  docs-configuration-parameters:
+    status: exempt
+    comment: |
+      This integration does not have any configuration parameters.
+  docs-installation-parameters: done
+  entity-unavailable:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  integration-owner: done
+  log-when-unavailable:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  parallel-updates:
+    status: exempt
+    comment: |
+      This integration does not have platforms.
+  reauthentication-flow: todo
+  test-coverage: done
+
+  # Gold
+  devices:
+    status: exempt
+    comment: |
+      This integration connects to a single service.
+  diagnostics:
+    status: exempt
+    comment: |
+      There is no data to diagnose.
+  discovery-update-info:
+    status: exempt
+    comment: |
+      This integration is a cloud service and does not support discovery.
+  discovery:
+    status: exempt
+    comment: |
+      This integration is a cloud service and does not support discovery.
+  docs-data-update:
+    status: exempt
+    comment: |
+      This integration does not poll or push.
+  docs-examples:
+    status: exempt
+    comment: |
+      This integration only serves backup.
+  docs-known-limitations: done
+  docs-supported-devices:
+    status: exempt
+    comment: |
+      This integration is a cloud service.
+  docs-supported-functions:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  docs-troubleshooting: done
+  docs-use-cases: done
+  dynamic-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single service.
+  entity-category:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  entity-device-class:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  entity-disabled-by-default:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  entity-translations:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  exception-translations: done
+  icon-translations:
+    status: exempt
+    comment: |
+      This integration does not have entities.
+  reconfiguration-flow: todo
+  repair-issues: done
+  stale-devices:
+    status: exempt
+    comment: |
+      This integration connects to a single service.
+  # Platinum
+  async-dependency: done
+  inject-websession: done
+  strict-typing: done
diff --git a/homeassistant/components/azure_storage/strings.json b/homeassistant/components/azure_storage/strings.json
new file mode 100644
index 00000000000..4bd4cb0dfba
--- /dev/null
+++ b/homeassistant/components/azure_storage/strings.json
@@ -0,0 +1,48 @@
+{
+  "config": {
+    "error": {
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "step": {
+      "user": {
+        "data": {
+          "storage_account_key": "Storage account key",
+          "account_name": "Account name",
+          "container_name": "Container name"
+        },
+        "data_description": {
+          "storage_account_key": "Storage account access key used for authorization",
+          "account_name": "Name of the storage account",
+          "container_name": "Name of the storage container to be used (will be created if it does not exist)"
+        },
+        "description": "Set up an Azure (Blob) storage account to be used for backups.",
+        "title": "Add Azure storage account"
+      }
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_account%]"
+    }
+  },
+  "issues": {
+    "container_not_found": {
+      "title": "Storage container not found",
+      "description": "The storage container {container_name} has not been found in the storage account. Please re-create it manually, then fix this issue."
+    }
+  },
+  "exceptions": {
+    "account_not_found": {
+      "message": "Storage account {account_name} not found"
+    },
+    "cannot_connect": {
+      "message": "Can not connect to storage account {account_name}"
+    },
+    "invalid_auth": {
+      "message": "Authentication failed for storage account {account_name}"
+    },
+    "container_not_found": {
+      "message": "Storage container {container_name} not found"
+    }
+  }
+}
diff --git a/homeassistant/generated/config_flows.py b/homeassistant/generated/config_flows.py
index de581c65297..8284f77ef94 100644
--- a/homeassistant/generated/config_flows.py
+++ b/homeassistant/generated/config_flows.py
@@ -79,6 +79,7 @@ FLOWS = {
         "azure_data_explorer",
         "azure_devops",
         "azure_event_hub",
+        "azure_storage",
         "baf",
         "balboa",
         "bang_olufsen",
diff --git a/homeassistant/generated/integrations.json b/homeassistant/generated/integrations.json
index 41083ee8e8c..01ff9d14d90 100644
--- a/homeassistant/generated/integrations.json
+++ b/homeassistant/generated/integrations.json
@@ -3800,6 +3800,12 @@
           "iot_class": "cloud_push",
           "name": "Azure Service Bus"
         },
+        "azure_storage": {
+          "integration_type": "service",
+          "config_flow": true,
+          "iot_class": "cloud_polling",
+          "name": "Azure Storage"
+        },
         "microsoft_face_detect": {
           "integration_type": "hub",
           "config_flow": false,
diff --git a/mypy.ini b/mypy.ini
index a04242dc66d..a6203993c87 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -785,6 +785,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.azure_storage.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.backup.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true
diff --git a/requirements_all.txt b/requirements_all.txt
index 87dd9bb204e..3b80e4f78a6 100644
--- a/requirements_all.txt
+++ b/requirements_all.txt
@@ -571,6 +571,9 @@ azure-kusto-ingest==4.5.1
 # homeassistant.components.azure_service_bus
 azure-servicebus==7.10.0
 
+# homeassistant.components.azure_storage
+azure-storage-blob==12.24.0
+
 # homeassistant.components.holiday
 babel==2.15.0
 
diff --git a/requirements_test_all.txt b/requirements_test_all.txt
index f55ea287d37..4ec3192285d 100644
--- a/requirements_test_all.txt
+++ b/requirements_test_all.txt
@@ -517,6 +517,9 @@ azure-kusto-data[aio]==4.5.1
 # homeassistant.components.azure_data_explorer
 azure-kusto-ingest==4.5.1
 
+# homeassistant.components.azure_storage
+azure-storage-blob==12.24.0
+
 # homeassistant.components.holiday
 babel==2.15.0
 
diff --git a/tests/components/azure_storage/__init__.py b/tests/components/azure_storage/__init__.py
new file mode 100644
index 00000000000..bfd2e72d979
--- /dev/null
+++ b/tests/components/azure_storage/__init__.py
@@ -0,0 +1,14 @@
+"""Azure Storage integration tests."""
+
+from homeassistant.core import HomeAssistant
+
+from tests.common import MockConfigEntry
+
+
+async def setup_integration(
+    hass: HomeAssistant, mock_config_entry: MockConfigEntry
+) -> None:
+    """Set up the azure_storage integration for testing."""
+    mock_config_entry.add_to_hass(hass)
+    await hass.config_entries.async_setup(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
diff --git a/tests/components/azure_storage/conftest.py b/tests/components/azure_storage/conftest.py
new file mode 100644
index 00000000000..7c583ac391e
--- /dev/null
+++ b/tests/components/azure_storage/conftest.py
@@ -0,0 +1,63 @@
+"""Fixtures for Azure Storage tests."""
+
+from collections.abc import AsyncIterator, Generator
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from azure.storage.blob import BlobProperties
+import pytest
+
+from homeassistant.components.azure_storage.const import DOMAIN
+
+from .const import BACKUP_METADATA, USER_INPUT
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture
+def mock_setup_entry() -> Generator[AsyncMock]:
+    """Mock setting up a config entry."""
+    with patch(
+        "homeassistant.components.azure_storage.async_setup_entry", return_value=True
+    ) as mock_setup:
+        yield mock_setup
+
+
+@pytest.fixture(autouse=True)
+def mock_client() -> Generator[MagicMock]:
+    """Mock the Azure Storage client."""
+    with (
+        patch(
+            "homeassistant.components.azure_storage.config_flow.ContainerClient",
+            autospec=True,
+        ) as container_client,
+        patch(
+            "homeassistant.components.azure_storage.ContainerClient",
+            new=container_client,
+        ),
+    ):
+        client = container_client.return_value
+        client.exists.return_value = False
+
+        async def async_list_blobs():
+            yield BlobProperties(metadata=BACKUP_METADATA)
+            yield BlobProperties(metadata=BACKUP_METADATA)
+
+        client.list_blobs.return_value = async_list_blobs()
+
+        class MockStream:
+            async def chunks(self) -> AsyncIterator[bytes]:
+                yield b"backup data"
+
+        client.download_blob.return_value = MockStream()
+
+        yield client
+
+
+@pytest.fixture
+def mock_config_entry() -> MockConfigEntry:
+    """Return the default mocked config entry."""
+    return MockConfigEntry(
+        title="account/container1",
+        domain=DOMAIN,
+        data=USER_INPUT,
+    )
diff --git a/tests/components/azure_storage/const.py b/tests/components/azure_storage/const.py
new file mode 100644
index 00000000000..4edb754f650
--- /dev/null
+++ b/tests/components/azure_storage/const.py
@@ -0,0 +1,36 @@
+"""Consts for Azure Storage tests."""
+
+from json import dumps
+
+from homeassistant.components.azure_storage.const import (
+    CONF_ACCOUNT_NAME,
+    CONF_CONTAINER_NAME,
+    CONF_STORAGE_ACCOUNT_KEY,
+)
+from homeassistant.components.backup import AgentBackup
+
+USER_INPUT = {
+    CONF_ACCOUNT_NAME: "account",
+    CONF_CONTAINER_NAME: "container1",
+    CONF_STORAGE_ACCOUNT_KEY: "test",
+}
+
+TEST_BACKUP = AgentBackup(
+    addons=[],
+    backup_id="23e64aec",
+    date="2024-11-22T11:48:48.727189+01:00",
+    database_included=True,
+    extra_metadata={},
+    folders=[],
+    homeassistant_included=True,
+    homeassistant_version="2024.12.0.dev0",
+    name="Core 2024.12.0.dev0",
+    protected=False,
+    size=34519040,
+)
+
+BACKUP_METADATA = {
+    "metadata_version": "1",
+    "backup_id": "23e64aec",
+    "backup_metadata": dumps(TEST_BACKUP.as_dict()),
+}
diff --git a/tests/components/azure_storage/test_backup.py b/tests/components/azure_storage/test_backup.py
new file mode 100644
index 00000000000..4dc1de0a26e
--- /dev/null
+++ b/tests/components/azure_storage/test_backup.py
@@ -0,0 +1,317 @@
+"""Test the backups for OneDrive."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from io import StringIO
+from unittest.mock import ANY, Mock, patch
+
+from azure.core.exceptions import HttpResponseError
+from azure.storage.blob import BlobProperties
+import pytest
+
+from homeassistant.components.azure_storage.backup import (
+    async_register_backup_agents_listener,
+)
+from homeassistant.components.azure_storage.const import (
+    DATA_BACKUP_AGENT_LISTENERS,
+    DOMAIN,
+)
+from homeassistant.components.backup import DOMAIN as BACKUP_DOMAIN
+from homeassistant.core import HomeAssistant
+from homeassistant.setup import async_setup_component
+
+from . import setup_integration
+from .const import BACKUP_METADATA, TEST_BACKUP
+
+from tests.common import MockConfigEntry
+from tests.typing import ClientSessionGenerator, MagicMock, WebSocketGenerator
+
+
+@pytest.fixture(autouse=True)
+async def setup_backup_integration(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> AsyncGenerator[None]:
+    """Set up onedrive integration."""
+    with (
+        patch("homeassistant.components.backup.is_hassio", return_value=False),
+        patch("homeassistant.components.backup.store.STORE_DELAY_SAVE", 0),
+    ):
+        assert await async_setup_component(hass, BACKUP_DOMAIN, {})
+        await setup_integration(hass, mock_config_entry)
+
+        await hass.async_block_till_done()
+        yield
+
+
+async def test_agents_info(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test backup agent info."""
+    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": f"{DOMAIN}.{mock_config_entry.entry_id}",
+                "name": mock_config_entry.title,
+            },
+        ],
+    }
+
+
+async def test_agents_list_backups(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test agent list backups."""
+
+    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"] == [
+        {
+            "addons": [],
+            "agents": {
+                f"{DOMAIN}.{mock_config_entry.entry_id}": {
+                    "protected": False,
+                    "size": 34519040,
+                }
+            },
+            "backup_id": "23e64aec",
+            "date": "2024-11-22T11:48:48.727189+01:00",
+            "database_included": True,
+            "folders": [],
+            "homeassistant_included": True,
+            "homeassistant_version": "2024.12.0.dev0",
+            "name": "Core 2024.12.0.dev0",
+            "failed_agent_ids": [],
+            "extra_metadata": {},
+            "with_automatic_settings": None,
+        }
+    ]
+
+
+async def test_agents_get_backup(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test agent get backup."""
+
+    backup_id = TEST_BACKUP.backup_id
+    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"] == {}
+    assert response["result"]["backup"] == {
+        "addons": [],
+        "agents": {
+            f"{DOMAIN}.{mock_config_entry.entry_id}": {
+                "protected": False,
+                "size": 34519040,
+            }
+        },
+        "backup_id": "23e64aec",
+        "date": "2024-11-22T11:48:48.727189+01:00",
+        "database_included": True,
+        "folders": [],
+        "homeassistant_included": True,
+        "homeassistant_version": "2024.12.0.dev0",
+        "extra_metadata": {},
+        "name": "Core 2024.12.0.dev0",
+        "failed_agent_ids": [],
+        "with_automatic_settings": None,
+    }
+
+
+async def test_agents_get_backup_does_not_throw_on_not_found(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+) -> None:
+    """Test agent get backup does not throw on a backup not found."""
+
+    client = await hass_ws_client(hass)
+    await client.send_json_auto_id({"type": "backup/details", "backup_id": "random"})
+    response = await client.receive_json()
+
+    assert response["success"]
+    assert response["result"]["agent_errors"] == {}
+    assert response["result"]["backup"] is None
+
+
+async def test_agents_delete(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_client: MagicMock,
+) -> None:
+    """Test agent delete backup."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "backup/delete",
+            "backup_id": TEST_BACKUP.backup_id,
+        }
+    )
+    response = await client.receive_json()
+
+    assert response["success"]
+    assert response["result"] == {"agent_errors": {}}
+    mock_client.delete_blob.assert_called_once()
+
+
+async def test_agents_delete_not_throwing_on_not_found(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_client: MagicMock,
+) -> None:
+    """Test agent delete backup does not throw on a backup not found."""
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "backup/delete",
+            "backup_id": "random",
+        }
+    )
+    response = await client.receive_json()
+
+    assert response["success"]
+    assert response["result"] == {"agent_errors": {}}
+    assert mock_client.delete_blob.call_count == 0
+
+
+async def test_agents_upload(
+    hass_client: ClientSessionGenerator,
+    caplog: pytest.LogCaptureFixture,
+    mock_client: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test agent upload backup."""
+    client = await hass_client()
+    with (
+        patch(
+            "homeassistant.components.backup.manager.BackupManager.async_get_backup",
+        ) as fetch_backup,
+        patch(
+            "homeassistant.components.backup.manager.read_backup",
+            return_value=TEST_BACKUP,
+        ),
+        patch("pathlib.Path.open") as mocked_open,
+    ):
+        mocked_open.return_value.read = Mock(side_effect=[b"test", b""])
+        fetch_backup.return_value = TEST_BACKUP
+        resp = await client.post(
+            f"/api/backup/upload?agent_id={DOMAIN}.{mock_config_entry.entry_id}",
+            data={"file": StringIO("test")},
+        )
+
+    assert resp.status == 201
+    assert f"Uploading backup {TEST_BACKUP.backup_id}" in caplog.text
+    mock_client.upload_blob.assert_called_once_with(
+        name="Core_2024.12.0.dev0_2024-11-22_11.48_48727189.tar",
+        metadata=BACKUP_METADATA,
+        data=ANY,
+        length=ANY,
+    )
+
+
+async def test_agents_download(
+    hass_client: ClientSessionGenerator,
+    mock_client: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test agent download backup."""
+    client = await hass_client()
+    backup_id = BACKUP_METADATA["backup_id"]
+
+    resp = await client.get(
+        f"/api/backup/download/{backup_id}?agent_id={DOMAIN}.{mock_config_entry.entry_id}"
+    )
+    assert resp.status == 200
+    assert await resp.content.read() == b"backup data"
+    mock_client.download_blob.assert_called_once()
+
+
+async def test_agents_error_on_download_not_found(
+    hass_client: ClientSessionGenerator,
+    mock_client: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test agent download backup."""
+
+    async def async_list_blobs(
+        metadata: dict[str, str],
+    ) -> AsyncGenerator[BlobProperties]:
+        yield BlobProperties(metadata=metadata)
+
+    mock_client.list_blobs.side_effect = [
+        async_list_blobs(BACKUP_METADATA),
+        async_list_blobs({}),
+    ]
+    client = await hass_client()
+    backup_id = BACKUP_METADATA["backup_id"]
+
+    resp = await client.get(
+        f"/api/backup/download/{backup_id}?agent_id={DOMAIN}.{mock_config_entry.entry_id}"
+    )
+    assert resp.status == 404
+    assert mock_client.download_blob.call_count == 0
+
+
+async def test_error_during_delete(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    mock_client: MagicMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test the error wrapper."""
+    mock_client.delete_blob.side_effect = HttpResponseError("Failed to delete backup")
+
+    client = await hass_ws_client(hass)
+
+    await client.send_json_auto_id(
+        {
+            "type": "backup/delete",
+            "backup_id": BACKUP_METADATA["backup_id"],
+        }
+    )
+    response = await client.receive_json()
+
+    assert response["success"]
+    assert response["result"] == {
+        "agent_errors": {
+            f"{DOMAIN}.{mock_config_entry.entry_id}": (
+                "Error during backup operation in async_delete_backup: "
+                "Status None, message: Failed to delete backup"
+            )
+        }
+    }
+
+
+async def test_listeners_get_cleaned_up(hass: HomeAssistant) -> None:
+    """Test listener gets cleaned up."""
+    listener = MagicMock()
+    remove_listener = async_register_backup_agents_listener(hass, listener=listener)
+
+    hass.data[DATA_BACKUP_AGENT_LISTENERS] = [
+        listener
+    ]  # make sure it's the last listener
+    remove_listener()
+
+    assert hass.data.get(DATA_BACKUP_AGENT_LISTENERS) is None
diff --git a/tests/components/azure_storage/test_config_flow.py b/tests/components/azure_storage/test_config_flow.py
new file mode 100644
index 00000000000..ed8bbed0718
--- /dev/null
+++ b/tests/components/azure_storage/test_config_flow.py
@@ -0,0 +1,113 @@
+"""Test the Azure storage config flow."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+from azure.core.exceptions import ClientAuthenticationError, ResourceNotFoundError
+import pytest
+
+from homeassistant.components.azure_storage.const import (
+    CONF_ACCOUNT_NAME,
+    CONF_CONTAINER_NAME,
+    CONF_STORAGE_ACCOUNT_KEY,
+    DOMAIN,
+)
+from homeassistant.config_entries import SOURCE_USER, ConfigFlowResult
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from .const import USER_INPUT
+
+from tests.common import MockConfigEntry
+
+
+async def __async_start_flow(
+    hass: HomeAssistant,
+) -> ConfigFlowResult:
+    """Initialize the  config flow."""
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    await hass.async_block_till_done()
+    assert result["type"] is FlowResultType.FORM
+
+    return await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        USER_INPUT,
+    )
+
+
+async def test_flow(
+    hass: HomeAssistant,
+    mock_client: MagicMock,
+    mock_setup_entry: AsyncMock,
+) -> None:
+    """Test config flow."""
+    mock_client.exists.return_value = False
+    result = await __async_start_flow(hass)
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert (
+        result["title"]
+        == f"{USER_INPUT[CONF_ACCOUNT_NAME]}/{USER_INPUT[CONF_CONTAINER_NAME]}"
+    )
+    assert result["data"] == {
+        CONF_ACCOUNT_NAME: "account",
+        CONF_CONTAINER_NAME: "container1",
+        CONF_STORAGE_ACCOUNT_KEY: "test",
+    }
+
+
+@pytest.mark.parametrize(
+    ("exception", "errors"),
+    [
+        (ResourceNotFoundError, {"base": "cannot_connect"}),
+        (ClientAuthenticationError, {CONF_STORAGE_ACCOUNT_KEY: "invalid_auth"}),
+        (Exception, {"base": "unknown"}),
+    ],
+)
+async def test_flow_errors(
+    hass: HomeAssistant,
+    mock_client: MagicMock,
+    mock_setup_entry: AsyncMock,
+    exception: Exception,
+    errors: dict[str, str],
+) -> None:
+    """Test config flow errors."""
+    mock_client.exists.side_effect = exception
+
+    result = await __async_start_flow(hass)
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["errors"] == errors
+
+    # fix and finish the test
+    mock_client.exists.side_effect = None
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        USER_INPUT,
+    )
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert (
+        result["title"]
+        == f"{USER_INPUT[CONF_ACCOUNT_NAME]}/{USER_INPUT[CONF_CONTAINER_NAME]}"
+    )
+    assert result["data"] == {
+        CONF_ACCOUNT_NAME: "account",
+        CONF_CONTAINER_NAME: "container1",
+        CONF_STORAGE_ACCOUNT_KEY: "test",
+    }
+
+
+async def test_abort_if_already_configured(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test we abort if the account is already configured."""
+    mock_config_entry.add_to_hass(hass)
+
+    result = await __async_start_flow(hass)
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"
diff --git a/tests/components/azure_storage/test_init.py b/tests/components/azure_storage/test_init.py
new file mode 100644
index 00000000000..ca725134737
--- /dev/null
+++ b/tests/components/azure_storage/test_init.py
@@ -0,0 +1,54 @@
+"""Test the Azure storage integration."""
+
+from unittest.mock import MagicMock
+
+from azure.core.exceptions import (
+    ClientAuthenticationError,
+    HttpResponseError,
+    ResourceNotFoundError,
+)
+import pytest
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from . import setup_integration
+
+from tests.common import MockConfigEntry
+
+
+async def test_load_unload_config_entry(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test loading and unloading the integration."""
+    await setup_integration(hass, mock_config_entry)
+
+    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
+
+
+@pytest.mark.parametrize(
+    ("exception", "state"),
+    [
+        (ClientAuthenticationError, ConfigEntryState.SETUP_ERROR),
+        (HttpResponseError, ConfigEntryState.SETUP_RETRY),
+        (ResourceNotFoundError, ConfigEntryState.SETUP_ERROR),
+    ],
+)
+async def test_setup_errors(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_client: MagicMock,
+    exception: Exception,
+    state: ConfigEntryState,
+) -> None:
+    """Test various setup errors."""
+    mock_client.exists.side_effect = exception()
+    await setup_integration(hass, mock_config_entry)
+
+    assert mock_config_entry.state is state