diff --git a/tests/unit/test_cli_error_paths.py b/tests/unit/test_cli_error_paths.py new file mode 100644 index 0000000..67e5a14 --- /dev/null +++ b/tests/unit/test_cli_error_paths.py @@ -0,0 +1,142 @@ +from types import SimpleNamespace + +import pytest +from click.testing import CliRunner + +from mysql_to_sqlite3.cli import cli as mysql2sqlite + + +class _FakeConverter: + def __init__(self, *args, **kwargs): + pass + + def transfer(self): + raise RuntimeError("should not run") + + +def _fake_supported_charsets(charset=None): + """Produce deterministic charset/collation pairs for tests.""" + # When called without a charset, emulate the public API generator used by click.Choice. + if charset is None: + return iter( + [ + SimpleNamespace(id=0, charset="utf8mb4", collation="utf8mb4_general_ci"), + SimpleNamespace(id=1, charset="latin1", collation="latin1_swedish_ci"), + ] + ) + # When scoped to a particular charset, expose only the primary collation. + return iter([SimpleNamespace(id=0, charset=charset, collation=f"{charset}_general_ci")]) + + +class TestCliErrorPaths: + def test_mysql_collation_must_match_charset(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Invalid charset/collation combinations should be rejected before transfer starts.""" + monkeypatch.setattr("mysql_to_sqlite3.cli.mysql_supported_character_sets", _fake_supported_charsets) + monkeypatch.setattr("mysql_to_sqlite3.cli.MySQLtoSQLite", _FakeConverter) + + runner = CliRunner() + result = runner.invoke( + mysql2sqlite, + [ + "-f", + "out.sqlite3", + "-d", + "db", + "-u", + "user", + "--mysql-charset", + "utf8mb4", + "--mysql-collation", + "latin1_swedish_ci", + ], + ) + assert result.exit_code == 1 + assert "Invalid value for '--collation'" in result.output + + def test_debug_reraises_keyboard_interrupt(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Debug mode should bubble up KeyboardInterrupt for easier debugging.""" + + class KeyboardInterruptConverter: + def __init__(self, *args, **kwargs): + pass + + def transfer(self): + raise KeyboardInterrupt() + + monkeypatch.setattr("mysql_to_sqlite3.cli.MySQLtoSQLite", KeyboardInterruptConverter) + + kwargs = { + "sqlite_file": "out.sqlite3", + "mysql_user": "user", + "prompt_mysql_password": False, + "mysql_password": None, + "mysql_database": "db", + "mysql_tables": None, + "exclude_mysql_tables": None, + "mysql_views_as_tables": False, + "limit_rows": 0, + "collation": "BINARY", + "prefix_indices": False, + "without_foreign_keys": False, + "without_tables": False, + "without_data": False, + "strict": False, + "mysql_host": "localhost", + "mysql_port": 3306, + "mysql_charset": "utf8mb4", + "mysql_collation": None, + "skip_ssl": False, + "chunk": 200000, + "log_file": None, + "json_as_text": False, + "vacuum": False, + "use_buffered_cursors": False, + "quiet": False, + "debug": True, + } + with pytest.raises(KeyboardInterrupt): + mysql2sqlite.callback(**kwargs) + + def test_debug_reraises_generic_exception(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Debug mode should bubble up unexpected exceptions.""" + + class ExplodingConverter: + def __init__(self, *args, **kwargs): + pass + + def transfer(self): + raise RuntimeError("boom") + + monkeypatch.setattr("mysql_to_sqlite3.cli.MySQLtoSQLite", ExplodingConverter) + + kwargs = { + "sqlite_file": "out.sqlite3", + "mysql_user": "user", + "prompt_mysql_password": False, + "mysql_password": None, + "mysql_database": "db", + "mysql_tables": None, + "exclude_mysql_tables": None, + "mysql_views_as_tables": False, + "limit_rows": 0, + "collation": "BINARY", + "prefix_indices": False, + "without_foreign_keys": False, + "without_tables": False, + "without_data": False, + "strict": False, + "mysql_host": "localhost", + "mysql_port": 3306, + "mysql_charset": "utf8mb4", + "mysql_collation": None, + "skip_ssl": False, + "chunk": 200000, + "log_file": None, + "json_as_text": False, + "vacuum": False, + "use_buffered_cursors": False, + "quiet": False, + "debug": True, + } + with pytest.raises(RuntimeError): + mysql2sqlite.callback(**kwargs) diff --git a/tests/unit/test_transporter.py b/tests/unit/test_transporter.py index 2618045..0994cab 100644 --- a/tests/unit/test_transporter.py +++ b/tests/unit/test_transporter.py @@ -1,15 +1,417 @@ import builtins +import importlib.util +import os +import re import sqlite3 +import sys +import types as pytypes +import typing as t +from types import SimpleNamespace from unittest.mock import MagicMock, patch +import mysql.connector import pytest +from mysql.connector import errorcode from pytest_mock import MockerFixture +from typing_extensions import Unpack as ExtensionsUnpack from mysql_to_sqlite3.sqlite_utils import CollatingSequences from mysql_to_sqlite3.transporter import MySQLtoSQLite +from tests.conftest import MySQLCredentials class TestMySQLtoSQLiteTransporter: + def test_transporter_uses_typing_extensions_unpack_when_missing(self) -> None: + """Reload transporter without typing.Unpack to exercise fallback branch.""" + import mysql_to_sqlite3.transporter as transporter_module + + module_path = transporter_module.__file__ + assert module_path is not None + + real_typing = sys.modules["typing"] + fake_typing = pytypes.ModuleType("typing") + fake_typing.__dict__.update({k: v for k, v in real_typing.__dict__.items() if k != "Unpack"}) + sys.modules["typing"] = fake_typing + + try: + spec = importlib.util.spec_from_file_location("mysql_to_sqlite3.transporter_fallback", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + finally: + sys.modules["typing"] = real_typing + sys.modules.pop("mysql_to_sqlite3.transporter_fallback", None) + + assert module.Unpack is ExtensionsUnpack + + def test_constructor_normalizes_default_utf8mb4_collation( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Ensure utf8mb4_0900_ai_ci defaults downgrade to utf8mb4_unicode_ci.""" + from mysql_to_sqlite3 import transporter as transporter_module + + class FakeMySQLConnection: + def __init__(self) -> None: + self.database = None + + def is_connected(self) -> bool: + return True + + def cursor(self, *args, **kwargs) -> MagicMock: + return MagicMock() + + def get_server_version(self): + return (8, 0, 21) + + def reconnect(self) -> None: + return None + + fake_conn = FakeMySQLConnection() + + fake_charset = SimpleNamespace( + get_default_collation=lambda charset: ("utf8mb4_0900_ai_ci", None), + get_supported=lambda: ("utf8mb4",), + ) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.CharacterSet", lambda: fake_charset) + monkeypatch.setattr( + "mysql_to_sqlite3.transporter.mysql.connector.connect", + lambda **kwargs: fake_conn, + ) + monkeypatch.setattr(MySQLtoSQLite, "_setup_logger", MagicMock(return_value=MagicMock())) + + original_isinstance = builtins.isinstance + + def fake_isinstance(obj: object, classinfo: object) -> bool: + if obj is fake_conn and classinfo is transporter_module.MySQLConnectionAbstract: + return True + return original_isinstance(obj, classinfo) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.isinstance", fake_isinstance, raising=False) + + instance = MySQLtoSQLite( + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, + ) + + assert instance._mysql_collation == "utf8mb4_unicode_ci" + + def test_constructor_raises_when_mysql_not_connected( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + """Raise ConnectionError when mysql.connector.connect returns disconnected handle.""" + from mysql_to_sqlite3 import transporter as transporter_module + + class FakeMySQLConnection: + def __init__(self) -> None: + self.database = None + + def is_connected(self) -> bool: + return False + + def cursor(self, *args, **kwargs) -> MagicMock: + return MagicMock() + + fake_conn = FakeMySQLConnection() + + fake_charset = SimpleNamespace( + get_default_collation=lambda charset: ("utf8mb4_0900_ai_ci", None), + get_supported=lambda: ("utf8mb4",), + ) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.CharacterSet", lambda: fake_charset) + monkeypatch.setattr( + "mysql_to_sqlite3.transporter.mysql.connector.connect", + lambda **kwargs: fake_conn, + ) + monkeypatch.setattr(MySQLtoSQLite, "_setup_logger", MagicMock(return_value=MagicMock())) + + original_isinstance = builtins.isinstance + + def fake_isinstance(obj: object, classinfo: object) -> bool: + if obj is fake_conn and classinfo is transporter_module.MySQLConnectionAbstract: + return True + return original_isinstance(obj, classinfo) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.isinstance", fake_isinstance, raising=False) + + with pytest.raises(ConnectionError, match="Unable to connect to MySQL"): + MySQLtoSQLite( + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, + ) + + def test_transpile_mysql_expr_to_sqlite_parse_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Gracefully handle sqlglot parse errors when evaluating expressions.""" + + def explode(*args, **kwargs): + raise ValueError("boom") + + monkeypatch.setattr("mysql_to_sqlite3.transporter.parse_one", explode) + assert MySQLtoSQLite._transpile_mysql_expr_to_sqlite("invalid SQL") is None + + def test_transpile_mysql_type_to_sqlite_handles_length_and_synonyms(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure sqlglot-assisted mapper preserves length suffixes and synonyms.""" + + def fake_parse_one(expr_sql: str, read: str): + class FakeExpression: + def __init__(self, text: str) -> None: + self._text = text + + def sql(self, dialect: str) -> str: + return self._text + + return FakeExpression(expr_sql) + + real_search = re.search + + def fake_search(pattern: str, string: str, flags: int = 0): + if string.startswith("CAST(NULL AS"): + extracted = string[len("CAST(NULL AS ") : -1] + + class FakeMatch: + def __init__(self, value: str) -> None: + self._value = value + + def group(self, index: int) -> str: + return self._value + + return FakeMatch(extracted) + return real_search(pattern, string, flags) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.parse_one", fake_parse_one) + monkeypatch.setattr("mysql_to_sqlite3.transporter.re.search", fake_search) + + assert MySQLtoSQLite._transpile_mysql_type_to_sqlite("VARCHAR(42)") == "VARCHAR(42)" + assert MySQLtoSQLite._transpile_mysql_type_to_sqlite("CHAR(5)") == "CHARACTER(5)" + assert MySQLtoSQLite._transpile_mysql_type_to_sqlite("DECIMAL(10,2)") == "DECIMAL" + assert MySQLtoSQLite._transpile_mysql_type_to_sqlite("VARBINARY(8)") == "BLOB" + + def test_quote_sqlite_identifier_handles_non_utf8_bytes(self) -> None: + """Bytes that are not UTF-8 decodable should still be quoted safely.""" + quoted = MySQLtoSQLite._quote_sqlite_identifier(b"\xff") + assert quoted.startswith('"') + assert "xff" in quoted + + def test_translate_default_bytes_binary_literal(self) -> None: + """Binary defaults encoded with charset introducers should be converted.""" + column_default = b"_utf8mb4 b\\'1000001\\'" + result = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, "VARBINARY", "DEFAULT_GENERATED") + assert result == "DEFAULT 'A'" + + def test_translate_default_bytes_hex_literal(self) -> None: + """Hex defaults encoded with charset introducers should be preserved.""" + column_default = b"_utf8mb4 x\\'41\\'" + result = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, "VARBINARY", "DEFAULT_GENERATED") + assert result == "DEFAULT x'41'" + + def test_translate_default_bytes_decode_error(self) -> None: + """Un-decodable bytes should fall back to their repr-safe form.""" + column_default = b"\xff" + result = MySQLtoSQLite._translate_default_from_mysql_to_sqlite(column_default, "TEXT") + assert result == "DEFAULT 'b''\\xff'''" + + def test_translate_default_bytes_without_literal_prefix(self) -> None: + """Charset introducer without hex/bin prefix should fall back to hex literal.""" + column_default = b"_utf8mb4'abc'" + result = MySQLtoSQLite._translate_default_from_mysql_to_sqlite( + column_default, + "BLOB", + "DEFAULT_GENERATED", + ) + assert result == "DEFAULT x'616263'" + + def test_translate_default_generated_expression_variants(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Generated defaults should handle arithmetic inside parentheses and plain expressions.""" + monkeypatch.setattr( + MySQLtoSQLite, + "_transpile_mysql_expr_to_sqlite", + lambda expr: "(1+2)", + ) + assert ( + MySQLtoSQLite._translate_default_from_mysql_to_sqlite("expr", column_extra="DEFAULT_GENERATED") + == "DEFAULT (1+2)" + ) + + monkeypatch.setattr( + MySQLtoSQLite, + "_transpile_mysql_expr_to_sqlite", + lambda expr: "123", + ) + assert ( + MySQLtoSQLite._translate_default_from_mysql_to_sqlite("expr", column_extra="DEFAULT_GENERATED") + == "DEFAULT 123" + ) + + monkeypatch.setattr( + MySQLtoSQLite, + "_transpile_mysql_expr_to_sqlite", + lambda expr: "1 + 3", + ) + assert ( + MySQLtoSQLite._translate_default_from_mysql_to_sqlite("expr", column_extra="DEFAULT_GENERATED") + == "DEFAULT 1 + 3" + ) + + def test_data_type_collation_sequence_uses_transpiled_mapping(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Apply collation when sqlglot-based mapping yields textual type.""" + + def fake_transpile(cls, column_type: str, sqlite_json1_extension_enabled: bool = False) -> str: + return "varchar(10)" + + monkeypatch.setattr( + MySQLtoSQLite, + "_transpile_mysql_type_to_sqlite", + classmethod(fake_transpile), + ) + + result = MySQLtoSQLite._data_type_collation_sequence( + collation=CollatingSequences.NOCASE, + column_type="custom type", + ) + assert result == f"COLLATE {CollatingSequences.NOCASE}" + + def test_get_unique_index_name_handles_existing_suffixes(self) -> None: + """Ensure duplicate index names increment suffix until free slot is found.""" + with patch.object(MySQLtoSQLite, "__init__", return_value=None): + instance = MySQLtoSQLite() + + instance._seen_sqlite_index_names = {"idx_name", "idx_name_2"} + instance._sqlite_index_name_counters = {"idx_name": 2} + instance._prefix_indices = False + instance._logger = MagicMock() + + result = instance._get_unique_index_name("idx_name") + assert result == "idx_name_3" + instance._logger.info.assert_called_once() + + def test_build_create_table_sql_warns_on_non_integer_auto_increment(self) -> None: + """Auto increment primary keys with non-integer types should trigger warning and index decoding branches.""" + with patch.object(MySQLtoSQLite, "__init__", return_value=None): + instance = MySQLtoSQLite() + + instance._sqlite_strict = False + instance._sqlite_json1_extension_enabled = False + instance._mysql_cur_dict = MagicMock() + instance._mysql_cur = MagicMock() + instance._mysql = MagicMock() + instance._sqlite = MagicMock() + instance._mysql_database = "demo" + instance._collation = CollatingSequences.NOCASE + instance._prefix_indices = False + instance._without_tables = False + instance._without_foreign_keys = True + instance._logger = MagicMock() + + columns_rows = [ + { + "Field": "id", + "Type": "TEXT", + "Null": "NO", + "Default": None, + "Key": "PRI", + "Extra": "auto_increment", + } + ] + index_rows = [ + { + "name": b"idx_primary", + "primary": 0, + "unique": 0, + "auto_increment": 0, + "columns": b"name", + "types": b"VARCHAR(10)", + }, + { + "name": 123, + "primary": 0, + "unique": 1, + "auto_increment": 0, + "columns": "email", + "types": "INT", + }, + ] + + instance._mysql_cur_dict.fetchall.side_effect = [columns_rows, index_rows] + instance._mysql_cur_dict.fetchone.side_effect = [{"count": 1}, {"count": 0}] + instance._get_unique_index_name = MagicMock(side_effect=lambda name: f"{name}_unique") + + sql = instance._build_create_table_sql("users") + assert "CREATE TABLE" in sql + instance._logger.warning.assert_called_once() + assert instance._mysql_cur_dict.fetchone.call_count == 2 + + def test_build_create_view_sql_fallbacks_to_show_create(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When information_schema lookup fails, SHOW CREATE VIEW fallback should decode bytes safely.""" + with patch.object(MySQLtoSQLite, "__init__", return_value=None): + instance = MySQLtoSQLite() + + instance._mysql_database = "demo" + instance._mysql_cur_dict = MagicMock() + instance._mysql_cur = MagicMock() + instance._logger = MagicMock() + + error = mysql.connector.Error(msg="boom") + instance._mysql_cur_dict.execute.side_effect = error + + create_stmt = b"CREATE VIEW \xff AS SELECT 1" + instance._mysql_cur.fetchone.return_value = ("demo", create_stmt) + + real_search = re.search + + def fake_search(pattern: str, string: str, flags: int = 0): + if pattern == r"\bAS\b\s*(.*)$": + return None + return real_search(pattern, string, flags) + + monkeypatch.setattr("mysql_to_sqlite3.transporter.re.search", fake_search) + + instance._mysql_viewdef_to_sqlite = MagicMock(return_value="CREATE VIEW demo AS SELECT 1") + + result = instance._build_create_view_sql("demo_view") + + assert result == "CREATE VIEW demo AS SELECT 1" + instance._mysql_viewdef_to_sqlite.assert_called_once() + view_sql = instance._mysql_viewdef_to_sqlite.call_args.kwargs["view_select_sql"] + assert "SELECT" in view_sql + + def test_create_view_reconnect_aborts_after_retry(self) -> None: + """Lost connections during retry should warn and propagate the mysql error.""" + with patch.object(MySQLtoSQLite, "__init__", return_value=None): + instance = MySQLtoSQLite() + + class LostError(mysql.connector.Error): + def __init__(self, msg: str = "lost") -> None: + super().__init__(msg) + self.errno = errorcode.CR_SERVER_LOST + + instance._mysql = MagicMock() + instance._mysql_cur = MagicMock() + instance._sqlite_cur = MagicMock() + instance._sqlite = MagicMock() + instance._logger = MagicMock() + instance._build_create_view_sql = MagicMock(side_effect=LostError()) + + with pytest.raises(LostError): + instance._create_view("demo_view", attempting_reconnect=True) + + instance._mysql.reconnect.assert_called_once() + instance._logger.warning.assert_called_with("Connection to MySQL server lost.\nReconnection attempt aborted.") + def test_transfer_creates_view_when_flag_enabled(self) -> None: """When views_as_views is True, encountering a MySQL VIEW should create a SQLite VIEW and skip data transfer.""" with patch.object(MySQLtoSQLite, "__init__", return_value=None): @@ -210,6 +612,8 @@ def test_sqlite_strict_supported_keeps_flag( self, mock_mysql_connect: MagicMock, mock_sqlite_connect: MagicMock, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, mocker: MockerFixture, ) -> None: """Ensure STRICT mode remains enabled when SQLite supports it.""" @@ -246,12 +650,12 @@ def fake_isinstance(obj: object, classinfo: object) -> bool: mocker.patch("mysql_to_sqlite3.transporter.isinstance", side_effect=fake_isinstance) instance = MySQLtoSQLite( - sqlite_file="file.db", - mysql_user="user", - mysql_password=None, - mysql_database="db", - mysql_host="localhost", - mysql_port=3306, + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, sqlite_strict=True, ) @@ -264,6 +668,8 @@ def test_sqlite_strict_unsupported_disables_flag( self, mock_mysql_connect: MagicMock, mock_sqlite_connect: MagicMock, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, mocker: MockerFixture, ) -> None: """Ensure STRICT mode is disabled with a warning on old SQLite versions.""" @@ -300,12 +706,12 @@ def fake_isinstance(obj: object, classinfo: object) -> bool: mocker.patch("mysql_to_sqlite3.transporter.isinstance", side_effect=fake_isinstance) instance = MySQLtoSQLite( - sqlite_file="file.db", - mysql_user="user", - mysql_password=None, - mysql_database="db", - mysql_host="localhost", - mysql_port=3306, + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, sqlite_strict=True, ) @@ -353,47 +759,86 @@ def test_build_create_table_sql_appends_strict(self) -> None: assert "STRICT;" in sql - def test_constructor_missing_mysql_database(self) -> None: + def test_constructor_missing_mysql_database( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + ) -> None: """Test constructor raises ValueError if mysql_database is missing.""" from mysql_to_sqlite3.transporter import MySQLtoSQLite with pytest.raises(ValueError, match="Please provide a MySQL database"): - MySQLtoSQLite(mysql_user="user", sqlite_file="file.db") + MySQLtoSQLite( + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + ) - def test_constructor_missing_mysql_user(self) -> None: + def test_constructor_missing_mysql_user( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + ) -> None: """Test constructor raises ValueError if mysql_user is missing.""" from mysql_to_sqlite3.transporter import MySQLtoSQLite with pytest.raises(ValueError, match="Please provide a MySQL user"): - MySQLtoSQLite(mysql_database="db", sqlite_file="file.db") + MySQLtoSQLite( + mysql_database=mysql_credentials.database, + sqlite_file=sqlite_database, + ) - def test_constructor_missing_sqlite_file(self) -> None: + def test_constructor_missing_sqlite_file( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + ) -> None: """Test constructor raises ValueError if sqlite_file is missing.""" from mysql_to_sqlite3.transporter import MySQLtoSQLite with pytest.raises(ValueError, match="Please provide an SQLite file"): - MySQLtoSQLite(mysql_database="db", mysql_user="user") + MySQLtoSQLite( + mysql_database=mysql_credentials.database, + mysql_user=mysql_credentials.user, + ) - def test_constructor_mutually_exclusive_tables(self) -> None: + def test_constructor_mutually_exclusive_tables( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + ) -> None: """Test constructor raises ValueError if both mysql_tables and exclude_mysql_tables are provided.""" from mysql_to_sqlite3.transporter import MySQLtoSQLite with pytest.raises(ValueError, match="mutually exclusive"): MySQLtoSQLite( - mysql_database="db", - mysql_user="user", - sqlite_file="file.db", + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, mysql_tables=["a"], exclude_mysql_tables=["b"], ) - def test_constructor_without_tables_and_data(self) -> None: + def test_constructor_without_tables_and_data( + self, + sqlite_database: "os.PathLike[t.Any]", + mysql_credentials: MySQLCredentials, + ) -> None: """Test constructor raises ValueError if both without_tables and without_data are True.""" from mysql_to_sqlite3.transporter import MySQLtoSQLite with pytest.raises(ValueError, match="Unable to continue without transferring data or creating tables!"): MySQLtoSQLite( - mysql_database="db", mysql_user="user", sqlite_file="file.db", without_tables=True, without_data=True + sqlite_file=sqlite_database, + mysql_user=mysql_credentials.user, + mysql_password=mysql_credentials.password, + mysql_host=mysql_credentials.host, + mysql_port=mysql_credentials.port, + mysql_database=mysql_credentials.database, + without_tables=True, + without_data=True, ) def test_translate_default_from_mysql_to_sqlite_none(self) -> None: diff --git a/tests/unit/test_types_fallback.py b/tests/unit/test_types_fallback.py new file mode 100644 index 0000000..f3ca84e --- /dev/null +++ b/tests/unit/test_types_fallback.py @@ -0,0 +1,30 @@ +import importlib.util +import sys +import types as pytypes + +from typing_extensions import TypedDict as ExtensionsTypedDict + + +def test_types_module_uses_typing_extensions_when_typed_dict_missing() -> None: + """Reload the types module without typing.TypedDict to exercise the fallback branch.""" + import mysql_to_sqlite3.types as original_module + + module_path = original_module.__file__ + assert module_path is not None + + # Swap in a stripped-down typing module that lacks TypedDict. + real_typing = sys.modules["typing"] + fake_typing = pytypes.ModuleType("typing") + fake_typing.__dict__.update({k: v for k, v in real_typing.__dict__.items() if k != "TypedDict"}) + sys.modules["typing"] = fake_typing + + try: + spec = importlib.util.spec_from_file_location("mysql_to_sqlite3.types_fallback", module_path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + finally: + sys.modules["typing"] = real_typing + sys.modules.pop("mysql_to_sqlite3.types_fallback", None) + + assert module.TypedDict is ExtensionsTypedDict