Coverage for snekql/testing/mariadb/server.py: 84%
310 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-07 21:13 +0300
1"""Temporary MariaDB Test Server support for integration tests."""
3from __future__ import annotations
5import asyncio
6import secrets
7import shutil
8import socket
9from collections.abc import AsyncGenerator, Mapping
10from contextlib import AbstractAsyncContextManager, asynccontextmanager
11from dataclasses import dataclass, field
12from pathlib import Path
13from tempfile import mkdtemp
15from snekql import mariadb
16from snekql.testing.mariadb._commands import MariaDBClientCommand
17from snekql.testing.mariadb._types import (
18 MariaDBAuth,
19 MariaDBCommandResult,
20 MariaDBTransport,
21 TemporaryMariaDBServerError,
22)
24_DEFAULT_DATABASE = "test"
25_DEFAULT_HOST = "127.0.0.1"
26_DEFAULT_TRANSPORTS: frozenset[MariaDBTransport] = frozenset({"unix_socket"})
27_IDENTIFIER_MAX_LENGTH = 64
28_MANAGED_SERVER_OPTIONS = frozenset(
29 {
30 "--bind-address",
31 "--datadir",
32 "--log-error",
33 "--no-defaults",
34 "--pid-file",
35 "--port",
36 "--skip-grant-tables",
37 "--skip-networking",
38 "--socket",
39 }
40)
41_RESET_DATABASE_SQL = """
42SET SESSION group_concat_max_len = 1000000;
43SET FOREIGN_KEY_CHECKS = 0;
44SELECT GROUP_CONCAT(CONCAT('`', REPLACE(TABLE_NAME, '`', '``'), '`'))
45INTO @snekql_tables_to_drop
46FROM information_schema.TABLES
47WHERE TABLE_SCHEMA = DATABASE()
48 AND TABLE_TYPE = 'BASE TABLE';
49SET @snekql_drop_tables = IF(
50 @snekql_tables_to_drop IS NULL,
51 'DO 0',
52 CONCAT('DROP TABLE ', @snekql_tables_to_drop)
53);
54PREPARE snekql_drop_tables_statement FROM @snekql_drop_tables;
55EXECUTE snekql_drop_tables_statement;
56DEALLOCATE PREPARE snekql_drop_tables_statement;
57SET FOREIGN_KEY_CHECKS = 1;
58"""
59_SHUTDOWN_TIMEOUT = 10.0
62@dataclass(frozen=True, kw_only=True)
63class TemporaryMariaDBServer:
64 """Connection details for a local Temporary MariaDB Test Server.
66 >>> server = TemporaryMariaDBServer(
67 ... auth="insecure",
68 ... database="test",
69 ... data_directory=Path("data"),
70 ... error_log_path=Path("mariadb.err"),
71 ... host=None,
72 ... password="",
73 ... pid_path=Path("mariadb.pid"),
74 ... port=None,
75 ... socket_path=Path("mariadb.sock"),
76 ... transports=frozenset({"unix_socket"}),
77 ... user="root",
78 ... )
79 >>> server.config().unix_socket
80 PosixPath('mariadb.sock')
81 """
83 auth: MariaDBAuth
84 database: str
85 data_directory: Path
86 error_log_path: Path
87 host: str | None
88 password: str
89 pid_path: Path
90 port: int | None
91 socket_path: Path | None
92 transports: frozenset[MariaDBTransport]
93 user: str
94 _client: str | Path = field(
95 default="mariadb",
96 init=False,
97 repr=False,
98 compare=False,
99 )
101 def config(
102 self,
103 *,
104 transport: MariaDBTransport | None = None,
105 pool_size: int = 5,
106 acquire_timeout: float = 30.0,
107 charset: str = "utf8mb4",
108 ) -> mariadb.Config:
109 """Build a snekql MariaDB runtime config for this test server."""
111 selected_transport = self._select_transport(transport)
112 if selected_transport == "unix_socket":
113 if self.socket_path is None:
114 msg = (
115 "Temporary MariaDB Test Server did not expose unix_socket transport"
116 )
117 raise TemporaryMariaDBServerError(msg)
118 return mariadb.Config(
119 acquire_timeout=acquire_timeout,
120 charset=charset,
121 database=self.database,
122 password=self.password,
123 pool_size=pool_size,
124 unix_socket=self.socket_path,
125 user=self.user,
126 )
127 if self.host is None or self.port is None:
128 msg = "Temporary MariaDB Test Server did not expose tcp transport"
129 raise TemporaryMariaDBServerError(msg)
130 return mariadb.Config(
131 acquire_timeout=acquire_timeout,
132 charset=charset,
133 database=self.database,
134 host=self.host,
135 password=self.password,
136 pool_size=pool_size,
137 port=self.port,
138 user=self.user,
139 )
141 async def reset_database(
142 self,
143 *,
144 transport: MariaDBTransport | None = None,
145 ) -> None:
146 """Drop all base tables from the configured test database."""
148 _ = await self.run_sql(_RESET_DATABASE_SQL, transport=transport)
150 async def run_sql(
151 self,
152 sql: str,
153 *,
154 transport: MariaDBTransport | None = None,
155 check: bool = True,
156 ) -> MariaDBCommandResult:
157 """Execute SQL through the MariaDB command-line client."""
159 selected_transport = self._select_transport(transport)
160 result = await _run_client_sql(
161 auth=self.auth,
162 client=self._client,
163 database=self.database,
164 host=self.host,
165 password=self.password,
166 port=self.port,
167 socket_path=self.socket_path,
168 sql=sql,
169 transport=selected_transport,
170 user=self.user,
171 )
172 if check and result.returncode != 0:
173 msg = f"mariadb command failed while executing SQL\n{result.stderr}"
174 raise TemporaryMariaDBServerError(msg)
175 return result
177 def _select_transport(self, transport: MariaDBTransport | None) -> MariaDBTransport:
178 """Prefer Unix socket while rejecting disabled explicit transports."""
180 if transport is None:
181 if "unix_socket" in self.transports:
182 return "unix_socket"
183 if "tcp" in self.transports:
184 return "tcp"
185 msg = "Temporary MariaDB Test Server has no enabled transports"
186 raise TemporaryMariaDBServerError(msg)
187 if transport not in {"unix_socket", "tcp"}:
188 msg = f"unsupported MariaDB test-server transport: {transport!r}"
189 raise TemporaryMariaDBServerError(msg)
190 if transport not in self.transports:
191 msg = f"Temporary MariaDB Test Server did not expose {transport} transport"
192 raise TemporaryMariaDBServerError(msg)
193 return transport
196@dataclass(frozen=True, kw_only=True)
197class _ProcessPaths:
198 """Filesystem paths that tie one mariadbd process lifecycle together."""
200 data_directory: Path
201 error_log_path: Path
202 internal_socket_path: Path
203 pid_path: Path
204 public_socket_path: Path | None
205 runtime_directory: Path
208@dataclass(frozen=True, kw_only=True)
209class _TemporaryMariaDBServerOptions:
210 """Validated startup options for a Temporary MariaDB Test Server."""
212 auth: MariaDBAuth
213 clean_before_start: bool
214 client: str | Path
215 data_directory: Path | None
216 database: str
217 install_db: str | Path
218 mariadbd: str | Path
219 password: str | None
220 port: int | None
221 reset_database: bool
222 server_args: tuple[str, ...]
223 socket_path: Path | None
224 startup_timeout: float
225 transports: frozenset[MariaDBTransport]
226 user: str
229@dataclass(frozen=True, kw_only=True)
230class _StartupPlan:
231 """Resolved lifecycle facts for one Temporary MariaDB Test Server run."""
233 options: _TemporaryMariaDBServerOptions
234 password: str
235 paths: _ProcessPaths
236 port: int | None
237 readiness_transport: MariaDBTransport
239 def server(self) -> TemporaryMariaDBServer:
240 """Build the public server value from resolved startup facts."""
242 server = TemporaryMariaDBServer(
243 auth=self.options.auth,
244 database=self.options.database,
245 data_directory=self.paths.data_directory,
246 error_log_path=self.paths.error_log_path,
247 host=_DEFAULT_HOST if "tcp" in self.options.transports else None,
248 password=self.password,
249 pid_path=self.paths.pid_path,
250 port=self.port,
251 socket_path=self.paths.public_socket_path,
252 transports=self.options.transports,
253 user=self.options.user,
254 )
255 object.__setattr__(server, "_client", self.options.client)
256 return server
258 @property
259 def tcp_enabled(self) -> bool:
260 """Whether this plan exposes local TCP transport."""
262 return "tcp" in self.options.transports
265def _resolve_user_path(path: Path) -> Path:
266 """Make caller-provided paths stable for MariaDB helper processes."""
268 return path.expanduser().resolve()
271async def _create_process_paths(
272 options: _TemporaryMariaDBServerOptions,
273) -> _ProcessPaths:
274 """Create retained runtime paths without deleting them at shutdown."""
276 runtime_directory = Path(mkdtemp(prefix="snekql-mariadb-"))
277 data_directory = (
278 _resolve_user_path(options.data_directory)
279 if options.data_directory is not None
280 else runtime_directory / "data"
281 )
282 public_socket_path = None
283 if "unix_socket" in options.transports:
284 public_socket_path = (
285 _resolve_user_path(options.socket_path)
286 if options.socket_path is not None
287 else runtime_directory / "mariadb.sock"
288 )
289 internal_socket_path = public_socket_path or runtime_directory / "internal.sock"
290 return _ProcessPaths(
291 data_directory=data_directory,
292 error_log_path=runtime_directory / "mariadb.err",
293 internal_socket_path=internal_socket_path,
294 pid_path=runtime_directory / "mariadb.pid",
295 public_socket_path=public_socket_path,
296 runtime_directory=runtime_directory,
297 )
300async def _ensure_data_directory(
301 *,
302 options: _TemporaryMariaDBServerOptions,
303 paths: _ProcessPaths,
304) -> None:
305 """Initialize missing/empty data directories and reject invalid reuse."""
307 if options.clean_before_start and paths.data_directory.exists():
308 await asyncio.to_thread(shutil.rmtree, paths.data_directory)
309 if not paths.data_directory.exists():
310 await asyncio.to_thread(paths.data_directory.mkdir, parents=True)
311 await _initialize_data_directory(options=options, paths=paths)
312 return
313 children = await asyncio.to_thread(lambda: tuple(paths.data_directory.iterdir()))
314 if len(children) == 0:
315 await _initialize_data_directory(options=options, paths=paths)
316 return
317 if (paths.data_directory / "mysql").is_dir():
318 return
319 msg = f"MariaDB data_directory is not empty and does not look initialized: {paths.data_directory}"
320 raise TemporaryMariaDBServerError(msg)
323async def _build_startup_plan(
324 options: _TemporaryMariaDBServerOptions,
325) -> _StartupPlan:
326 """Resolve one startup plan before process lifecycle work begins."""
328 paths = await _create_process_paths(options)
329 await _ensure_data_directory(options=options, paths=paths)
330 password = options.password or (
331 secrets.token_urlsafe(24) if options.auth == "password" else ""
332 )
333 port = None
334 if "tcp" in options.transports:
335 port = options.port or _find_free_tcp_port()
336 readiness_transport: MariaDBTransport = (
337 "unix_socket" if "unix_socket" in options.transports else "tcp"
338 )
339 return _StartupPlan(
340 options=options,
341 password=password,
342 paths=paths,
343 port=port,
344 readiness_transport=readiness_transport,
345 )
348def _is_quota_limited_install_error(stderr: str) -> bool:
349 """Recognize MariaDB's quota-exhaustion initialization failure."""
351 return "error 122" in stderr and "preallocating" in stderr
354async def _initialize_data_directory(
355 *,
356 options: _TemporaryMariaDBServerOptions,
357 paths: _ProcessPaths,
358) -> None:
359 """Create MariaDB system tables in a retained data directory."""
361 result = await _run_command(
362 str(options.install_db),
363 "--no-defaults",
364 f"--datadir={paths.data_directory}",
365 "--auth-root-authentication-method=normal",
366 "--skip-test-db",
367 )
368 if result.returncode != 0:
369 message = f"mariadb-install-db failed\n{result.stderr}"
370 if _is_quota_limited_install_error(result.stderr):
371 message += (
372 "\nMariaDB reported error 122 while preallocating files. "
373 "This usually means the data_directory is on a full or "
374 "quota-limited filesystem. Pass data_directory on a filesystem "
375 "with enough free space, or clean retained temporary MariaDB "
376 "data directories such as /tmp/snekql-mariadb-* before retrying."
377 )
378 raise TemporaryMariaDBServerError(message)
381async def _start_process(
382 *,
383 options: _TemporaryMariaDBServerOptions,
384 paths: _ProcessPaths,
385 port: int | None,
386 skip_grant_tables: bool,
387 tcp_enabled: bool,
388) -> asyncio.subprocess.Process:
389 """Start mariadbd with managed local-test lifecycle arguments."""
391 arguments = [
392 str(options.mariadbd),
393 "--no-defaults",
394 f"--datadir={paths.data_directory}",
395 f"--socket={paths.internal_socket_path}",
396 f"--pid-file={paths.pid_path}",
397 f"--log-error={paths.error_log_path}",
398 ]
399 if tcp_enabled:
400 if port is None:
401 msg = "internal error: TCP startup requires a port"
402 raise TemporaryMariaDBServerError(msg)
403 arguments.extend(
404 (
405 f"--port={port}",
406 f"--bind-address={_DEFAULT_HOST}",
407 "--skip-networking=0",
408 )
409 )
410 else:
411 arguments.append("--skip-networking=1")
412 if skip_grant_tables:
413 arguments.append("--skip-grant-tables")
414 arguments.extend(options.server_args)
415 try:
416 return await asyncio.create_subprocess_exec(
417 *arguments,
418 stderr=asyncio.subprocess.DEVNULL,
419 stdout=asyncio.subprocess.DEVNULL,
420 )
421 except OSError as error:
422 msg = f"failed to start mariadbd: {error}"
423 raise TemporaryMariaDBServerError(msg) from error
426async def _stop_process(process: asyncio.subprocess.Process) -> None:
427 """Terminate a child server process without deleting retained data."""
429 if process.returncode is not None:
430 return
431 process.terminate()
432 try:
433 _ = await asyncio.wait_for(process.wait(), timeout=_SHUTDOWN_TIMEOUT)
434 except TimeoutError:
435 process.kill()
436 _ = await process.wait()
439async def _wait_until_ready( # noqa: PLR0913
440 *,
441 client: str | Path,
442 database: str | None,
443 host: str | None,
444 password: str,
445 port: int | None,
446 process: asyncio.subprocess.Process,
447 socket_path: Path | None,
448 startup_timeout: float,
449 transport: MariaDBTransport,
450 user: str,
451 auth: MariaDBAuth,
452 error_log_path: Path,
453) -> None:
454 """Poll the MariaDB CLI until the server accepts local connections."""
456 loop = asyncio.get_running_loop()
457 deadline = loop.time() + startup_timeout
458 while loop.time() < deadline:
459 if process.returncode is not None:
460 error_log = await _read_error_log(error_log_path)
461 msg = f"mariadbd exited before becoming ready\n{error_log}"
462 raise TemporaryMariaDBServerError(msg)
463 result = await _run_client_sql(
464 auth=auth,
465 client=client,
466 database=database,
467 host=host,
468 password=password,
469 port=port,
470 socket_path=socket_path,
471 sql="SELECT 1",
472 transport=transport,
473 user=user,
474 )
475 if result.returncode == 0:
476 return
477 await asyncio.sleep(0.25)
478 error_log = await _read_error_log(error_log_path)
479 msg = f"mariadbd did not become ready\n{error_log}"
480 raise TemporaryMariaDBServerError(msg)
483async def _read_error_log(error_log_path: Path) -> str:
484 """Read MariaDB's error log only when it exists."""
486 exists = await asyncio.to_thread(error_log_path.exists)
487 if not exists:
488 return ""
489 return await asyncio.to_thread(error_log_path.read_text)
492async def _run_command(
493 *arguments: str,
494 env: Mapping[str, str] | None = None,
495) -> MariaDBCommandResult:
496 """Run one subprocess command and capture decoded output."""
498 try:
499 process = await asyncio.create_subprocess_exec(
500 *arguments,
501 env=dict(env) if env is not None else None,
502 stderr=asyncio.subprocess.PIPE,
503 stdout=asyncio.subprocess.PIPE,
504 )
505 except OSError as error:
506 command = " ".join(arguments)
507 msg = f"failed to run command: {command}: {error}"
508 raise TemporaryMariaDBServerError(msg) from error
509 stdout, stderr = await process.communicate()
510 return MariaDBCommandResult(
511 returncode=process.returncode if process.returncode is not None else -1,
512 stderr=stderr.decode(),
513 stdout=stdout.decode(),
514 )
517async def _run_client_sql( # noqa: PLR0913
518 *,
519 auth: MariaDBAuth,
520 client: str | Path,
521 database: str | None,
522 host: str | None,
523 password: str,
524 port: int | None,
525 socket_path: Path | None,
526 sql: str,
527 transport: MariaDBTransport,
528 user: str,
529) -> MariaDBCommandResult:
530 """Execute SQL through the configured MariaDB CLI transport."""
532 command = MariaDBClientCommand(
533 auth=auth,
534 client=client,
535 database=database,
536 host=host,
537 password=password,
538 port=port,
539 socket_path=socket_path,
540 transport=transport,
541 user=user,
542 )
543 return await _run_command(
544 *command.arguments(),
545 "-e",
546 sql,
547 env=command.environment(),
548 )
551async def _create_database(
552 *,
553 plan: _StartupPlan,
554 process: asyncio.subprocess.Process,
555) -> None:
556 """Create the requested test database after the server is reachable."""
558 result = await _run_client_sql(
559 auth=plan.options.auth,
560 client=plan.options.client,
561 database=None,
562 host=_DEFAULT_HOST if plan.port is not None else None,
563 password=plan.password,
564 port=plan.port,
565 socket_path=(
566 plan.paths.internal_socket_path
567 if plan.readiness_transport == "unix_socket"
568 else None
569 ),
570 sql=f"CREATE DATABASE IF NOT EXISTS `{plan.options.database}`",
571 transport=plan.readiness_transport,
572 user=plan.options.user,
573 )
574 if result.returncode != 0:
575 await _stop_process(process)
576 msg = f"failed to create MariaDB test database\n{result.stderr}"
577 raise TemporaryMariaDBServerError(msg)
580async def _bootstrap_password_auth(plan: _StartupPlan) -> None:
581 """Use a short insecure local bootstrap server to set password auth."""
583 process = await _start_process(
584 options=plan.options,
585 paths=plan.paths,
586 port=None,
587 skip_grant_tables=True,
588 tcp_enabled=False,
589 )
590 try:
591 await _wait_until_ready(
592 auth="insecure",
593 client=plan.options.client,
594 database=None,
595 error_log_path=plan.paths.error_log_path,
596 host=None,
597 password="",
598 port=None,
599 process=process,
600 socket_path=plan.paths.internal_socket_path,
601 startup_timeout=plan.options.startup_timeout,
602 transport="unix_socket",
603 user="root",
604 )
605 bootstrap_sql = _password_bootstrap_sql(
606 database=plan.options.database,
607 password=plan.password,
608 user=plan.options.user,
609 )
610 result = await _run_client_sql(
611 auth="insecure",
612 client=plan.options.client,
613 database=None,
614 host=None,
615 password="",
616 port=None,
617 socket_path=plan.paths.internal_socket_path,
618 sql=bootstrap_sql,
619 transport="unix_socket",
620 user="root",
621 )
622 if result.returncode != 0:
623 msg = f"failed to bootstrap MariaDB password auth\n{result.stderr}"
624 raise TemporaryMariaDBServerError(msg)
625 finally:
626 await _stop_process(process)
629async def _start_ready_server(plan: _StartupPlan) -> asyncio.subprocess.Process:
630 """Start the final server and wait on its preferred public transport."""
632 process = await _start_process(
633 options=plan.options,
634 paths=plan.paths,
635 port=plan.port,
636 skip_grant_tables=plan.options.auth == "insecure",
637 tcp_enabled=plan.tcp_enabled,
638 )
639 try:
640 await _wait_until_ready(
641 auth=plan.options.auth,
642 client=plan.options.client,
643 database=None,
644 error_log_path=plan.paths.error_log_path,
645 host=_DEFAULT_HOST if plan.readiness_transport == "tcp" else None,
646 password=plan.password,
647 port=plan.port if plan.readiness_transport == "tcp" else None,
648 process=process,
649 socket_path=(
650 plan.paths.public_socket_path
651 if plan.readiness_transport == "unix_socket"
652 else None
653 ),
654 startup_timeout=plan.options.startup_timeout,
655 transport=plan.readiness_transport,
656 user=plan.options.user,
657 )
658 if plan.options.auth == "insecure":
659 await _create_database(plan=plan, process=process)
660 except TemporaryMariaDBServerError:
661 await _stop_process(process)
662 raise
663 else:
664 return process
667@asynccontextmanager
668async def _temporary_mariadb_server_context(
669 options: _TemporaryMariaDBServerOptions,
670) -> AsyncGenerator[TemporaryMariaDBServer]:
671 """Manage the full MariaDB child process lifecycle."""
673 plan = await _build_startup_plan(options)
674 if options.auth == "password":
675 await _bootstrap_password_auth(plan)
676 process = await _start_ready_server(plan)
677 server = plan.server()
678 if options.reset_database:
679 await server.reset_database()
680 try:
681 yield server
682 finally:
683 await _stop_process(process)
686def _contains_managed_server_option(server_args: tuple[str, ...]) -> str | None:
687 """Detect raw mariadbd arguments that would override managed behavior."""
689 for argument in server_args:
690 option = argument.split("=", maxsplit=1)[0]
691 if option in _MANAGED_SERVER_OPTIONS:
692 return option
693 return None
696def _find_free_tcp_port() -> int:
697 """Reserve a local TCP port long enough to learn an available number."""
699 with socket.socket() as server_socket:
700 server_socket.bind((_DEFAULT_HOST, 0))
701 return int(server_socket.getsockname()[1])
704def _is_simple_identifier(value: str) -> bool:
705 """Validate the intentionally narrow bootstrap identifier subset."""
707 return 0 < len(value) <= _IDENTIFIER_MAX_LENGTH and all(
708 character.isalnum() or character == "_" for character in value
709 )
712def _normalize_transports(
713 transports: set[MariaDBTransport] | None,
714) -> frozenset[MariaDBTransport]:
715 """Normalize omitted transports to Unix socket and reject invalid values."""
717 if transports is None:
718 return _DEFAULT_TRANSPORTS
719 normalized = frozenset(transports)
720 if len(normalized) == 0:
721 msg = "Temporary MariaDB Test Server requires at least one transport"
722 raise TemporaryMariaDBServerError(msg)
723 for transport in normalized:
724 if transport not in {"unix_socket", "tcp"}:
725 msg = f"unsupported MariaDB test-server transport: {transport!r}"
726 raise TemporaryMariaDBServerError(msg)
727 return normalized
730def _password_bootstrap_sql(*, database: str, password: str, user: str) -> str:
731 """Build idempotent bootstrap SQL for the narrow validated identifier set."""
733 escaped_password = password.replace("'", "''")
734 if user == "root":
735 user_sql = f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{escaped_password}';"
736 else:
737 user_sql = "".join(
738 (
739 f"CREATE USER IF NOT EXISTS '{user}'@'localhost' IDENTIFIED BY '{escaped_password}';",
740 f"ALTER USER '{user}'@'localhost' IDENTIFIED BY '{escaped_password}';",
741 f"GRANT ALL PRIVILEGES ON `{database}`.* TO '{user}'@'localhost';",
742 f"CREATE USER IF NOT EXISTS '{user}'@'%' IDENTIFIED BY '{escaped_password}';",
743 f"ALTER USER '{user}'@'%' IDENTIFIED BY '{escaped_password}';",
744 f"GRANT ALL PRIVILEGES ON `{database}`.* TO '{user}'@'%';",
745 )
746 )
747 return "".join(
748 (
749 "FLUSH PRIVILEGES;",
750 f"CREATE DATABASE IF NOT EXISTS `{database}`;",
751 user_sql,
752 "FLUSH PRIVILEGES;",
753 )
754 )
757def _validate_options(options: _TemporaryMariaDBServerOptions) -> None:
758 """Reject option combinations that cannot honor the public contract."""
760 if options.auth not in {"insecure", "password"}:
761 msg = f"unsupported MariaDB test-server auth policy: {options.auth!r}"
762 raise TemporaryMariaDBServerError(msg)
763 if options.clean_before_start and options.data_directory is None:
764 msg = "clean_before_start requires data_directory"
765 raise TemporaryMariaDBServerError(msg)
766 if options.clean_before_start and options.reset_database:
767 msg = "reset_database is incompatible with clean_before_start"
768 raise TemporaryMariaDBServerError(msg)
769 if options.port is not None and "tcp" not in options.transports:
770 msg = "port requires tcp transport"
771 raise TemporaryMariaDBServerError(msg)
772 if options.socket_path is not None and "unix_socket" not in options.transports:
773 msg = "socket_path requires unix_socket transport"
774 raise TemporaryMariaDBServerError(msg)
775 if options.password is not None and options.auth != "password":
776 msg = "password requires auth='password'"
777 raise TemporaryMariaDBServerError(msg)
778 if not _is_simple_identifier(options.database):
779 msg = "MariaDB database must be a non-empty alphanumeric or underscore identifier up to 64 characters"
780 raise TemporaryMariaDBServerError(msg)
781 if not _is_simple_identifier(options.user):
782 msg = "MariaDB user must be a non-empty alphanumeric or underscore identifier up to 64 characters"
783 raise TemporaryMariaDBServerError(msg)
784 managed_option = _contains_managed_server_option(options.server_args)
785 if managed_option is not None:
786 msg = f"server_args contains managed mariadbd option: {managed_option}"
787 raise TemporaryMariaDBServerError(msg)
790def temporary_mariadb_server( # noqa: PLR0913
791 *,
792 auth: MariaDBAuth = "insecure",
793 transports: set[MariaDBTransport] | None = None,
794 data_directory: Path | None = None,
795 clean_before_start: bool = False,
796 reset_database: bool = False,
797 database: str = _DEFAULT_DATABASE,
798 user: str = "root",
799 password: str | None = None,
800 port: int | None = None,
801 socket_path: Path | None = None,
802 server_args: tuple[str, ...] = (),
803 mariadbd: str | Path = "mariadbd",
804 install_db: str | Path = "mariadb-install-db",
805 client: str | Path = "mariadb",
806 startup_timeout: float = 20.0,
807) -> AbstractAsyncContextManager[TemporaryMariaDBServer]:
808 """Start a local Temporary MariaDB Test Server."""
810 options = _TemporaryMariaDBServerOptions(
811 auth=auth,
812 clean_before_start=clean_before_start,
813 client=client,
814 data_directory=data_directory,
815 database=database,
816 install_db=install_db,
817 mariadbd=mariadbd,
818 password=password,
819 port=port,
820 reset_database=reset_database,
821 server_args=server_args,
822 socket_path=socket_path,
823 startup_timeout=startup_timeout,
824 transports=_normalize_transports(transports),
825 user=user,
826 )
827 _validate_options(options)
828 return _temporary_mariadb_server_context(options)
831__all__ = [
832 "MariaDBAuth",
833 "MariaDBCommandResult",
834 "MariaDBTransport",
835 "TemporaryMariaDBServer",
836 "TemporaryMariaDBServerError",
837 "temporary_mariadb_server",
838]