Coverage for tests/mariadb_server.py: 87%

1"""Shared MariaDB server fixture for integration tests.""" 

2 

3from __future__ import annotations 

4 

5import socket 

6import subprocess 

7import time 

8from collections.abc import Generator 

9from dataclasses import dataclass 

10from pathlib import Path 

11from tempfile import TemporaryDirectory 

12 

13from snektest import session_fixture 

14 

15 

16@dataclass(frozen=True) 

17class MariaDBCommandResult: 

18 """Captured result from a MariaDB command-line client invocation.""" 

19 

20 returncode: int 

21 stderr: str 

22 stdout: str 

23 

24 

25@dataclass(frozen=True) 

26class MariaDBServer: 

27 """Connection details for a local unprivileged MariaDB test server.""" 

28 

29 database: str 

30 data_directory: Path 

31 host: str 

32 port: int 

33 socket_path: Path 

34 user: str 

35 

36 def run_sql(self, sql: str) -> MariaDBCommandResult: 

37 """Execute SQL against the local test server through the MariaDB CLI.""" 

38 

39 return _run_command( 

40 "mariadb", 

41 "--protocol=tcp", 

42 "-h", 

43 self.host, 

44 "-P", 

45 str(self.port), 

46 "-u", 

47 self.user, 

48 "-D", 

49 self.database, 

50 "-e", 

51 sql, 

52 ) 

53 

54 

55def _find_free_tcp_port() -> int: 

56 """Reserve a local TCP port long enough to learn an available number.""" 

57 

58 with socket.socket() as server_socket: 

59 server_socket.bind(("127.0.0.1", 0)) 

60 return int(server_socket.getsockname()[1]) 

61 

62 

63def _run_command(*args: str) -> MariaDBCommandResult: 

64 """Run a command and fail tests with captured diagnostics.""" 

65 

66 completed_process = subprocess.run( 

67 args, 

68 check=False, 

69 capture_output=True, 

70 text=True, 

71 ) 

72 result = MariaDBCommandResult( 

73 returncode=completed_process.returncode, 

74 stderr=completed_process.stderr, 

75 stdout=completed_process.stdout, 

76 ) 

77 if result.returncode != 0: 

78 command = " ".join(args) 

79 msg = f"command failed: {command}\n{result.stderr}" 

80 raise AssertionError(msg) 

81 return result 

82 

83 

84def _initialize_data_directory(data_directory: Path) -> None: 

85 """Create an isolated MariaDB data directory for a throwaway server.""" 

86 

87 _ = _run_command( 

88 "mariadb-install-db", 

89 "--no-defaults", 

90 f"--datadir={data_directory}", 

91 "--auth-root-authentication-method=normal", 

92 "--skip-test-db", 

93 ) 

94 

95 

96def _start_process( 

97 *, 

98 data_directory: Path, 

99 error_log_path: Path, 

100 pid_path: Path, 

101 port: int, 

102 socket_path: Path, 

103) -> subprocess.Popen[bytes]: 

104 """Start mariadbd without reading global configuration files.""" 

105 

106 return subprocess.Popen( 

107 [ 

108 "mariadbd", 

109 "--no-defaults", 

110 f"--datadir={data_directory}", 

111 f"--socket={socket_path}", 

112 f"--pid-file={pid_path}", 

113 f"--port={port}", 

114 "--bind-address=127.0.0.1", 

115 "--skip-grant-tables", 

116 "--skip-networking=0", 

117 f"--log-error={error_log_path}", 

118 ], 

119 stdout=subprocess.DEVNULL, 

120 stderr=subprocess.DEVNULL, 

121 ) 

122 

123 

124def _stop_process(process: subprocess.Popen[bytes]) -> None: 

125 """Terminate the test server and avoid leaving a child process behind.""" 

126 

127 if process.poll() is not None: 

128 return 

129 process.terminate() 

130 try: 

131 _ = process.wait(timeout=10.0) 

132 except subprocess.TimeoutExpired: 

133 process.kill() 

134 _ = process.wait() 

135 

136 

137def _wait_until_ready( 

138 *, 

139 error_log_path: Path, 

140 process: subprocess.Popen[bytes], 

141 server: MariaDBServer, 

142) -> None: 

143 """Poll the MariaDB CLI until the server accepts local TCP connections.""" 

144 

145 for _ in range(80): 

146 if process.poll() is not None: 

147 error_log = error_log_path.read_text() if error_log_path.exists() else "" 

148 msg = f"mariadbd exited before becoming ready\n{error_log}" 

149 raise AssertionError(msg) 

150 try: 

151 _ = _run_command( 

152 "mariadb", 

153 "--protocol=tcp", 

154 "-h", 

155 server.host, 

156 "-P", 

157 str(server.port), 

158 "-u", 

159 server.user, 

160 "-e", 

161 "SELECT 1", 

162 ) 

163 except AssertionError: 

164 time.sleep(0.25) 

165 else: 

166 return 

167 error_log = error_log_path.read_text() if error_log_path.exists() else "" 

168 msg = f"mariadbd did not become ready\n{error_log}" 

169 raise AssertionError(msg) 

170 

171 

172@session_fixture() 

173def provide_mariadb_server() -> Generator[MariaDBServer]: 

174 """Provide one local MariaDB server shared by medium integration tests.""" 

175 

176 with TemporaryDirectory(prefix="snekql-mariadb-") as temporary_directory_name: 

177 temporary_directory = Path(temporary_directory_name) 

178 data_directory = temporary_directory / "data" 

179 socket_path = temporary_directory / "mariadb.sock" 

180 pid_path = temporary_directory / "mariadb.pid" 

181 error_log_path = temporary_directory / "mariadb.err" 

182 server = MariaDBServer( 

183 database="snekql_test", 

184 data_directory=data_directory, 

185 host="127.0.0.1", 

186 port=_find_free_tcp_port(), 

187 socket_path=socket_path, 

188 user="root", 

189 ) 

190 _initialize_data_directory(data_directory) 

191 process = _start_process( 

192 data_directory=data_directory, 

193 error_log_path=error_log_path, 

194 pid_path=pid_path, 

195 port=server.port, 

196 socket_path=socket_path, 

197 ) 

198 try: 

199 _wait_until_ready( 

200 error_log_path=error_log_path, 

201 process=process, 

202 server=server, 

203 ) 

204 _ = _run_command( 

205 "mariadb", 

206 "--protocol=tcp", 

207 "-h", 

208 server.host, 

209 "-P", 

210 str(server.port), 

211 "-u", 

212 server.user, 

213 "-e", 

214 f"CREATE DATABASE {server.database}", 

215 ) 

216 yield server 

217 finally: 

218 _stop_process(process)