mirror of
https://github.com/home-assistant/core.git
synced 2025-04-24 09:17:53 +00:00
Clarify SQLite can't drop foreign key constraints (#123898)
This commit is contained in:
parent
ea7e88d000
commit
e050d187c4
@ -585,7 +585,18 @@ def _modify_columns(
|
||||
def _update_states_table_with_foreign_key_options(
|
||||
session_maker: Callable[[], Session], engine: Engine
|
||||
) -> None:
|
||||
"""Add the options to foreign key constraints."""
|
||||
"""Add the options to foreign key constraints.
|
||||
|
||||
This is not supported for SQLite because it does not support
|
||||
dropping constraints.
|
||||
"""
|
||||
|
||||
if engine.dialect.name not in (SupportedDialect.MYSQL, SupportedDialect.POSTGRESQL):
|
||||
raise RuntimeError(
|
||||
"_update_states_table_with_foreign_key_options not supported for "
|
||||
f"{engine.dialect.name}"
|
||||
)
|
||||
|
||||
inspector = sqlalchemy.inspect(engine)
|
||||
tmp_states_table = Table(TABLE_STATES, MetaData())
|
||||
alters = [
|
||||
@ -633,7 +644,17 @@ def _update_states_table_with_foreign_key_options(
|
||||
def _drop_foreign_key_constraints(
|
||||
session_maker: Callable[[], Session], engine: Engine, table: str, column: str
|
||||
) -> tuple[bool, list[tuple[str, str, ReflectedForeignKeyConstraint]]]:
|
||||
"""Drop foreign key constraints for a table on specific columns."""
|
||||
"""Drop foreign key constraints for a table on specific columns.
|
||||
|
||||
This is not supported for SQLite because it does not support
|
||||
dropping constraints.
|
||||
"""
|
||||
|
||||
if engine.dialect.name not in (SupportedDialect.MYSQL, SupportedDialect.POSTGRESQL):
|
||||
raise RuntimeError(
|
||||
f"_drop_foreign_key_constraints not supported for {engine.dialect.name}"
|
||||
)
|
||||
|
||||
inspector = sqlalchemy.inspect(engine)
|
||||
dropped_constraints = [
|
||||
(table, column, foreign_key)
|
||||
@ -1026,7 +1047,17 @@ class _SchemaVersion11Migrator(_SchemaVersionMigrator, target_version=11):
|
||||
def _apply_update(self) -> None:
|
||||
"""Version specific update method."""
|
||||
_create_index(self.session_maker, "states", "ix_states_old_state_id")
|
||||
_update_states_table_with_foreign_key_options(self.session_maker, self.engine)
|
||||
|
||||
# _update_states_table_with_foreign_key_options first drops foreign
|
||||
# key constraints, and then re-adds them with the correct settings.
|
||||
# This is not supported by SQLite
|
||||
if self.engine.dialect.name in (
|
||||
SupportedDialect.MYSQL,
|
||||
SupportedDialect.POSTGRESQL,
|
||||
):
|
||||
_update_states_table_with_foreign_key_options(
|
||||
self.session_maker, self.engine
|
||||
)
|
||||
|
||||
|
||||
class _SchemaVersion12Migrator(_SchemaVersionMigrator, target_version=12):
|
||||
@ -1080,9 +1111,14 @@ class _SchemaVersion15Migrator(_SchemaVersionMigrator, target_version=15):
|
||||
class _SchemaVersion16Migrator(_SchemaVersionMigrator, target_version=16):
|
||||
def _apply_update(self) -> None:
|
||||
"""Version specific update method."""
|
||||
_drop_foreign_key_constraints(
|
||||
self.session_maker, self.engine, TABLE_STATES, "old_state_id"
|
||||
)
|
||||
# Dropping foreign key constraints is not supported by SQLite
|
||||
if self.engine.dialect.name in (
|
||||
SupportedDialect.MYSQL,
|
||||
SupportedDialect.POSTGRESQL,
|
||||
):
|
||||
_drop_foreign_key_constraints(
|
||||
self.session_maker, self.engine, TABLE_STATES, "old_state_id"
|
||||
)
|
||||
|
||||
|
||||
class _SchemaVersion17Migrator(_SchemaVersionMigrator, target_version=17):
|
||||
|
@ -1053,3 +1053,51 @@ def test_delete_foreign_key_violations_unsupported_engine(
|
||||
RuntimeError, match="_delete_foreign_key_violations not supported for sqlite"
|
||||
):
|
||||
migration._delete_foreign_key_violations(session_maker, engine, "", "", "", "")
|
||||
|
||||
|
||||
def test_drop_foreign_key_constraints_unsupported_engine(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test calling _drop_foreign_key_constraints with an unsupported engine."""
|
||||
|
||||
connection = Mock()
|
||||
connection.execute = Mock(side_effect=InternalError(None, None, None))
|
||||
session = Mock()
|
||||
session.connection = Mock(return_value=connection)
|
||||
instance = Mock()
|
||||
instance.get_session = Mock(return_value=session)
|
||||
engine = Mock()
|
||||
engine.dialect = Mock()
|
||||
engine.dialect.name = "sqlite"
|
||||
|
||||
session_maker = Mock(return_value=session)
|
||||
with pytest.raises(
|
||||
RuntimeError, match="_drop_foreign_key_constraints not supported for sqlite"
|
||||
):
|
||||
migration._drop_foreign_key_constraints(session_maker, engine, "", "")
|
||||
|
||||
|
||||
def test_update_states_table_with_foreign_key_options_unsupported_engine(
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""Test calling function with an unsupported engine.
|
||||
|
||||
This tests _update_states_table_with_foreign_key_options.
|
||||
"""
|
||||
|
||||
connection = Mock()
|
||||
connection.execute = Mock(side_effect=InternalError(None, None, None))
|
||||
session = Mock()
|
||||
session.connection = Mock(return_value=connection)
|
||||
instance = Mock()
|
||||
instance.get_session = Mock(return_value=session)
|
||||
engine = Mock()
|
||||
engine.dialect = Mock()
|
||||
engine.dialect.name = "sqlite"
|
||||
|
||||
session_maker = Mock(return_value=session)
|
||||
with pytest.raises(
|
||||
RuntimeError,
|
||||
match="_update_states_table_with_foreign_key_options not supported for sqlite",
|
||||
):
|
||||
migration._update_states_table_with_foreign_key_options(session_maker, engine)
|
||||
|
Loading…
x
Reference in New Issue
Block a user