Coverage for tests/test_mariadb_namespace.py: 100%

1"""MariaDB backend namespace declaration tests.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any, cast 

6 

7from snektest import ( 

8 assert_eq, 

9 assert_in, 

10 assert_isinstance, 

11 assert_ne, 

12 assert_raises, 

13 test, 

14) 

15 

16import snekql 

17from snekql import DatabaseRuntimeError 

18 

19 

20@test() 

21def mariadb_namespace_exports_backend_specific_names() -> None: 

22 """The MariaDB namespace exposes distinct declaration types.""" 

23 

24 assert_in("mariadb", snekql.__all__) 

25 assert_in("Config", snekql.mariadb.__all__) 

26 assert_ne(snekql.mariadb.Model, snekql.sqlite.Model) 

27 assert_ne(snekql.mariadb.Integer, snekql.sqlite.Integer) 

28 assert_ne(snekql.mariadb.Text, snekql.sqlite.Text) 

29 

30 class MariadbUser( 

31 snekql.mariadb.Model[snekql.Pending, "MariadbUser[snekql.Fetched]"] 

32 ): 

33 """MariaDB table model declared through the MariaDB namespace.""" 

34 

35 id: MariadbUser.GenCol[int] = snekql.mariadb.Integer( 

36 primary_key=True, 

37 auto_increment=True, 

38 default=snekql.MISSING, 

39 ) 

40 email: MariadbUser.Col[str] = snekql.mariadb.Text(nullable=False) 

41 

42 assert_isinstance(MariadbUser.id, snekql.mariadb.Attr) 

43 assert_isinstance(MariadbUser.email.eq("alice@example.com"), snekql.Predicate) 

44 

45 

46@test() 

47def mariadb_config_validates_connection_and_pool_settings() -> None: 

48 """MariaDB config rejects invalid declaration-time settings.""" 

49 

50 config = snekql.mariadb.Config( 

51 database="app", 

52 host="127.0.0.1", 

53 password="secret", 

54 port=3306, 

55 user="snekql", 

56 ) 

57 

58 assert_eq(config.host, "127.0.0.1") 

59 assert_eq(config.port, 3306) 

60 assert "secret" not in repr(config) 

61 

62 config_factory = cast("Any", snekql.mariadb.Config) 

63 

64 with assert_raises(DatabaseRuntimeError): 

65 _ = config_factory(database="app", host="", user="snekql") 

66 

67 with assert_raises(DatabaseRuntimeError): 

68 _ = config_factory(database="app", port=0, user="snekql") 

69 

70 with assert_raises(DatabaseRuntimeError): 

71 _ = config_factory(database="app", port=70000, user="snekql") 

72 

73 with assert_raises(DatabaseRuntimeError): 

74 _ = config_factory(database="", user="snekql") 

75 

76 with assert_raises(DatabaseRuntimeError): 

77 _ = config_factory(database="app", pool_size=True, user="snekql")