Coverage for tests/runtime/test_database_runtime.py: 99%

1"""Database runtime lifecycle behavior tests.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from pathlib import Path 

7from sqlite3 import connect 

8from tempfile import TemporaryDirectory 

9 

10from snektest import assert_eq, assert_raises, test 

11 

12from snekql import ( 

13 MISSING, 

14 Database, 

15 DatabaseClosedError, 

16 DatabaseCloseTimeoutError, 

17 DatabaseClosingError, 

18 DatabaseRuntimeError, 

19 Fetched, 

20 Integer, 

21 Model, 

22 Pending, 

23 PoolTimeoutError, 

24 Text, 

25 insert, 

26) 

27from tests.helpers import NULL_LOGGER 

28 

29 

30class RuntimeUser[S = Pending](Model[S, "RuntimeUser[Fetched]"]): 

31 """Table model used by transaction runtime tests.""" 

32 

33 id: RuntimeUser.GenCol[int] = Integer( 

34 primary_key=True, 

35 auto_increment=True, 

36 default=MISSING, 

37 ) 

38 email: RuntimeUser.Col[str] = Text(nullable=False) 

39 

40 

41def _count_users(database_path: Path) -> int: 

42 connection = connect(database_path) 

43 try: 

44 cursor = connection.execute('SELECT COUNT(*) FROM "runtime_user"') 

45 value = cursor.fetchone()[0] 

46 assert isinstance(value, int) 

47 return value 

48 finally: 

49 connection.close() 

50 

51 

52@test(mark="medium") 

53async def successful_transaction_commits() -> None: 

54 """A transaction commits writes when its context exits successfully.""" 

55 

56 with TemporaryDirectory() as directory: 

57 database_path = Path(directory) / "app.db" 

58 database = await Database.initialize( 

59 logger=NULL_LOGGER, database=database_path, models=[RuntimeUser] 

60 ) 

61 try: 

62 async with database.transaction() as tx: 

63 await tx.execute(insert(RuntimeUser(email="alice@example.com"))) 

64 finally: 

65 await database.close() 

66 

67 assert_eq(_count_users(database_path), 1) 

68 

69 

70@test(mark="medium") 

71async def exceptional_transaction_rolls_back() -> None: 

72 """A transaction rolls back writes when its context exits exceptionally.""" 

73 

74 with TemporaryDirectory() as directory: 

75 database_path = Path(directory) / "app.db" 

76 database = await Database.initialize( 

77 logger=NULL_LOGGER, database=database_path, models=[RuntimeUser] 

78 ) 

79 try: 

80 with assert_raises(ValueError): 

81 async with database.transaction() as tx: 

82 await tx.execute( 

83 insert(RuntimeUser(email="rollback@example.com")), 

84 ) 

85 msg = "force rollback" 

86 raise ValueError(msg) 

87 finally: 

88 await database.close() 

89 

90 assert_eq(_count_users(database_path), 0) 

91 

92 

93@test(mark="medium") 

94async def pool_exhaustion_raises_pool_timeout() -> None: 

95 """A checkout beyond pool_size waits only up to the transaction timeout.""" 

96 

97 database = await Database.initialize( 

98 logger=NULL_LOGGER, 

99 database=":memory:", 

100 pool_size=1, 

101 acquire_timeout=0.0, 

102 ) 

103 try: 

104 async with database.transaction(): 

105 with assert_raises(PoolTimeoutError): 

106 async with database.transaction(timeout=0.0): 

107 pass 

108 finally: 

109 await database.close() 

110 

111 

112@test(mark="medium") 

113async def pool_configuration_rejects_invalid_bounds() -> None: 

114 """Pool size and acquisition timeout validate their documented lower bounds.""" 

115 

116 with assert_raises(DatabaseRuntimeError): 

117 _ = await Database.initialize( 

118 logger=NULL_LOGGER, database=":memory:", pool_size=0 

119 ) 

120 

121 with assert_raises(DatabaseRuntimeError): 

122 _ = await Database.initialize( 

123 logger=NULL_LOGGER, database=":memory:", acquire_timeout=-0.1 

124 ) 

125 

126 database = await Database.initialize( 

127 logger=NULL_LOGGER, database=":memory:", pool_size=5 

128 ) 

129 await database.close() 

130 

131 

132@test(mark="medium") 

133async def close_rejects_new_transactions_while_waiting_for_checkouts() -> None: 

134 """Closing temporarily rejects new transactions until checked-out work exits.""" 

135 

136 database = await Database.initialize( 

137 logger=NULL_LOGGER, 

138 database=":memory:", 

139 pool_size=1, 

140 acquire_timeout=1.0, 

141 ) 

142 async with database.transaction(): 

143 close_task = asyncio.create_task(database.close()) 

144 await asyncio.sleep(0) 

145 with assert_raises(DatabaseClosingError): 

146 _ = database.transaction() 

147 await close_task 

148 

149 with assert_raises(DatabaseClosedError): 

150 _ = database.transaction() 

151 

152 

153@test(mark="medium") 

154async def timed_out_close_keeps_database_retryable() -> None: 

155 """A close timeout leaves the database open once checked-out work returns.""" 

156 

157 database = await Database.initialize( 

158 logger=NULL_LOGGER, 

159 database=":memory:", 

160 pool_size=1, 

161 acquire_timeout=0.0, 

162 ) 

163 transaction = database.transaction() 

164 _ = await transaction.__aenter__() 

165 

166 with assert_raises(DatabaseCloseTimeoutError): 

167 await database.close() 

168 

169 await transaction.__aexit__(None, None, None) 

170 async with database.transaction(timeout=0.0): 

171 pass 

172 await database.close() 

173 

174 with assert_raises(DatabaseClosedError): 

175 _ = database.transaction()