Coverage for tests/testing/mariadb/test_public_api.py: 98%
99 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-07 21:13 +0300
1"""Temporary MariaDB Test Server public API tests."""
3from __future__ import annotations
5import asyncio
6import shutil
7from pathlib import Path
8from tempfile import TemporaryDirectory
10from snektest import (
11 assert_eq,
12 assert_in,
13 assert_is_none,
14 assert_raises,
15 assert_true,
16 test,
17)
19from snekql import mariadb
20from snekql.testing.mariadb import (
21 TemporaryMariaDBServer,
22 TemporaryMariaDBServerError,
23 temporary_mariadb_server,
24)
27@test(mark="fast")
28def temporary_mariadb_server_rejects_invalid_option_combinations() -> None:
29 """Invalid test-server option combinations fail before startup."""
31 invalid_calls = (
32 (
33 lambda: temporary_mariadb_server(transports=set()),
34 "at least one transport",
35 ),
36 (
37 lambda: temporary_mariadb_server(clean_before_start=True),
38 "clean_before_start requires data_directory",
39 ),
40 (
41 lambda: temporary_mariadb_server(port=3307),
42 "port requires tcp transport",
43 ),
44 (
45 lambda: temporary_mariadb_server(
46 transports={"tcp"},
47 socket_path=Path("mariadb.sock"),
48 ),
49 "socket_path requires unix_socket transport",
50 ),
51 (
52 lambda: temporary_mariadb_server(auth="insecure", password="test"),
53 "password requires auth='password'",
54 ),
55 (
56 lambda: temporary_mariadb_server(database="test-db"),
57 "MariaDB database must be",
58 ),
59 (
60 lambda: temporary_mariadb_server(user="test-user"),
61 "MariaDB user must be",
62 ),
63 (
64 lambda: temporary_mariadb_server(server_args=("--port=3307",)),
65 "managed mariadbd option",
66 ),
67 (
68 lambda: temporary_mariadb_server(
69 clean_before_start=True,
70 data_directory=Path("data"),
71 reset_database=True,
72 ),
73 "reset_database is incompatible with clean_before_start",
74 ),
75 )
77 for invalid_call, expected_message in invalid_calls:
78 with assert_raises(TemporaryMariaDBServerError) as error:
79 _ = invalid_call()
80 assert_in(expected_message, str(error.exception))
83@test(mark="fast")
84def temporary_mariadb_server_config_prefers_unix_socket() -> None:
85 """Server configs expose requested transports and prefer Unix socket."""
87 server = TemporaryMariaDBServer(
88 auth="password",
89 database="test",
90 data_directory=Path("data"),
91 error_log_path=Path("mariadb.err"),
92 host="127.0.0.1",
93 password="secret",
94 pid_path=Path("mariadb.pid"),
95 port=4306,
96 socket_path=Path("mariadb.sock"),
97 transports=frozenset({"unix_socket", "tcp"}),
98 user="root",
99 )
101 default_config = server.config(pool_size=2, acquire_timeout=3.0)
102 assert_eq(
103 default_config,
104 mariadb.Config(
105 database="test",
106 user="root",
107 password="secret",
108 unix_socket=Path("mariadb.sock"),
109 pool_size=2,
110 acquire_timeout=3.0,
111 ),
112 )
114 tcp_config = server.config(transport="tcp")
115 assert_eq(tcp_config.host, "127.0.0.1")
116 assert_eq(tcp_config.port, 4306)
117 assert_is_none(tcp_config.unix_socket)
119 with assert_raises(TemporaryMariaDBServerError):
120 _ = TemporaryMariaDBServer(
121 auth="insecure",
122 database="test",
123 data_directory=Path("data"),
124 error_log_path=Path("mariadb.err"),
125 host=None,
126 password="",
127 pid_path=Path("mariadb.pid"),
128 port=None,
129 socket_path=Path("mariadb.sock"),
130 transports=frozenset({"unix_socket"}),
131 user="root",
132 ).config(transport="tcp")
135@test(mark="fast")
136async def temporary_mariadb_server_resolves_relative_data_directories() -> None:
137 """Relative data directories are made absolute before invoking MariaDB."""
139 with TemporaryDirectory() as temporary_directory:
140 base_directory = Path(temporary_directory)
141 install_db = base_directory / "mariadb-install-db"
142 _ = install_db.write_text(
143 "".join(
144 (
145 "#!/bin/sh\n",
146 "printf '%s\\n' \"$@\" >&2\n",
147 "exit 1\n",
148 )
149 ),
150 )
151 _ = install_db.chmod(0o700)
152 relative_data_directory = Path("relative-mariadb-data")
153 expected_data_directory = Path.cwd() / relative_data_directory
155 try:
156 with assert_raises(TemporaryMariaDBServerError) as error:
157 async with temporary_mariadb_server(
158 data_directory=relative_data_directory,
159 install_db=install_db,
160 ):
161 pass
162 finally:
163 await asyncio.to_thread(
164 shutil.rmtree,
165 expected_data_directory,
166 ignore_errors=True,
167 )
169 assert_in(f"--datadir={expected_data_directory}", str(error.exception))
172@test(mark="fast")
173async def temporary_mariadb_server_explains_quota_limited_install_failures() -> None:
174 """Install failures caused by quota exhaustion include cleanup guidance."""
176 with TemporaryDirectory() as temporary_directory:
177 install_db = Path(temporary_directory) / "mariadb-install-db"
178 _ = install_db.write_text(
179 "".join(
180 (
181 "#!/bin/sh\n",
182 "echo 'InnoDB: preallocating 100663296 bytes failed with error 122' >&2\n",
183 "exit 1\n",
184 )
185 ),
186 )
187 _ = install_db.chmod(0o700)
189 data_directory = Path(temporary_directory) / "data"
191 with assert_raises(TemporaryMariaDBServerError) as error:
192 async with temporary_mariadb_server(
193 data_directory=data_directory,
194 install_db=install_db,
195 ):
196 pass
198 message = str(error.exception)
199 assert_in("quota-limited filesystem", message)
200 assert_in("clean retained temporary MariaDB data directories", message)
201 assert_in("data_directory", message)
204@test(mark="medium")
205async def temporary_mariadb_server_starts_with_default_unix_socket() -> None:
206 """The public helper starts a queryable Unix-socket server by default."""
208 data_directory: Path | None = None
209 try:
210 async with temporary_mariadb_server() as server:
211 result = await server.run_sql("SELECT 1")
212 config = server.config()
213 data_directory = server.data_directory
215 assert_eq(server.transports, frozenset({"unix_socket"}))
216 assert_is_none(server.host)
217 assert_is_none(server.port)
218 assert_eq(config.unix_socket, server.socket_path)
219 assert_in("1", result.stdout)
221 assert_true(data_directory.exists())
222 finally:
223 if data_directory is not None:
224 await asyncio.to_thread(shutil.rmtree, data_directory, ignore_errors=True)
227@test(mark="medium")
228async def temporary_mariadb_server_reset_database_drops_reused_tables() -> None:
229 """The public reset helper removes stale tables from a reused database."""
231 with TemporaryDirectory() as temporary_directory:
232 data_directory = Path(temporary_directory) / "data"
233 async with temporary_mariadb_server(data_directory=data_directory) as server:
234 _ = await server.run_sql("CREATE TABLE stale_public_table (`id` INT)")
236 async with temporary_mariadb_server(data_directory=data_directory) as server:
237 await server.reset_database()
238 result = await server.run_sql("SHOW TABLES LIKE 'stale_public_table'")
240 assert_eq(result.stdout, "")
243@test(mark="medium")
244async def temporary_mariadb_server_reset_database_option_runs_before_yield() -> None:
245 """The startup option resets reused data directories before yielding."""
247 with TemporaryDirectory() as temporary_directory:
248 data_directory = Path(temporary_directory) / "data"
249 async with temporary_mariadb_server(data_directory=data_directory) as server:
250 _ = await server.run_sql("CREATE TABLE stale_option_table (`id` INT)")
252 async with temporary_mariadb_server(
253 data_directory=data_directory,
254 reset_database=True,
255 ) as server:
256 result = await server.run_sql("SHOW TABLES LIKE 'stale_option_table'")
258 assert_eq(result.stdout, "")
261@test(mark="medium")
262async def temporary_mariadb_server_supports_password_auth() -> None:
263 """Password-auth test servers generate usable credentials."""
265 with TemporaryDirectory() as temporary_directory:
266 data_directory = Path(temporary_directory) / "data"
267 async with temporary_mariadb_server(
268 auth="password",
269 data_directory=data_directory,
270 ) as server:
271 result = await server.run_sql("SELECT 1")
273 assert_eq(server.auth, "password")
274 assert server.password != ""
275 assert_in("1", result.stdout)