home-assistant/core
1
0
Fork 1
mirror of https://github.com/home-assistant/core.git synced 2025-09-05 12:51:37 +02:00
Code Issues Packages Projects Releases Wiki Activity

Mod tests

Browse Source
This commit is contained in:
G Johansson
2025-08-16 15:22:16 +00:00
parent 54940bced0
commit 26d5e33cdf
3 changed files with 148 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
tests/components/sql/__init__.py
View File

@@ -292,22 +292,29 @@ YAML_CONFIG_ALL_TEMPLATES = {
} }
async def init_integration( async def init_integration( # pylint: disable=W0102
hass: HomeAssistant, hass: HomeAssistant,
config: dict[str, Any] | None = None, *,
title: str = "Select value SQL query",
config: dict[str, Any] = {},
options: dict[str, Any] | None = None,
entry_id: str = "1", entry_id: str = "1",
source: str = SOURCE_USER, source: str = SOURCE_USER,
) -> MockConfigEntry: ) -> MockConfigEntry:
"""Set up the SQL integration in Home Assistant.""" """Set up the SQL integration in Home Assistant."""
if not config: if not options:
config = ENTRY_CONFIG options = ENTRY_CONFIG
if CONF_ADVANCED_OPTIONS not in options:
options[CONF_ADVANCED_OPTIONS] = {}
config_entry = MockConfigEntry( config_entry = MockConfigEntry(
title=title,
domain=DOMAIN, domain=DOMAIN,
source=source, source=source,
data={}, data=config,
options=config, options=options,
entry_id=entry_id, entry_id=entry_id,
version=2,
) )
config_entry.add_to_hass(hass) config_entry.add_to_hass(hass)

39
tests/components/sql/test_init.py
View File

@@ -8,7 +8,6 @@ import pytest
import voluptuous as vol import voluptuous as vol
from homeassistant.components.recorder import Recorder from homeassistant.components.recorder import Recorder
from homeassistant.components.recorder.util import get_instance
from homeassistant.components.sql import validate_sql_select from homeassistant.components.sql import validate_sql_select
from homeassistant.components.sql.const import DOMAIN from homeassistant.components.sql.const import DOMAIN
from homeassistant.config_entries import ConfigEntryState from homeassistant.config_entries import ConfigEntryState
@@ -84,41 +83,3 @@ async def test_multiple_queries(hass: HomeAssistant) -> None:
"""Test multiple queries.""" """Test multiple queries."""
with pytest.raises(vol.Invalid): with pytest.raises(vol.Invalid):
validate_sql_select("SELECT 5 as value; UPDATE states SET state = 10;") validate_sql_select("SELECT 5 as value; UPDATE states SET state = 10;")
async def test_remove_configured_db_url_if_not_needed_when_not_needed(
recorder_mock: Recorder,
hass: HomeAssistant,
) -> None:
"""Test configured db_url is replaced with None if matching the recorder db."""
recorder_db_url = get_instance(hass).db_url
config = {
"db_url": recorder_db_url,
"query": "SELECT 5 as value",
"column": "value",
"name": "count_tables",
}
config_entry = await init_integration(hass, config)
assert config_entry.options.get("db_url") is None
async def test_remove_configured_db_url_if_not_needed_when_needed(
recorder_mock: Recorder,
hass: HomeAssistant,
) -> None:
"""Test configured db_url is not replaced if it differs from the recorder db."""
db_url = "mssql://"
config = {
"db_url": db_url,
"query": "SELECT 5 as value",
"column": "value",
"name": "count_tables",
}
config_entry = await init_integration(hass, config)
assert config_entry.options.get("db_url") == db_url

256
tests/components/sql/test_sensor.py
View File

@@ -12,14 +12,27 @@ from freezegun.api import FrozenDateTimeFactory
import pytest import pytest
from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.exc import SQLAlchemyError
from homeassistant.components.recorder import Recorder from homeassistant.components.recorder import CONF_DB_URL, Recorder
from homeassistant.components.sensor import SensorDeviceClass, SensorStateClass from homeassistant.components.sensor import (
from homeassistant.components.sql.const import CONF_QUERY, DOMAIN CONF_STATE_CLASS,
SensorDeviceClass,
SensorStateClass,
)
from homeassistant.components.sql.const import (
CONF_ADVANCED_OPTIONS,
CONF_COLUMN_NAME,
CONF_QUERY,
DOMAIN,
)
from homeassistant.components.sql.sensor import _generate_lambda_stmt from homeassistant.components.sql.sensor import _generate_lambda_stmt
from homeassistant.config_entries import SOURCE_USER from homeassistant.config_entries import SOURCE_USER
from homeassistant.const import ( from homeassistant.const import (
CONF_DEVICE_CLASS,
CONF_ICON, CONF_ICON,
CONF_NAME,
CONF_UNIQUE_ID, CONF_UNIQUE_ID,
CONF_UNIT_OF_MEASUREMENT,
CONF_VALUE_TEMPLATE,
STATE_UNAVAILABLE, STATE_UNAVAILABLE,
STATE_UNKNOWN, STATE_UNKNOWN,
UnitOfInformation, UnitOfInformation,
@@ -37,7 +50,6 @@ from . import (
YAML_CONFIG_FULL_TABLE_SCAN, YAML_CONFIG_FULL_TABLE_SCAN,
YAML_CONFIG_FULL_TABLE_SCAN_NO_UNIQUE_ID, YAML_CONFIG_FULL_TABLE_SCAN_NO_UNIQUE_ID,
YAML_CONFIG_FULL_TABLE_SCAN_WITH_MULTIPLE_COLUMNS, YAML_CONFIG_FULL_TABLE_SCAN_WITH_MULTIPLE_COLUMNS,
YAML_CONFIG_WITH_VIEW_THAT_CONTAINS_ENTITY_ID,
init_integration, init_integration,
) )
@@ -46,14 +58,11 @@ from tests.common import MockConfigEntry, async_fire_time_changed
async def test_query_basic(recorder_mock: Recorder, hass: HomeAssistant) -> None: async def test_query_basic(recorder_mock: Recorder, hass: HomeAssistant) -> None:
"""Test the SQL sensor.""" """Test the SQL sensor."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "Select value SQL query",
"unique_id": "very_unique_id",
} }
await init_integration(hass, config) await init_integration(hass, title="Select value SQL query", options=options)
state = hass.states.get("sensor.select_value_sql_query") state = hass.states.get("sensor.select_value_sql_query")
assert state.state == "5" assert state.state == "5"
@@ -62,14 +71,11 @@ async def test_query_basic(recorder_mock: Recorder, hass: HomeAssistant) -> None
async def test_query_cte(recorder_mock: Recorder, hass: HomeAssistant) -> None: async def test_query_cte(recorder_mock: Recorder, hass: HomeAssistant) -> None:
"""Test the SQL sensor with CTE.""" """Test the SQL sensor with CTE."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "WITH test AS (SELECT 1 AS row_num, 10 AS state) SELECT state FROM test WHERE row_num = 1 LIMIT 1;",
"query": "WITH test AS (SELECT 1 AS row_num, 10 AS state) SELECT state FROM test WHERE row_num = 1 LIMIT 1;", CONF_COLUMN_NAME: "state",
"column": "state",
"name": "Select value SQL query CTE",
"unique_id": "very_unique_id",
} }
await init_integration(hass, config) await init_integration(hass, title="Select value SQL query CTE", options=options)
state = hass.states.get("sensor.select_value_sql_query_cte") state = hass.states.get("sensor.select_value_sql_query_cte")
assert state.state == "10" assert state.state == "10"
@@ -80,31 +86,39 @@ async def test_query_value_template(
recorder_mock: Recorder, hass: HomeAssistant recorder_mock: Recorder, hass: HomeAssistant
) -> None: ) -> None:
"""Test the SQL sensor.""" """Test the SQL sensor."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5.01 as value",
"query": "SELECT 5.01 as value", CONF_COLUMN_NAME: "value",
"column": "value", CONF_ADVANCED_OPTIONS: {
"name": "count_tables", CONF_VALUE_TEMPLATE: "{{ value | int }}",
"value_template": "{{ value | int }}", CONF_UNIT_OF_MEASUREMENT: "MiB",
CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
CONF_STATE_CLASS: SensorStateClass.MEASUREMENT,
},
} }
await init_integration(hass, config) await init_integration(hass, title="count_tables", options=options)
state = hass.states.get("sensor.count_tables") state = hass.states.get("sensor.count_tables")
assert state.state == "5" assert state.state == "5"
assert state.attributes == {
"device_class": "data_size",
"friendly_name": "count_tables",
"state_class": "measurement",
"unit_of_measurement": "MiB",
"value": 5.01,
}
async def test_query_value_template_invalid( async def test_query_value_template_invalid(
recorder_mock: Recorder, hass: HomeAssistant recorder_mock: Recorder, hass: HomeAssistant
) -> None: ) -> None:
"""Test the SQL sensor.""" """Test the SQL sensor."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5.01 as value",
"query": "SELECT 5.01 as value", CONF_COLUMN_NAME: "value",
"column": "value", CONF_VALUE_TEMPLATE: "{{ value | dontwork }}",
"name": "count_tables",
"value_template": "{{ value | dontwork }}",
} }
await init_integration(hass, config) await init_integration(hass, title="count_tables", options=options)
state = hass.states.get("sensor.count_tables") state = hass.states.get("sensor.count_tables")
assert state.state == "5.01" assert state.state == "5.01"
@@ -112,13 +126,11 @@ async def test_query_value_template_invalid(
async def test_query_limit(recorder_mock: Recorder, hass: HomeAssistant) -> None: async def test_query_limit(recorder_mock: Recorder, hass: HomeAssistant) -> None:
"""Test the SQL sensor with a query containing 'LIMIT' in lowercase.""" """Test the SQL sensor with a query containing 'LIMIT' in lowercase."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5 as value limit 1",
"query": "SELECT 5 as value limit 1", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "Select value SQL query",
} }
await init_integration(hass, config) await init_integration(hass, options=options)
state = hass.states.get("sensor.select_value_sql_query") state = hass.states.get("sensor.select_value_sql_query")
assert state.state == "5" assert state.state == "5"
@@ -129,13 +141,11 @@ async def test_query_no_value(
recorder_mock: Recorder, hass: HomeAssistant, caplog: pytest.LogCaptureFixture recorder_mock: Recorder, hass: HomeAssistant, caplog: pytest.LogCaptureFixture
) -> None: ) -> None:
"""Test the SQL sensor with a query that returns no value.""" """Test the SQL sensor with a query that returns no value."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5 as value where 1=2",
"query": "SELECT 5 as value where 1=2", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "count_tables",
} }
await init_integration(hass, config) await init_integration(hass, title="count_tables", options=options)
state = hass.states.get("sensor.count_tables") state = hass.states.get("sensor.count_tables")
assert state.state == STATE_UNKNOWN assert state.state == STATE_UNKNOWN
@@ -163,13 +173,13 @@ async def test_query_on_disk_sqlite_no_result(
await hass.async_add_executor_job(make_test_db) await hass.async_add_executor_job(make_test_db)
config = { config = {CONF_DB_URL: db_path_str}
"db_url": db_path_str, options = {
"query": "SELECT value from users", CONF_QUERY: "SELECT value from users",
"column": "value", CONF_COLUMN_NAME: "value",
"name": "count_users", CONF_NAME: "count_users",
} }
await init_integration(hass, config) await init_integration(hass, title="count_users", options=options, config=config)
state = hass.states.get("sensor.count_users") state = hass.states.get("sensor.count_users")
assert state.state == STATE_UNKNOWN assert state.state == STATE_UNKNOWN
@@ -203,17 +213,17 @@ async def test_invalid_url_setup(
) -> None: ) -> None:
"""Test invalid db url with redacted credentials.""" """Test invalid db url with redacted credentials."""
config = { config = {
"db_url": url, CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "count_tables",
} }
entry = MockConfigEntry( entry = MockConfigEntry(
title="count_tables",
domain=DOMAIN, domain=DOMAIN,
source=SOURCE_USER, source=SOURCE_USER,
data={}, data={CONF_DB_URL: url},
options=config, options=config,
entry_id="1", entry_id="1",
version=2,
) )
entry.add_to_hass(hass) entry.add_to_hass(hass)
@@ -237,11 +247,9 @@ async def test_invalid_url_on_update(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test invalid db url with redacted credentials on retry.""" """Test invalid db url with redacted credentials on retry."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "count_tables",
} }
class MockSession: class MockSession:
@@ -255,7 +263,7 @@ async def test_invalid_url_on_update(
"homeassistant.components.sql.sensor.scoped_session", "homeassistant.components.sql.sensor.scoped_session",
return_value=MockSession, return_value=MockSession,
): ):
await init_integration(hass, config) await init_integration(hass, title="count_tables", options=options)
async_fire_time_changed( async_fire_time_changed(
hass, hass,
dt_util.utcnow() + timedelta(minutes=1), dt_util.utcnow() + timedelta(minutes=1),
@@ -343,12 +351,12 @@ async def test_config_from_old_yaml(
config = { config = {
"sensor": { "sensor": {
"platform": "sql", "platform": "sql",
"db_url": "sqlite://", CONF_DB_URL: "sqlite://",
"queries": [ "queries": [
{ {
"name": "count_tables", CONF_NAME: "count_tables",
"query": "SELECT 5 as value", CONF_QUERY: "SELECT 5 as value",
"column": "value", CONF_COLUMN_NAME: "value",
} }
], ],
} }
@@ -386,10 +394,10 @@ async def test_invalid_url_setup_from_yaml(
"""Test invalid db url with redacted credentials from yaml setup.""" """Test invalid db url with redacted credentials from yaml setup."""
config = { config = {
"sql": { "sql": {
"db_url": url, CONF_DB_URL: url,
"query": "SELECT 5 as value", CONF_QUERY: "SELECT 5 as value",
"column": "value", CONF_COLUMN_NAME: "value",
"name": "count_tables", CONF_NAME: "count_tables",
} }
} }
@@ -417,9 +425,9 @@ async def test_attributes_from_yaml_setup(
state = hass.states.get("sensor.get_value") state = hass.states.get("sensor.get_value")
assert state.state == "5" assert state.state == "5"
assert state.attributes["device_class"] == SensorDeviceClass.DATA_SIZE assert state.attributes[CONF_DEVICE_CLASS] == SensorDeviceClass.DATA_SIZE
assert state.attributes["state_class"] == SensorStateClass.MEASUREMENT assert state.attributes[CONF_STATE_CLASS] == SensorStateClass.MEASUREMENT
assert state.attributes["unit_of_measurement"] == UnitOfInformation.MEBIBYTES assert state.attributes[CONF_UNIT_OF_MEASUREMENT] == UnitOfInformation.MEBIBYTES
async def test_binary_data_from_yaml_setup( async def test_binary_data_from_yaml_setup(
@@ -455,7 +463,7 @@ async def test_issue_when_using_old_query(
issue = issue_registry.async_get_issue( issue = issue_registry.async_get_issue(
DOMAIN, f"entity_id_query_does_full_table_scan_{unique_id}" DOMAIN, f"entity_id_query_does_full_table_scan_{unique_id}"
) )
assert issue.translation_placeholders == {"query": config[CONF_QUERY]} assert issue.translation_placeholders == {CONF_QUERY: config[CONF_QUERY]}
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -486,7 +494,7 @@ async def test_issue_when_using_old_query_without_unique_id(
issue = issue_registry.async_get_issue( issue = issue_registry.async_get_issue(
DOMAIN, f"entity_id_query_does_full_table_scan_{query}" DOMAIN, f"entity_id_query_does_full_table_scan_{query}"
) )
assert issue.translation_placeholders == {"query": query} assert issue.translation_placeholders == {CONF_QUERY: query}
async def test_no_issue_when_view_has_the_text_entity_id_in_it( async def test_no_issue_when_view_has_the_text_entity_id_in_it(
@@ -498,7 +506,12 @@ async def test_no_issue_when_view_has_the_text_entity_id_in_it(
"homeassistant.components.sql.sensor.scoped_session", "homeassistant.components.sql.sensor.scoped_session",
): ):
await init_integration( await init_integration(
hass, YAML_CONFIG_WITH_VIEW_THAT_CONTAINS_ENTITY_ID["sql"] hass,
title="Get entity_id",
options={
CONF_QUERY: "SELECT value from view_sensor_db_unique_entity_ids;",
CONF_COLUMN_NAME: "value",
},
) )
async_fire_time_changed( async_fire_time_changed(
hass, hass,
@@ -516,20 +529,18 @@ async def test_multiple_sensors_using_same_db(
recorder_mock: Recorder, hass: HomeAssistant recorder_mock: Recorder, hass: HomeAssistant
) -> None: ) -> None:
"""Test multiple sensors using the same db.""" """Test multiple sensors using the same db."""
config = { options = {
"db_url": "sqlite:///", CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "Select value SQL query",
} }
config2 = { options2 = {
"db_url": "sqlite:///", CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value",
"name": "Select value SQL query 2",
} }
await init_integration(hass, config) await init_integration(hass, title="Select value SQL query", options=options)
await init_integration(hass, config2, entry_id="2") await init_integration(
hass, title="Select value SQL query 2", options=options2, entry_id="2"
)
state = hass.states.get("sensor.select_value_sql_query") state = hass.states.get("sensor.select_value_sql_query")
assert state.state == "5" assert state.state == "5"
@@ -547,13 +558,14 @@ async def test_engine_is_disposed_at_stop(
recorder_mock: Recorder, hass: HomeAssistant recorder_mock: Recorder, hass: HomeAssistant
) -> None: ) -> None:
"""Test we dispose of the engine at stop.""" """Test we dispose of the engine at stop."""
config = { config = {CONF_DB_URL: "sqlite:///"}
"db_url": "sqlite:///", options = {
"query": "SELECT 5 as value", CONF_QUERY: "SELECT 5 as value",
"column": "value", CONF_COLUMN_NAME: "value",
"name": "Select value SQL query",
} }
await init_integration(hass, config) await init_integration(
hass, title="Select value SQL query", config=config, options=options
)
state = hass.states.get("sensor.select_value_sql_query") state = hass.states.get("sensor.select_value_sql_query")
assert state.state == "5" assert state.state == "5"
@@ -572,13 +584,15 @@ async def test_attributes_from_entry_config(
await init_integration( await init_integration(
hass, hass,
config={ title="Get Value - With",
"name": "Get Value - With", options={
"query": "SELECT 5 as value", CONF_QUERY: "SELECT 5 as value",
"column": "value", CONF_COLUMN_NAME: "value",
"unit_of_measurement": "MiB", CONF_ADVANCED_OPTIONS: {
"device_class": SensorDeviceClass.DATA_SIZE, CONF_UNIT_OF_MEASUREMENT: "MiB",
"state_class": SensorStateClass.TOTAL, CONF_DEVICE_CLASS: SensorDeviceClass.DATA_SIZE,
CONF_STATE_CLASS: SensorStateClass.TOTAL,
},
}, },
entry_id="8693d4782ced4fb1ecca4743f29ab8f1", entry_id="8693d4782ced4fb1ecca4743f29ab8f1",
) )
@@ -586,27 +600,29 @@ async def test_attributes_from_entry_config(
state = hass.states.get("sensor.get_value_with") state = hass.states.get("sensor.get_value_with")
assert state.state == "5" assert state.state == "5"
assert state.attributes["value"] == 5 assert state.attributes["value"] == 5
assert state.attributes["unit_of_measurement"] == "MiB" assert state.attributes[CONF_UNIT_OF_MEASUREMENT] == "MiB"
assert state.attributes["device_class"] == SensorDeviceClass.DATA_SIZE assert state.attributes[CONF_DEVICE_CLASS] == SensorDeviceClass.DATA_SIZE
assert state.attributes["state_class"] == SensorStateClass.TOTAL assert state.attributes[CONF_STATE_CLASS] == SensorStateClass.TOTAL
await init_integration( await init_integration(
hass, hass,
config={ title="Get Value - Without",
"name": "Get Value - Without", options={
"query": "SELECT 5 as value", CONF_QUERY: "SELECT 6 as value",
"column": "value", CONF_COLUMN_NAME: "value",
"unit_of_measurement": "MiB", CONF_ADVANCED_OPTIONS: {
CONF_UNIT_OF_MEASUREMENT: "MiB",
},
}, },
entry_id="7aec7cd8045fba4778bb0621469e3cd9", entry_id="7aec7cd8045fba4778bb0621469e3cd9",
) )
state = hass.states.get("sensor.get_value_without") state = hass.states.get("sensor.get_value_without")
assert state.state == "5" assert state.state == "6"
assert state.attributes["value"] == 5 assert state.attributes["value"] == 6
assert state.attributes["unit_of_measurement"] == "MiB" assert state.attributes[CONF_UNIT_OF_MEASUREMENT] == "MiB"
assert "device_class" not in state.attributes assert CONF_DEVICE_CLASS not in state.attributes
assert "state_class" not in state.attributes assert CONF_STATE_CLASS not in state.attributes
async def test_query_recover_from_rollback( async def test_query_recover_from_rollback(
@@ -616,14 +632,12 @@ async def test_query_recover_from_rollback(
caplog: pytest.LogCaptureFixture, caplog: pytest.LogCaptureFixture,
) -> None: ) -> None:
"""Test the SQL sensor.""" """Test the SQL sensor."""
config = { options = {
"db_url": "sqlite://", CONF_QUERY: "SELECT 5 as value",
"query": "SELECT 5 as value", CONF_COLUMN_NAME: "value",
"column": "value", CONF_UNIQUE_ID: "very_unique_id",
"name": "Select value SQL query",
"unique_id": "very_unique_id",
} }
await init_integration(hass, config) await init_integration(hass, title="Select value SQL query", options=options)
platforms = async_get_platforms(hass, "sql") platforms = async_get_platforms(hass, "sql")
sql_entity = platforms[0].entities["sensor.select_value_sql_query"] sql_entity = platforms[0].entities["sensor.select_value_sql_query"]
@@ -671,7 +685,7 @@ async def test_availability_blocks_value_template(
"""Test availability blocks value_template from rendering.""" """Test availability blocks value_template from rendering."""
error = "Error parsing value for sensor.get_value: 'x' is undefined" error = "Error parsing value for sensor.get_value: 'x' is undefined"
config = YAML_CONFIG config = YAML_CONFIG
config["sql"]["value_template"] = "{{ x - 0 }}" config["sql"][CONF_VALUE_TEMPLATE] = "{{ x - 0 }}"
config["sql"]["availability"] = '{{ states("sensor.input1")=="on" }}' config["sql"]["availability"] = '{{ states("sensor.input1")=="on" }}'
hass.states.async_set("sensor.input1", "off") hass.states.async_set("sensor.input1", "off")