Coverage for tests/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 Integer, 

20 Model, 

21 Pending, 

22 PoolTimeoutError, 

23 Text, 

24 insert, 

25) 

26from tests.logging_helpers import NULL_LOGGER 

27 

28 

29class RuntimeUser[S = Pending](Model[S, "RuntimeUser[object]"]): 

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

31 

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

33 primary_key=True, 

34 auto_increment=True, 

35 default=MISSING, 

36 ) 

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

38 

39 

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

41 connection = connect(database_path) 

42 try: 

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

44 value = cursor.fetchone()[0] 

45 assert isinstance(value, int) 

46 return value 

47 finally: 

48 connection.close() 

49 

50 

51@test(mark="medium") 

52async def successful_transaction_commits() -> None: 

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

54 

55 with TemporaryDirectory() as directory: 

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

57 database = await Database.initialize( 

58 NULL_LOGGER, database=database_path, models=[RuntimeUser] 

59 ) 

60 try: 

61 async with database.transaction() as transaction: 

62 await transaction.execute( 

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

64 ) 

65 finally: 

66 await database.close() 

67 

68 assert_eq(_count_users(database_path), 1) 

69 

70 

71@test(mark="medium") 

72async def exceptional_transaction_rolls_back() -> None: 

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

74 

75 with TemporaryDirectory() as directory: 

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

77 database = await Database.initialize( 

78 NULL_LOGGER, database=database_path, models=[RuntimeUser] 

79 ) 

80 try: 

81 with assert_raises(ValueError): 

82 async with database.transaction() as transaction: 

83 await transaction.execute( 

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

85 ) 

86 msg = "force rollback" 

87 raise ValueError(msg) 

88 finally: 

89 await database.close() 

90 

91 assert_eq(_count_users(database_path), 0) 

92 

93 

94@test(mark="medium") 

95async def pool_exhaustion_raises_pool_timeout() -> None: 

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

97 

98 database = await Database.initialize( 

99 NULL_LOGGER, 

100 database=":memory:", 

101 pool_size=1, 

102 acquire_timeout=0.0, 

103 ) 

104 try: 

105 async with database.transaction(): 

106 with assert_raises(PoolTimeoutError): 

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

108 pass 

109 finally: 

110 await database.close() 

111 

112 

113@test(mark="medium") 

114async def pool_configuration_rejects_invalid_bounds() -> None: 

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

116 

117 with assert_raises(DatabaseRuntimeError): 

118 _ = await Database.initialize(NULL_LOGGER, database=":memory:", pool_size=0) 

119 

120 with assert_raises(DatabaseRuntimeError): 

121 _ = await Database.initialize( 

122 NULL_LOGGER, database=":memory:", acquire_timeout=-0.1 

123 ) 

124 

125 database = await Database.initialize(NULL_LOGGER, database=":memory:", pool_size=5) 

126 await database.close() 

127 

128 

129@test(mark="medium") 

130async def close_rejects_new_transactions_while_waiting_for_checkouts() -> None: 

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

132 

133 database = await Database.initialize( 

134 NULL_LOGGER, 

135 database=":memory:", 

136 pool_size=1, 

137 acquire_timeout=1.0, 

138 ) 

139 async with database.transaction(): 

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

141 await asyncio.sleep(0) 

142 with assert_raises(DatabaseClosingError): 

143 _ = database.transaction() 

144 await close_task 

145 

146 with assert_raises(DatabaseClosedError): 

147 _ = database.transaction() 

148 

149 

150@test(mark="medium") 

151async def timed_out_close_keeps_database_retryable() -> None: 

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

153 

154 database = await Database.initialize( 

155 NULL_LOGGER, 

156 database=":memory:", 

157 pool_size=1, 

158 acquire_timeout=0.0, 

159 ) 

160 transaction = database.transaction() 

161 _ = await transaction.__aenter__() 

162 

163 with assert_raises(DatabaseCloseTimeoutError): 

164 await database.close() 

165 

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

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

168 pass 

169 await database.close() 

170 

171 with assert_raises(DatabaseClosedError): 

172 _ = database.transaction()