Coverage for tests/test_runtime_selection.py: 100%

1"""Backend Runtime Adapter selection seam tests.""" 

2 

3from __future__ import annotations 

4 

5import subprocess 

6import sys 

7from pathlib import Path 

8from typing import cast 

9 

10from snektest import assert_eq, assert_raises, test 

11 

12from snekql import DatabaseRuntimeError, mariadb, sqlite 

13from snekql._runtime_selection import resolve_runtime_selection 

14 

15 

16@test(mark="fast") 

17def runtime_selection_resolves_backend_family_and_config() -> None: 

18 """Backend config resolution lives in one Runtime Adapter selection seam.""" 

19 

20 legacy_selection = resolve_runtime_selection( 

21 backend=None, 

22 database=Path("app.db"), 

23 pool_size=2, 

24 acquire_timeout=0.5, 

25 ) 

26 sqlite_config = sqlite.Config(database=Path("explicit.db"), pool_size=3) 

27 sqlite_selection = resolve_runtime_selection( 

28 backend=sqlite_config, 

29 database=None, 

30 pool_size=5, 

31 acquire_timeout=30.0, 

32 ) 

33 mariadb_config = mariadb.Config(database="app", user="snekql") 

34 mariadb_selection = resolve_runtime_selection( 

35 backend=mariadb_config, 

36 database=None, 

37 pool_size=5, 

38 acquire_timeout=30.0, 

39 ) 

40 

41 assert_eq(legacy_selection.backend_family, "sqlite") 

42 assert_eq(cast("sqlite.Config", legacy_selection.config).database, Path("app.db")) 

43 assert_eq(cast("sqlite.Config", legacy_selection.config).pool_size, 2) 

44 assert_eq(cast("sqlite.Config", legacy_selection.config).acquire_timeout, 0.5) 

45 assert_eq(sqlite_selection.backend_family, "sqlite") 

46 assert_eq(sqlite_selection.config, sqlite_config) 

47 assert_eq(mariadb_selection.backend_family, "mariadb") 

48 assert_eq(mariadb_selection.config, mariadb_config) 

49 

50 

51@test(mark="fast") 

52def runtime_selection_rejects_unsupported_or_ambiguous_configuration() -> None: 

53 """Unsupported backend selections fail before runtime modules initialize.""" 

54 

55 with assert_raises(DatabaseRuntimeError): 

56 _ = resolve_runtime_selection( 

57 backend=object(), 

58 database=None, 

59 pool_size=5, 

60 acquire_timeout=30.0, 

61 ) 

62 

63 with assert_raises(DatabaseRuntimeError): 

64 _ = resolve_runtime_selection( 

65 backend=sqlite.Config(database=":memory:"), 

66 database=Path("app.db"), 

67 pool_size=5, 

68 acquire_timeout=30.0, 

69 ) 

70 

71 with assert_raises(DatabaseRuntimeError): 

72 _ = resolve_runtime_selection( 

73 backend=None, 

74 database=None, 

75 pool_size=5, 

76 acquire_timeout=30.0, 

77 ) 

78 

79 

80@test(mark="fast") 

81def runtime_selection_does_not_import_backend_runtime_modules() -> None: 

82 """Resolving a backend adapter selection avoids optional runtime imports.""" 

83 

84 script = """ 

85from pathlib import Path 

86import sys 

87from snekql import mariadb, sqlite 

88from snekql._runtime_selection import resolve_runtime_selection 

89 

90_ = resolve_runtime_selection(backend=None, database=Path('app.db'), pool_size=5, acquire_timeout=30.0) 

91_ = resolve_runtime_selection(backend=sqlite.Config(database=':memory:'), database=None, pool_size=5, acquire_timeout=30.0) 

92_ = resolve_runtime_selection(backend=mariadb.Config(database='app', user='snekql'), database=None, pool_size=5, acquire_timeout=30.0) 

93for name in ('snekql.sqlite.runtime', 'snekql.mariadb.runtime', 'aiosqlite', 'aiomysql'): 

94 if name in sys.modules: 

95 raise AssertionError(f'{name} was imported') 

96print('ok') 

97""" 

98 

99 result = subprocess.run( 

100 [sys.executable, "-c", script], 

101 check=False, 

102 capture_output=True, 

103 text=True, 

104 ) 

105 

106 assert_eq(result.returncode, 0) 

107 assert_eq(result.stdout.strip(), "ok")