Coverage for snekql/testing/mariadb/_commands.py: 86%

1"""MariaDB client command construction for test-server helpers.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import shlex 

7from collections.abc import Mapping 

8from dataclasses import dataclass 

9from pathlib import Path 

10 

11from snekql.testing.mariadb._types import ( 

12 MariaDBAuth, 

13 MariaDBTransport, 

14 TemporaryMariaDBServerError, 

15) 

16 

17 

18@dataclass(frozen=True, kw_only=True) 

19class MariaDBClientCommand: 

20 """Build executable and renderable MariaDB client commands.""" 

21 

22 auth: MariaDBAuth 

23 client: str | Path 

24 database: str | None 

25 host: str | None 

26 password: str 

27 port: int | None 

28 socket_path: Path | None 

29 transport: MariaDBTransport 

30 user: str 

31 

32 def arguments(self, *, password_prompt: bool = False) -> tuple[str, ...]: 

33 """Build argv while keeping transport rules in one module.""" 

34 

35 arguments = [str(self.client)] 

36 if self.transport == "unix_socket": 

37 if self.socket_path is None: 

38 msg = "unix_socket client command requires socket_path" 

39 raise TemporaryMariaDBServerError(msg) 

40 arguments.extend(("--socket", str(self.socket_path))) 

41 else: 

42 if self.host is None or self.port is None: 

43 msg = "tcp client command requires host and port" 

44 raise TemporaryMariaDBServerError(msg) 

45 arguments.extend(("--protocol=tcp", "-h", self.host, "-P", str(self.port))) 

46 arguments.extend(("-u", self.user)) 

47 if self.auth == "password" and password_prompt: 

48 arguments.append("-p") 

49 if self.database is not None: 

50 arguments.extend(("-D", self.database)) 

51 return tuple(arguments) 

52 

53 def environment(self) -> Mapping[str, str] | None: 

54 """Provide password credentials without exposing them in argv.""" 

55 

56 if self.auth != "password": 

57 return None 

58 environment = os.environ.copy() 

59 environment["MYSQL_PWD"] = self.password 

60 return environment 

61 

62 def shell_command(self) -> str: 

63 """Render a ready-to-copy interactive client command.""" 

64 

65 return " ".join( 

66 shlex.quote(argument) for argument in self.arguments(password_prompt=True) 

67 )