Coverage for tests/test_mariadb_query_compilation.py: 100%

1"""MariaDB query compilation seam tests.""" 

2 

3from __future__ import annotations 

4 

5from snektest import assert_eq, test 

6 

7import snekql.mariadb.query as mariadb_query 

8from snekql import MISSING, Pending, insert, mariadb, select, update 

9from snekql.mariadb.query import compile_mariadb_select_sql, compile_mariadb_write_sql 

10 

11 

12@test(mark="fast") 

13def mariadb_query_compiler_uses_dialect_without_sqlite_translation() -> None: 

14 """MariaDB SQL compilation is direct, not translated from SQLite SQL.""" 

15 

16 assert_eq(hasattr(mariadb_query, "_translate_sqlite_sql"), False) 

17 assert_eq(hasattr(mariadb_query, "compile_select_sql"), False) 

18 assert_eq(hasattr(mariadb_query, "compile_write_sql"), False) 

19 

20 

21@test(mark="fast") 

22def mariadb_query_compilation_renders_backend_sql_and_codecs_directly() -> None: 

23 """MariaDB compilation owns quotes, placeholders, and value codecs.""" 

24 

25 class User[S = Pending](mariadb.Model[S, "User[object]"]): 

26 """MariaDB model used by direct dialect-compilation checks.""" 

27 

28 __tablename__ = "select" 

29 id: User.GenCol[int] = mariadb.Integer( 

30 primary_key=True, 

31 auto_increment=True, 

32 default=MISSING, 

33 ) 

34 enabled: User.Col[bool] = mariadb.Boolean(nullable=False) 

35 status: User.Col[str] = mariadb.Text(nullable=False) 

36 where: User.Col[str] = mariadb.Text(nullable=False) 

37 

38 select_sql, select_params = compile_mariadb_select_sql( 

39 select(User.where) 

40 .where(User.enabled.eq(True), User.status.in_("active", "paused")) 

41 .order_by(User.where.desc()) 

42 .limit(2) 

43 .offset(1), 

44 ) 

45 update_sql, update_params = compile_mariadb_write_sql( 

46 update(User).set(User.enabled.to(False)).where(User.where.ne("old")), 

47 ) 

48 insert_sql, insert_params = compile_mariadb_write_sql( 

49 insert(User(enabled=True, status="active", where="new")), 

50 ) 

51 

52 expected_select_sql = "".join( 

53 ( 

54 "SELECT `where` FROM `select` WHERE (`enabled` = %s) ", 

55 "AND (`status` IN (%s, %s)) ORDER BY `where` DESC LIMIT %s OFFSET %s", 

56 ) 

57 ) 

58 assert_eq(select_sql, expected_select_sql) 

59 assert_eq(select_params, (1, "active", "paused", 2, 1)) 

60 assert_eq(update_sql, "UPDATE `select` SET `enabled` = %s WHERE (`where` != %s)") 

61 assert_eq(update_params, (0, "old")) 

62 assert_eq( 

63 insert_sql, 

64 "INSERT INTO `select` (`enabled`, `status`, `where`) VALUES (%s, %s, %s)", 

65 ) 

66 assert_eq(insert_params, (1, "active", "new"))