Check exception causes for matching strings during recorder migration (#49999)

homeassistant/components/recorder/migration.py

@@ -17,6 +17,17 @@ from .util import session_scope
 _LOGGER = logging.getLogger(__name__)
 
 
+def raise_if_exception_missing_str(ex, match_substrs):
+    """Raise an exception if the exception and cause do not contain the match substrs."""
+    lower_ex_strs = [str(ex).lower(), str(ex.__cause__).lower()]
+    for str_sub in match_substrs:
+        for exc_str in lower_ex_strs:
+            if exc_str and str_sub in exc_str:
+                return
+
+    raise ex
+
+
 def get_schema_version(instance):
     """Get the schema version."""
     with session_scope(session=instance.get_session()) as session:
@@ -80,11 +91,7 @@ def _create_index(connection, table_name, index_name):
     try:
         index.create(connection)
     except (InternalError, ProgrammingError, OperationalError) as err:
-        lower_err_str = str(err).lower()
-
-        if "already exists" not in lower_err_str and "duplicate" not in lower_err_str:
-            raise
-
+        raise_if_exception_missing_str(err, ["already exists", "duplicate"])
         _LOGGER.warning(
             "Index %s already exists on %s, continuing", index_name, table_name
         )
@@ -199,9 +206,7 @@ def _add_columns(connection, table_name, columns_def):
                 )
             )
         except (InternalError, OperationalError) as err:
-            if "duplicate" not in str(err).lower():
-                raise
-
+            raise_if_exception_missing_str(err, ["duplicate"])
             _LOGGER.warning(
                 "Column %s already exists on %s, continuing",
                 column_def.split(" ")[1],

tests/components/recorder/test_migrate.py

@@ -326,3 +326,36 @@ def test_forgiving_add_index_with_other_db_types(caplog, exception_type):
 
     assert "already exists on states" in caplog.text
     assert "continuing" in caplog.text
+
+
+class MockPyODBCProgrammingError(Exception):
+    """A mock pyodbc error."""
+
+
+def test_raise_if_exception_missing_str():
+    """Test we raise an exception if strings are not present."""
+    programming_exc = ProgrammingError("select * from;", Mock(), Mock())
+    programming_exc.__cause__ = MockPyODBCProgrammingError(
+        "[42S11] [FreeTDS][SQL Server]The operation failed because an index or statistics with name 'ix_states_old_state_id' already exists on table 'states'. (1913) (SQLExecDirectW)"
+    )
+
+    migration.raise_if_exception_missing_str(
+        programming_exc, ["already exists", "duplicate"]
+    )
+
+    with pytest.raises(ProgrammingError):
+        migration.raise_if_exception_missing_str(programming_exc, ["not present"])
+
+
+def test_raise_if_exception_missing_empty_cause_str():
+    """Test we raise an exception if strings are not present with an empty cause."""
+    programming_exc = ProgrammingError("select * from;", Mock(), Mock())
+    programming_exc.__cause__ = MockPyODBCProgrammingError()
+
+    with pytest.raises(ProgrammingError):
+        migration.raise_if_exception_missing_str(
+            programming_exc, ["already exists", "duplicate"]
+        )
+
+    with pytest.raises(ProgrammingError):
+        migration.raise_if_exception_missing_str(programming_exc, ["not present"])