Read the recorder database name with the SQLAlchemy URL parser

homeassistant/components/recorder/system_health/__init__.py

@@ -1,7 +1,8 @@
 """Provide info to system health."""
 
 from typing import Any
-from urllib.parse import urlparse
+
+from sqlalchemy.engine.url import make_url
 
 from homeassistant.components import system_health
 from homeassistant.core import HomeAssistant, callback
@@ -58,7 +59,7 @@ async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
     instance = get_instance(hass)
 
     recorder_runs_manager = instance.recorder_runs_manager
-    database_name = urlparse(instance.db_url).path.lstrip("/")
+    database_name = make_url(instance.db_url).database or ""
     db_engine_info = _async_get_db_engine_info(instance)
     db_stats: dict[str, Any] = {}
 

tests/components/recorder/test_system_health.py

@@ -128,3 +128,42 @@ async def test_recorder_system_health_crashed_recorder_runs_table(
         "database_engine": SupportedDialect.SQLITE.value,
         "database_version": ANY,
     }
+
+
+@pytest.mark.parametrize("db_engine", [SupportedDialect.POSTGRESQL])
+@pytest.mark.parametrize(
+    "db_url",
+    [
+        "postgresql://homeassistant:secret@192.168.0.2:5432/home_assistant",
+        "postgresql://homeassistant:pa#ss@192.168.0.2:5432/home_assistant",
+        "postgresql://homeassistant:pa?ss@192.168.0.2:5432/home_assistant",
+        "postgresql://homeassistant:pa/ss@192.168.0.2:5432/home_assistant",
+    ],
+    ids=["plain", "hash", "question_mark", "slash"],
+)
+@pytest.mark.usefixtures("recorder_mock")
+async def test_recorder_system_health_db_name_with_special_characters(
+    hass: HomeAssistant,
+    db_engine: SupportedDialect,
+    db_url: str,
+    recorder_dialect_name: None,
+) -> None:
+    """Test the database name is read correctly when the password needs escaping.
+
+    Characters that are structural in a generic URL, such as ``#`` and ``?``,
+    must not be allowed to swallow the database name that follows them.
+    """
+    assert await async_setup_component(hass, "system_health", {})
+    await async_wait_recording_done(hass)
+
+    instance = get_instance(hass)
+    with (
+        patch.object(instance, "db_url", db_url),
+        patch(
+            "sqlalchemy.orm.session.Session.execute",
+            return_value=Mock(scalar=Mock(return_value=("1048576"))),
+        ) as execute_mock,
+    ):
+        await get_system_health_info(hass, "recorder")
+
+    assert execute_mock.call_args.args[1] == {"database_name": "home_assistant"}