Store

homeassistant/components/sql/config_flow.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 import logging
 from typing import Any
 
@@ -11,9 +12,11 @@ from sqlalchemy.exc import MultipleResultsFound, NoSuchColumnError, SQLAlchemyEr
 from sqlalchemy.orm import Session, scoped_session, sessionmaker
 import voluptuous as vol
 
+from homeassistant.components import websocket_api
 from homeassistant.components.recorder import CONF_DB_URL, get_instance
 from homeassistant.components.sensor import (
     CONF_STATE_CLASS,
+    DOMAIN as SENSOR_DOMAIN,
     SensorDeviceClass,
     SensorStateClass,
 )
@@ -29,18 +32,23 @@ from homeassistant.const import (
     CONF_UNIT_OF_MEASUREMENT,
     CONF_VALUE_TEMPLATE,
 )
-from homeassistant.core import async_get_hass, callback
+from homeassistant.core import HomeAssistant, async_get_hass, callback
 from homeassistant.data_entry_flow import section
-from homeassistant.exceptions import TemplateError
+from homeassistant.exceptions import HomeAssistantError, TemplateError
 from homeassistant.helpers import selector
+from homeassistant.helpers.entity_platform import PlatformData
+from homeassistant.helpers.template import Template
+from homeassistant.helpers.trigger_template_entity import ValueTemplate
 
 from .const import CONF_ADVANCED_OPTIONS, CONF_COLUMN_NAME, CONF_QUERY, DOMAIN
+from .sensor import TRIGGER_ENTITY_OPTIONS, SQLSensor
 from .util import (
     EmptyQueryError,
     InvalidSqlQuery,
     MultipleQueryError,
     NotSelectQueryError,
     UnknownQueryTypeError,
+    async_create_sessionmaker,
     check_and_render_sql_query,
     resolve_db_url,
 )
@@ -168,6 +176,11 @@ class SQLConfigFlow(ConfigFlow, domain=DOMAIN):
 
     data: dict[str, Any]
 
+    @staticmethod
+    async def async_setup_preview(hass: HomeAssistant) -> None:
+        """Set up preview WS API."""
+        websocket_api.async_register_command(hass, ws_start_preview)
+
     @staticmethod
     @callback
     def async_get_options_flow(
@@ -258,6 +271,7 @@ class SQLConfigFlow(ConfigFlow, domain=DOMAIN):
             data_schema=self.add_suggested_values_to_schema(OPTIONS_SCHEMA, user_input),
             errors=errors,
             description_placeholders=description_placeholders,
+            preview="sql",
         )
 
 
@@ -321,4 +335,106 @@ class SQLOptionsFlowHandler(OptionsFlowWithReload):
             ),
             errors=errors,
             description_placeholders=description_placeholders,
+            preview="sql",
         )
+
+    @staticmethod
+    async def async_setup_preview(hass: HomeAssistant) -> None:
+        """Set up preview WS API."""
+        websocket_api.async_register_command(hass, ws_start_preview)
+
+
+@websocket_api.websocket_command(
+    {
+        vol.Required("type"): "sql/start_preview",
+        vol.Required("flow_id"): str,
+        vol.Required("flow_type"): vol.Any("config_flow", "options_flow"),
+        vol.Required("user_input"): dict,
+    }
+)
+@websocket_api.async_response
+async def ws_start_preview(
+    hass: HomeAssistant,
+    connection: websocket_api.ActiveConnection,
+    msg: dict[str, Any],
+) -> None:
+    """Generate a preview."""
+
+    if msg["flow_type"] == "config_flow":
+        flow_status = hass.config_entries.flow.async_get(msg["flow_id"])
+        flow_sets = hass.config_entries.flow._handler_progress_index.get(  # noqa: SLF001
+            flow_status["handler"]
+        )
+        assert flow_sets
+        config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
+        print(list(flow_sets)[0].data)
+        name = list(flow_sets)[0].data[CONF_NAME]
+        db_url = resolve_db_url(hass, list(flow_sets)[0].data.get(CONF_DB_URL))
+
+    else:
+        flow_status = hass.config_entries.options.async_get(msg["flow_id"])
+        config_entry = hass.config_entries.async_get_entry(flow_status["handler"])
+        if not config_entry:
+            raise HomeAssistantError("Config entry not found")
+        name = config_entry.title
+        db_url = resolve_db_url(hass, config_entry.data.get(CONF_DB_URL))
+
+    @callback
+    def async_preview_updated(state: str, attributes: Mapping[str, Any]) -> None:
+        """Forward config entry state events to websocket."""
+        connection.send_message(
+            websocket_api.event_message(
+                msg["id"], {"attributes": attributes, "state": state}
+            )
+        )
+
+    (
+        sessmaker,
+        _,
+        use_database_executor,
+    ) = await async_create_sessionmaker(hass, db_url)
+    print(sessmaker)
+    print(db_url)
+    if sessmaker is None:
+        # Can not just return, needs to pass something
+        return
+
+    name_template = Template(name, hass)
+    trigger_entity_config = {CONF_NAME: name_template}
+    for key in TRIGGER_ENTITY_OPTIONS:
+        if key in msg["user_input"]:
+            trigger_entity_config[key] = msg["user_input"][key]
+        if key in msg["user_input"].get(CONF_ADVANCED_OPTIONS, {}):
+            trigger_entity_config[key] = msg["user_input"][CONF_ADVANCED_OPTIONS][key]
+
+    query_str: str = msg["user_input"].get(CONF_QUERY)
+    template: str | None = msg["user_input"].get(CONF_VALUE_TEMPLATE)
+    column_name: str = msg["user_input"].get(CONF_COLUMN_NAME)
+
+    value_template: ValueTemplate | None = None
+    if template is not None:
+        try:
+            value_template = ValueTemplate(template, hass)
+            value_template.ensure_valid()
+        except TemplateError:
+            value_template = None
+
+    preview_entity = SQLSensor(
+        trigger_entity_config=trigger_entity_config,
+        sessmaker=sessmaker,
+        query=ValueTemplate(query_str, hass),
+        column=column_name,
+        value_template=value_template,
+        yaml=False,
+        use_database_executor=use_database_executor,
+    )
+    preview_entity.hass = hass
+
+    # Create PlatformData, needed for name translations
+    platform_data = PlatformData(hass=hass, domain=SENSOR_DOMAIN, platform_name=DOMAIN)
+    await platform_data.async_load_translations()
+
+    connection.send_result(msg["id"])
+    connection.subscriptions[msg["id"]] = await preview_entity.async_start_preview(
+        async_preview_updated
+    )

homeassistant/components/sql/sensor.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable, Mapping
 import logging
 from typing import Any
 
@@ -21,7 +22,7 @@ from homeassistant.const import (
     CONF_VALUE_TEMPLATE,
     MATCH_ALL,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant
 from homeassistant.exceptions import PlatformNotReady, TemplateError
 from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
 from homeassistant.helpers.entity_platform import (
@@ -235,6 +236,7 @@ class SQLSensor(ManualTriggerSensorEntity):
                 manufacturer="SQL",
                 name=self._rendered.get(CONF_NAME),
             )
+        self._preview_callback: Callable[[str, Mapping[str, Any]], None] | None = None
 
     @property
     def name(self) -> str | None:
@@ -253,12 +255,32 @@ class SQLSensor(ManualTriggerSensorEntity):
         """Return extra attributes."""
         return dict(self._attr_extra_state_attributes)
 
+    async def async_start_preview(
+        self,
+        preview_callback: Callable[[str, Mapping[str, Any]], None],
+    ) -> CALLBACK_TYPE:
+        """Render a preview."""
+        # abort early if there is needed data missing
+        if not self._query or not self._column_name:
+            self._attr_available = False
+            calculated_state = self._async_calculate_state()
+            preview_callback(calculated_state.state, calculated_state.attributes)
+            return self._call_on_remove_callbacks
+
+        self._preview_callback = preview_callback
+
+        await self.async_update()
+        return self._call_on_remove_callbacks
+
     async def async_update(self) -> None:
         """Retrieve sensor data from the query using the right executor."""
         if self._use_database_executor:
             await get_instance(self.hass).async_add_executor_job(self._update)
         else:
             await self.hass.async_add_executor_job(self._update)
+        if self._preview_callback:
+            calculated_state = self._async_calculate_state()
+            self._preview_callback(calculated_state.state, calculated_state.attributes)
 
     def _update(self) -> None:
         """Retrieve sensor data from the query."""
@@ -297,6 +319,8 @@ class SQLSensor(ManualTriggerSensorEntity):
         if data is not None and isinstance(data, (bytes, bytearray)):
             data = f"0x{data.hex()}"
 
+        print(data, self._template)
+
         if data is not None and self._template is not None:
             variables = self._template_variables_with_value(data)
             if self._render_availability_template(variables):
@@ -305,6 +329,7 @@ class SQLSensor(ManualTriggerSensorEntity):
                 )
                 self._set_native_value_with_possible_timestamp(_value)
                 self._process_manual_data(variables)
+                print(self._attr_native_value)
         else:
             self._attr_native_value = data
 

tests/components/sql/conftest.py

@@ -3,11 +3,21 @@
 from __future__ import annotations
 
 from collections.abc import Generator
+import logging
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
 
+@pytest.hookimpl(tryfirst=True)
+def pytest_configure(config):
+    """Configure pytest."""
+    logger = logging.getLogger("sqlalchemy.engine")
+    logger.setLevel(logging.CRITICAL)
+    logger2 = logging.getLogger("homeassistant.components.recorder")
+    logger2.setLevel(logging.CRITICAL)
+
+
 @pytest.fixture
 def mock_setup_entry() -> Generator[AsyncMock]:
     """Override async_setup_entry."""

tests/components/sql/snapshots/test_config_flow.ambr

@@ -0,0 +1,80 @@
+# serializer version: 1
+# name: test_config_flow_preview[incorrect_column]
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Get Value',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+    }),
+    'state': 'unknown',
+  })
+# ---
+# name: test_config_flow_preview[missing_column]
+  dict({
+    'attributes': dict({
+      'friendly_name': 'Get Value',
+    }),
+    'state': 'unknown',
+  })
+# ---
+# name: test_config_flow_preview[success]
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Get Value',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+      'value': 5,
+    }),
+    'state': '5',
+  })
+# ---
+# name: test_config_flow_preview[with_value_template]
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Get Value',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+      'value': 5,
+    }),
+    'state': '5',
+  })
+# ---
+# name: test_config_flow_preview[with_value_template_invalid]
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Get Value',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+      'value': 5,
+    }),
+    'state': '5',
+  })
+# ---
+# name: test_config_flow_preview_no_database
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Get Value',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+      'value': 5,
+    }),
+    'state': '5',
+  })
+# ---
+# name: test_options_flow_preview
+  dict({
+    'attributes': dict({
+      'device_class': 'data_size',
+      'friendly_name': 'Select value SQL query',
+      'state_class': 'total',
+      'unit_of_measurement': 'MiB',
+      'value': 6,
+    }),
+    'state': '6',
+  })
+# ---

tests/components/sql/test_config_flow.py

@@ -5,13 +5,14 @@ from __future__ import annotations
 from pathlib import Path
 import re
 from typing import Any
-from unittest.mock import patch
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from sqlalchemy.exc import SQLAlchemyError
+from syrupy.assertion import SnapshotAssertion
 
 from homeassistant import config_entries
-from homeassistant.components.recorder import CONF_DB_URL, Recorder
+from homeassistant.components.recorder import CONF_DB_URL
 from homeassistant.components.sensor import (
     CONF_STATE_CLASS,
     SensorDeviceClass,
@@ -53,11 +54,13 @@ from . import (
     ENTRY_CONFIG_WITH_BROKEN_QUERY_TEMPLATE_OPT,
     ENTRY_CONFIG_WITH_QUERY_TEMPLATE,
     ENTRY_CONFIG_WITH_VALUE_TEMPLATE,
+    init_integration,
 )
 
 from tests.common import MockConfigEntry
+from tests.typing import WebSocketGenerator
 
-pytestmark = pytest.mark.usefixtures("mock_setup_entry", "recorder_mock")
+pytestmark = pytest.mark.usefixtures("recorder_mock")
 
 DATA_CONFIG = {CONF_NAME: "Get Value"}
 DATA_CONFIG_DB = {CONF_NAME: "Get Value", CONF_DB_URL: "sqlite://"}
@@ -72,6 +75,7 @@ OPTIONS_DATA_CONFIG = {}
     ],
 )
 async def test_form_simple(
+    mock_setup_entry: AsyncMock,
     hass: HomeAssistant,
     data_config: dict[str, Any],
     result_config: dict[str, Any],
@@ -110,9 +114,7 @@ async def test_form_simple(
     }
 
 
-async def test_form_with_query_template(
-    recorder_mock: Recorder, hass: HomeAssistant
-) -> None:
+async def test_form_with_query_template(hass: HomeAssistant) -> None:
     """Test for with query template."""
 
     result = await hass.config_entries.flow.async_init(
@@ -148,9 +150,7 @@ async def test_form_with_query_template(
     assert len(mock_setup_entry.mock_calls) == 1
 
 
-async def test_form_with_broken_query_template(
-    recorder_mock: Recorder, hass: HomeAssistant
-) -> None:
+async def test_form_with_broken_query_template(hass: HomeAssistant) -> None:
     """Test form with broken query template."""
 
     result = await hass.config_entries.flow.async_init(
@@ -193,7 +193,7 @@ async def test_form_with_broken_query_template(
 
 
 async def test_form_with_value_template(
-    recorder_mock: Recorder, hass: HomeAssistant
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
 ) -> None:
     """Test for with value template."""
 
@@ -228,7 +228,9 @@ async def test_form_with_value_template(
     }
 
 
-async def test_flow_fails_db_url(hass: HomeAssistant) -> None:
+async def test_flow_fails_db_url(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test config flow fails incorrect db url."""
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -249,7 +251,9 @@ async def test_flow_fails_db_url(hass: HomeAssistant) -> None:
     assert result["errors"] == {CONF_DB_URL: "db_url_invalid"}
 
 
-async def test_flow_fails_invalid_query(hass: HomeAssistant) -> None:
+async def test_flow_fails_invalid_query(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test config flow fails incorrect db url."""
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -352,7 +356,9 @@ async def test_flow_fails_invalid_query(hass: HomeAssistant) -> None:
     }
 
 
-async def test_flow_fails_invalid_column_name(hass: HomeAssistant) -> None:
+async def test_flow_fails_invalid_column_name(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test config flow fails invalid column name."""
     result = await hass.config_entries.flow.async_init(
         DOMAIN, context={"source": config_entries.SOURCE_USER}
@@ -395,7 +401,7 @@ async def test_flow_fails_invalid_column_name(hass: HomeAssistant) -> None:
     }
 
 
-async def test_options_flow(hass: HomeAssistant) -> None:
+async def test_options_flow(mock_setup_entry: AsyncMock, hass: HomeAssistant) -> None:
     """Test options config flow."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -448,7 +454,9 @@ async def test_options_flow(hass: HomeAssistant) -> None:
     }
 
 
-async def test_options_flow_name_previously_removed(hass: HomeAssistant) -> None:
+async def test_options_flow_name_previously_removed(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test options config flow where the name was missing."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -495,7 +503,9 @@ async def test_options_flow_name_previously_removed(hass: HomeAssistant) -> None
     }
 
 
-async def test_options_flow_fails_db_url(hass: HomeAssistant) -> None:
+async def test_options_flow_fails_db_url(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test options flow fails incorrect db url."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -534,7 +544,9 @@ async def test_options_flow_fails_db_url(hass: HomeAssistant) -> None:
     assert result["errors"] == {CONF_DB_URL: "db_url_invalid"}
 
 
-async def test_options_flow_fails_invalid_query(hass: HomeAssistant) -> None:
+async def test_options_flow_fails_invalid_query(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test options flow fails incorrect query and template."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -642,7 +654,9 @@ async def test_options_flow_fails_invalid_query(hass: HomeAssistant) -> None:
     }
 
 
-async def test_options_flow_fails_invalid_column_name(hass: HomeAssistant) -> None:
+async def test_options_flow_fails_invalid_column_name(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test options flow fails invalid column name."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -694,7 +708,9 @@ async def test_options_flow_fails_invalid_column_name(hass: HomeAssistant) -> No
     }
 
 
-async def test_options_flow_db_url_empty(hass: HomeAssistant) -> None:
+async def test_options_flow_db_url_empty(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test options config flow with leaving db_url empty."""
     entry = MockConfigEntry(
         domain=DOMAIN,
@@ -741,6 +757,7 @@ async def test_options_flow_db_url_empty(hass: HomeAssistant) -> None:
 
 
 async def test_full_flow_not_recorder_db(
+    mock_setup_entry: AsyncMock,
     hass: HomeAssistant,
     tmp_path: Path,
 ) -> None:
@@ -809,7 +826,9 @@ async def test_full_flow_not_recorder_db(
     }
 
 
-async def test_device_state_class(hass: HomeAssistant) -> None:
+async def test_device_state_class(
+    mock_setup_entry: AsyncMock, hass: HomeAssistant
+) -> None:
     """Test we get the form."""
 
     entry = MockConfigEntry(
@@ -881,3 +900,243 @@ async def test_device_state_class(hass: HomeAssistant) -> None:
             CONF_UNIT_OF_MEASUREMENT: "MiB",
         },
     }
+
+
+@pytest.mark.parametrize(
+    "user_input",
+    [
+        (
+            {
+                CONF_NAME: "Get Value",
+                CONF_QUERY: "SELECT 5 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_UNIT_OF_MEASUREMENT: "MiB",
+                CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                CONF_STATE_CLASS: SensorStateClass.TOTAL,
+            }
+        ),
+        (
+            {
+                CONF_NAME: "Get Value",
+                CONF_QUERY: "SELECT 5 as value",
+                CONF_COLUMN_NAME: "state",
+                CONF_UNIT_OF_MEASUREMENT: "MiB",
+                CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                CONF_STATE_CLASS: SensorStateClass.TOTAL,
+            }
+        ),
+        (
+            {
+                CONF_NAME: "Get Value",
+                CONF_QUERY: "SELECT 5 as value",
+            }
+        ),
+        (
+            {
+                CONF_NAME: "Get Value",
+                CONF_QUERY: "SELECT 5 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_VALUE_TEMPLATE: "{{ value }}",
+                CONF_UNIT_OF_MEASUREMENT: "MiB",
+                CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                CONF_STATE_CLASS: SensorStateClass.TOTAL,
+            }
+        ),
+        (
+            {
+                CONF_NAME: "Get Value",
+                CONF_QUERY: "SELECT 5 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_VALUE_TEMPLATE: "{{ value",
+                CONF_UNIT_OF_MEASUREMENT: "MiB",
+                CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                CONF_STATE_CLASS: SensorStateClass.TOTAL,
+            }
+        ),
+    ],
+    ids=(
+        "success",
+        "incorrect_column",
+        "missing_column",
+        "with_value_template",
+        "with_value_template_invalid",
+    ),
+)
+async def test_config_flow_preview(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    user_input: str,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the config flow preview."""
+    client = await hass_ws_client(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+    assert result["preview"] == "sql"
+
+    await client.send_json_auto_id(
+        {
+            "type": "sql/start_preview",
+            "flow_id": result["flow_id"],
+            "flow_type": "config_flow",
+            "user_input": user_input,
+        }
+    )
+    msg = await client.receive_json()
+    assert msg["success"]
+    assert msg["result"] is None
+
+    msg = await client.receive_json()
+    assert msg["event"] == snapshot
+    assert len(hass.states.async_all()) == 0
+
+
+async def test_config_flow_preview_no_database(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the config flow preview with no database."""
+    client = await hass_ws_client(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": config_entries.SOURCE_USER}
+    )
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {}
+
+    with patch("homeassistant.components.sql.config_flow.validate_db_connection"):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DB_URL: "sqlite://not_exist.local", CONF_NAME: "Get Value"},
+        )
+        await hass.async_block_till_done()
+
+    assert result["type"] == FlowResultType.FORM
+    assert result["step_id"] == "options"
+    assert result["errors"] == {}
+    assert result["preview"] == "sql"
+
+    await client.send_json_auto_id(
+        {
+            "type": "sql/start_preview",
+            "flow_id": result["flow_id"],
+            "flow_type": "config_flow",
+            "user_input": {
+                CONF_QUERY: "SELECT 5 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_ADVANCED_OPTIONS: {
+                    CONF_UNIT_OF_MEASUREMENT: "MiB",
+                    CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                    CONF_STATE_CLASS: SensorStateClass.TOTAL,
+                },
+            },
+        }
+    )
+    msg = await client.receive_json()
+    assert msg["success"]
+    assert msg["result"] is None
+
+    msg = await client.receive_json()
+    print(msg)
+    assert msg["event"] == snapshot
+    assert False
+
+
+async def test_options_flow_preview(
+    hass: HomeAssistant,
+    hass_ws_client: WebSocketGenerator,
+    snapshot: SnapshotAssertion,
+) -> None:
+    """Test the options flow preview."""
+    client = await hass_ws_client(hass)
+
+    # Setup the config entry
+    config_entry = await init_integration(hass)
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+    assert result["type"] == FlowResultType.FORM
+    assert result["preview"] == "sql"
+
+    await client.send_json_auto_id(
+        {
+            "type": "sql/start_preview",
+            "flow_id": result["flow_id"],
+            "flow_type": "options_flow",
+            "user_input": {
+                CONF_QUERY: "SELECT 6 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_ADVANCED_OPTIONS: {
+                    CONF_UNIT_OF_MEASUREMENT: "MiB",
+                    CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                    CONF_STATE_CLASS: SensorStateClass.TOTAL,
+                },
+            },
+        }
+    )
+
+    msg = await client.receive_json()
+    assert msg["success"]
+    assert msg["result"] is None
+
+    msg = await client.receive_json()
+    assert msg["event"] == snapshot
+    assert len(hass.states.async_all()) == 1
+
+
+async def test_options_flow_sensor_preview_config_entry_removed(
+    hass: HomeAssistant, hass_ws_client: WebSocketGenerator
+) -> None:
+    """Test the option flow preview where the config entry is removed."""
+    client = await hass_ws_client(hass)
+
+    # Setup the config entry
+    config_entry = MockConfigEntry(
+        data={},
+        domain=DOMAIN,
+        options={
+            CONF_NAME: "Get Value",
+            CONF_QUERY: "SELECT 5 as value",
+            CONF_COLUMN_NAME: "value",
+            CONF_UNIT_OF_MEASUREMENT: "MiB",
+            CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+            CONF_STATE_CLASS: SensorStateClass.TOTAL,
+        },
+        title="Get Value",
+    )
+    config_entry.add_to_hass(hass)
+    assert await hass.config_entries.async_setup(config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+    assert result["type"] == FlowResultType.FORM
+    assert result["preview"] == "sql"
+
+    await hass.config_entries.async_remove(config_entry.entry_id)
+
+    await client.send_json_auto_id(
+        {
+            "type": "sql/start_preview",
+            "flow_id": result["flow_id"],
+            "flow_type": "options_flow",
+            "user_input": {
+                CONF_QUERY: "SELECT 6 as value",
+                CONF_COLUMN_NAME: "value",
+                CONF_UNIT_OF_MEASUREMENT: "MiB",
+                CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
+                CONF_STATE_CLASS: SensorStateClass.TOTAL,
+            },
+        }
+    )
+    msg = await client.receive_json()
+    assert not msg["success"]
+    assert msg["error"] == {
+        "code": "home_assistant_error",
+        "message": "Config entry not found",
+    }