|
6 | 6 | from mysql_to_sqlite3.transporter import MySQLtoSQLite |
7 | 7 |
|
8 | 8 |
|
9 | | -class TestViewsSqlglot: |
10 | | - def test_mysql_viewdef_to_sqlite_strips_schema_and_transpiles(self) -> None: |
11 | | - mysql_select = "SELECT `u`.`id`, `u`.`name` FROM `db`.`users` AS `u` WHERE `u`.`id` > 1" |
12 | | - # Use an instance to ensure access to _mysql_database for stripping |
13 | | - with patch.object(MySQLtoSQLite, "__init__", return_value=None): |
14 | | - inst = MySQLtoSQLite() # type: ignore[call-arg] |
15 | | - inst._mysql_database = "db" # type: ignore[attr-defined] |
16 | | - sql = inst._mysql_viewdef_to_sqlite( |
17 | | - view_select_sql=mysql_select, |
18 | | - view_name="v_users", |
19 | | - ) |
20 | | - assert sql.startswith('CREATE VIEW IF NOT EXISTS "v_users" AS') |
21 | | - # Ensure schema qualifier was removed |
22 | | - assert '"db".' not in sql |
23 | | - assert "`db`." not in sql |
24 | | - # Ensure it targets sqlite dialect (identifiers quoted with ") |
25 | | - assert 'FROM "users"' in sql |
26 | | - # Ends with single semicolon |
27 | | - assert re.search(r";\s*$", sql) is not None |
28 | | - |
29 | | - def test_mysql_viewdef_to_sqlite_parse_fallback(self, monkeypatch: pytest.MonkeyPatch) -> None: |
30 | | - # Force parse_one to raise so we hit the fallback path |
31 | | - from sqlglot.errors import ParseError |
32 | | - |
33 | | - def boom(*args, **kwargs): |
34 | | - raise ParseError("boom") |
35 | | - |
36 | | - monkeypatch.setattr("mysql_to_sqlite3.transporter.parse_one", boom) |
37 | | - |
38 | | - sql_in = "SELECT 1" |
39 | | - out = MySQLtoSQLite._mysql_viewdef_to_sqlite( |
40 | | - view_select_sql=sql_in, |
41 | | - view_name="v1", |
42 | | - schema_name="db", |
43 | | - ) |
44 | | - assert out.startswith('CREATE VIEW IF NOT EXISTS "v1" AS') |
45 | | - assert "SELECT 1" in out |
46 | | - assert out.strip().endswith(";") |
47 | | - |
48 | | - def test_mysql_viewdef_to_sqlite_parse_fallback_strips_schema(self, monkeypatch: pytest.MonkeyPatch) -> None: |
49 | | - # Force parse_one to raise so we exercise the fallback path with schema qualifiers |
50 | | - from sqlglot.errors import ParseError |
51 | | - |
52 | | - def boom(*args, **kwargs): |
53 | | - raise ParseError("boom") |
54 | | - |
55 | | - monkeypatch.setattr("mysql_to_sqlite3.transporter.parse_one", boom) |
56 | | - |
57 | | - mysql_select = "SELECT `u`.`id` FROM `db`.`users` AS `u` WHERE `u`.`id` > 1" |
58 | | - out = MySQLtoSQLite._mysql_viewdef_to_sqlite( |
59 | | - view_select_sql=mysql_select, |
60 | | - view_name="v_users", |
61 | | - schema_name="db", |
62 | | - ) |
63 | | - # Should not contain schema qualifier anymore |
64 | | - assert "`db`." not in out and '"db".' not in out and " db." not in out |
65 | | - # Should still reference the table name |
66 | | - assert "FROM `users`" in out or 'FROM "users"' in out or "FROM users" in out |
67 | | - assert out.strip().endswith(";") |
68 | | - |
69 | | - def test_mysql_viewdef_to_sqlite_keep_schema_true_preserves_qualifiers(self) -> None: |
70 | | - mysql_select = "SELECT `u`.`id` FROM `db`.`users` AS `u`" |
71 | | - sql = MySQLtoSQLite._mysql_viewdef_to_sqlite( |
72 | | - view_select_sql=mysql_select, |
73 | | - view_name="v_users", |
74 | | - schema_name="db", |
75 | | - keep_schema=True, |
76 | | - ) |
77 | | - # Should not strip the schema when keep_schema=True |
78 | | - assert "`db`." in sql or '"db".' in sql |
79 | | - assert sql.strip().endswith(";") |
80 | | - |
81 | | - |
82 | | -import re |
83 | | -from unittest.mock import patch |
84 | | - |
85 | | -import pytest |
86 | | - |
87 | | -from mysql_to_sqlite3.transporter import MySQLtoSQLite |
88 | | - |
89 | | - |
90 | 9 | class TestViewsSqlglot: |
91 | 10 | def test_mysql_viewdef_to_sqlite_strips_schema_and_transpiles(self) -> None: |
92 | 11 | mysql_select = "SELECT `u`.`id`, `u`.`name` FROM `db`.`users` AS `u` WHERE `u`.`id` > 1" |
|
0 commit comments