🦺 improve robustness and correctness of MySQL-to-SQLite migrations (#111)

Author: techouse

src/mysql_to_sqlite3/transporter.py

@@ -0,0 +1,97 @@
+import re
+from unittest.mock import MagicMock, patch
+
+from mysql_to_sqlite3.transporter import MySQLtoSQLite
+
+
+def _make_base_instance():
+    with patch.object(MySQLtoSQLite, "__init__", return_value=None):
+        inst = MySQLtoSQLite()  # type: ignore[call-arg]
+    inst._mysql_cur_dict = MagicMock()
+    inst._mysql_database = "db"
+    inst._sqlite_json1_extension_enabled = False
+    inst._collation = "BINARY"
+    inst._prefix_indices = False
+    inst._without_tables = False
+    inst._without_foreign_keys = True
+    inst._logger = MagicMock()
+    inst._sqlite_strict = False
+    # Track index names for uniqueness
+    inst._seen_sqlite_index_names = set()
+    inst._sqlite_index_name_counters = {}
+    return inst
+
+
+def test_show_columns_backticks_are_escaped_in_mysql_query() -> None:
+    inst = _make_base_instance()
+
+    # Capture executed SQL
+    executed_sql = []
+
+    def capture_execute(sql: str, *_, **__):
+        executed_sql.append(sql)
+
+    inst._mysql_cur_dict.execute.side_effect = capture_execute
+
+    # SHOW COLUMNS -> then STATISTICS query
+    inst._mysql_cur_dict.fetchall.side_effect = [
+        [
+            {
+                "Field": "id",
+                "Type": "INT",
+                "Null": "NO",
+                "Default": None,
+                "Key": "PRI",
+                "Extra": "",
+            }
+        ],
+        [],
+    ]
+    # TABLE collision check -> 0
+    inst._mysql_cur_dict.fetchone.return_value = {"count": 0}
+
+    sql = inst._build_create_table_sql("we`ird")
+    assert sql.startswith('CREATE TABLE IF NOT EXISTS "we`ird" (')
+
+    # First executed SQL should be SHOW COLUMNS with backticks escaped
+    assert executed_sql
+    assert executed_sql[0] == "SHOW COLUMNS FROM `we``ird`"
+
+
+def test_identifiers_with_double_quotes_are_safely_quoted_in_create_and_index() -> None:
+    inst = _make_base_instance()
+    inst._prefix_indices = True  # ensure an index is emitted with a deterministic name prefix
+
+    # SHOW COLUMNS first call, then STATISTICS rows
+    inst._mysql_cur_dict.fetchall.side_effect = [
+        [
+            {
+                "Field": 'na"me',
+                "Type": "VARCHAR(10)",
+                "Null": "YES",
+                "Default": None,
+                "Key": "",
+                "Extra": "",
+            },
+        ],
+        [
+            {
+                "name": "idx",
+                "primary": 0,
+                "unique": 0,
+                "auto_increment": 0,
+                "columns": 'na"me',
+                "types": "VARCHAR(10)",
+            }
+        ],
+    ]
+    inst._mysql_cur_dict.fetchone.return_value = {"count": 0}
+
+    sql = inst._build_create_table_sql('ta"ble')
+
+    # Column should be quoted with doubled quotes inside
+    assert '"na""me" VARCHAR(10)' in sql or '"na""me" TEXT' in sql
+
+    # Index should quote table and column names with doubled quotes
+    norm = re.sub(r"\s+", " ", sql)
+    assert 'CREATE INDEX IF NOT EXISTS "ta""ble_idx" ON "ta""ble" ("na""me")' in norm

tests/unit/test_build_create_table_sql_sqlglot_identifiers.py

@@ -0,0 +1,51 @@
+import pytest
+
+from mysql_to_sqlite3.sqlite_utils import CollatingSequences
+from mysql_to_sqlite3.transporter import MySQLtoSQLite
+
+
+class TestCollationSqlglotAugmented:
+    @pytest.mark.parametrize(
+        "mysql_type",
+        [
+            "char varying(12)",
+            "CHARACTER VARYING(12)",
+        ],
+    )
+    def test_collation_applied_for_char_varying_synonyms(self, mysql_type: str) -> None:
+        out = MySQLtoSQLite._data_type_collation_sequence(collation=CollatingSequences.NOCASE, column_type=mysql_type)
+        assert out == f"COLLATE {CollatingSequences.NOCASE}"
+
+    def test_collation_applied_for_national_character_varying(self) -> None:
+        out = MySQLtoSQLite._data_type_collation_sequence(
+            collation=CollatingSequences.NOCASE, column_type="national character varying(15)"
+        )
+        assert out == f"COLLATE {CollatingSequences.NOCASE}"
+
+    def test_no_collation_for_json(self) -> None:
+        # Regardless of case or synonym handling, JSON should not have collation applied
+        assert (
+            MySQLtoSQLite._data_type_collation_sequence(collation=CollatingSequences.NOCASE, column_type="json") == ""
+        )
+
+    def test_no_collation_when_binary_collation(self) -> None:
+        # BINARY collation disables COLLATE clause entirely
+        assert (
+            MySQLtoSQLite._data_type_collation_sequence(collation=CollatingSequences.BINARY, column_type="VARCHAR(10)")
+            == ""
+        )
+
+    @pytest.mark.parametrize(
+        "numeric_synonym",
+        [
+            "double precision",
+            "FIXED(10,2)",
+        ],
+    )
+    def test_no_collation_for_numeric_synonyms(self, numeric_synonym: str) -> None:
+        assert (
+            MySQLtoSQLite._data_type_collation_sequence(
+                collation=CollatingSequences.NOCASE, column_type=numeric_synonym
+            )
+            == ""
+        )

tests/unit/test_collation_sqlglot_augmented.py

@@ -0,0 +1,60 @@
+import pytest
+
+from mysql_to_sqlite3.transporter import MySQLtoSQLite
+
+
+class TestDefaultsSqlglotEnhanced:
+    @pytest.mark.parametrize(
+        "expr,expected",
+        [
+            ("CURRENT_TIME", "DEFAULT CURRENT_TIME"),
+            ("CURRENT_DATE", "DEFAULT CURRENT_DATE"),
+            ("CURRENT_TIMESTAMP", "DEFAULT CURRENT_TIMESTAMP"),
+        ],
+    )
+    def test_current_tokens_passthrough(self, expr: str, expected: str) -> None:
+        assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite(expr, column_extra="DEFAULT_GENERATED") == expected
+
+    def test_null_literal_generated(self) -> None:
+        assert (
+            MySQLtoSQLite._translate_default_from_mysql_to_sqlite("NULL", column_extra="DEFAULT_GENERATED")
+            == "DEFAULT NULL"
+        )
+
+    @pytest.mark.parametrize(
+        "expr,boolean_type,expected",
+        [
+            ("true", "BOOLEAN", {"DEFAULT(TRUE)", "DEFAULT '1'"}),
+            ("false", "BOOLEAN", {"DEFAULT(FALSE)", "DEFAULT '0'"}),
+            ("true", "INTEGER", {"DEFAULT '1'"}),
+            ("false", "INTEGER", {"DEFAULT '0'"}),
+        ],
+    )
+    def test_boolean_tokens_generated(self, expr: str, boolean_type: str, expected: set) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(
+            expr, column_type=boolean_type, column_extra="DEFAULT_GENERATED"
+        )
+        assert out in expected
+
+    def test_parenthesized_string_literal_generated(self) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("('abc')", column_extra="DEFAULT_GENERATED")
+        # Either DEFAULT 'abc' or DEFAULT ('abc') depending on normalization
+        assert out in {"DEFAULT 'abc'", "DEFAULT ('abc')"}
+
+    def test_parenthesized_numeric_literal_generated(self) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("(42)", column_extra="DEFAULT_GENERATED")
+        assert out in {"DEFAULT 42", "DEFAULT (42)"}
+
+    def test_constant_arithmetic_expression_generated(self) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("1+2*3", column_extra="DEFAULT_GENERATED")
+        # sqlglot formats with spaces for sqlite dialect
+        assert out in {"DEFAULT 1 + 2 * 3", "DEFAULT (1 + 2 * 3)"}
+
+    def test_hex_blob_literal_generated(self) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("x'41'", column_extra="DEFAULT_GENERATED")
+        # Should recognize as blob literal and keep as-is
+        assert out.upper() == "DEFAULT X'41'"
+
+    def test_plain_string_escaping_single_quote(self) -> None:
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("O'Reilly")
+        assert out == "DEFAULT 'O''Reilly'"

tests/unit/test_defaults_sqlglot_enhanced.py

@@ -49,7 +49,7 @@ def test_build_create_table_sql_prefix_indices_true_prefixes_index_names() -> No
     sql = inst._build_create_table_sql("users")
 
     # With prefix_indices=True, the index name should be prefixed with table name
-    assert 'CREATE  INDEX IF NOT EXISTS "users_idx_name" ON "users" ("name");' in sql
+    assert 'CREATE INDEX IF NOT EXISTS "users_idx_name" ON "users" ("name");' in sql
 
 
 def test_build_create_table_sql_collision_renamed_and_uniqueness_suffix() -> None:

tests/unit/test_indices_prefix_and_uniqueness.py

@@ -40,6 +40,18 @@ def test_data_type_collation_sequence(self) -> None:
     def test_translate_default_common_keywords(self, default: str, expected: str) -> None:
         assert MySQLtoSQLite._translate_default_from_mysql_to_sqlite(default) == expected
 
+    def test_translate_default_current_timestamp_precision_transpiled(self) -> None:
+        # MySQL allows fractional seconds: CURRENT_TIMESTAMP(6). Ensure it's normalized to SQLite token.
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(
+            "CURRENT_TIMESTAMP(6)", column_extra="DEFAULT_GENERATED"
+        )
+        assert out == "DEFAULT CURRENT_TIMESTAMP"
+
+    def test_translate_default_generated_expr_fallback_quotes(self) -> None:
+        # Unknown expressions should fall back to quoted string default for safety
+        out = MySQLtoSQLite._translate_default_from_mysql_to_sqlite("uuid()", column_extra="DEFAULT_GENERATED")
+        assert out == "DEFAULT 'uuid()'"
+
     def test_translate_default_charset_introducer_str_hex_and_bin(self) -> None:
         # DEFAULT_GENERATED with charset introducer and hex (escaped as in MySQL)
         s = "_utf8mb4 X\\'41\\'"  # hex for 'A'

tests/unit/test_types_and_defaults_extra.py

@@ -0,0 +1,72 @@
+import pytest
+
+from mysql_to_sqlite3.transporter import MySQLtoSQLite
+
+
+class TestSqlglotAugmentedTypeTranslation:
+    @pytest.mark.parametrize("mysql_type", ["double precision", "DOUBLE PRECISION", "DoUbLe PrEcIsIoN"])
+    def test_double_precision_maps_to_numeric_type(self, mysql_type: str) -> None:
+        # Prior mapper would resolve this to TEXT; sqlglot fallback should improve it
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite(mysql_type)
+        assert out in {"DOUBLE", "REAL"}
+
+    def test_fixed_maps_to_decimal(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("fixed(10,2)")
+        # Normalize to DECIMAL (without length) to match existing style
+        assert out == "DECIMAL"
+
+    def test_character_varying_keeps_length_as_varchar(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("character varying(20)")
+        assert out == "VARCHAR(20)"
+
+    def test_char_varying_keeps_length_as_varchar(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("char varying(12)")
+        assert out == "VARCHAR(12)"
+
+    def test_national_character_varying_maps_to_nvarchar(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("national character varying(15)")
+        assert out == "NVARCHAR(15)"
+
+    def test_national_character_maps_to_nchar(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("national character(5)")
+        assert out == "NCHAR(5)"
+
+    @pytest.mark.parametrize(
+        "mysql_type,expected",
+        [
+            ("int unsigned", "INTEGER"),
+            ("mediumint unsigned", "MEDIUMINT"),
+            ("smallint unsigned", "SMALLINT"),
+            ("tinyint unsigned", "TINYINT"),
+            ("bigint unsigned", "BIGINT"),
+        ],
+    )
+    def test_unsigned_variants_strip_unsigned(self, mysql_type: str, expected: str) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite(mysql_type)
+        assert out == expected
+
+    def test_timestamp_maps_to_datetime(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("timestamp")
+        assert out == "DATETIME"
+
+    def test_varbinary_and_blobs_map_to_blob(self) -> None:
+        assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite("varbinary(16)") == "BLOB"
+        assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite("mediumblob") == "BLOB"
+
+    def test_char_maps_to_character_with_length(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("char(3)")
+        assert out == "CHARACTER(3)"
+
+    def test_json_mapping_respects_json1(self) -> None:
+        assert (
+            MySQLtoSQLite._translate_type_from_mysql_to_sqlite("json", sqlite_json1_extension_enabled=False) == "TEXT"
+        )
+        assert MySQLtoSQLite._translate_type_from_mysql_to_sqlite("json", sqlite_json1_extension_enabled=True) == "JSON"
+
+    def test_fallback_to_text_on_unknown_type(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("geography")
+        assert out == "TEXT"
+
+    def test_enum_remains_text(self) -> None:
+        out = MySQLtoSQLite._translate_type_from_mysql_to_sqlite("enum('a','b')")
+        assert out == "TEXT"

tests/unit/test_types_sqlglot_augmented.py

@@ -112,4 +112,4 @@ import-order-style = pycharm
 application-import-names = flake8
 
 [pylint]
-disable = C0209,C0301,C0411,R,W0107,W0622,C0103
+disable = C0209,C0301,C0411,R,W0107,W0622,C0103,C0302