Add Mysql: specialised Container class (664c64ea) · Commits · Dom Sekotill / Behave Utils

behave_utils/init.py

+2 −0

Original line number	Diff line number	Diff line
		@@ -18,6 +18,7 @@ from .json import JSONObject
		from .proc import exec_io
		from .secret import make_secret
		from .url import URL
		from .utils import wait

		__all__ = (
		"JSONArray",
		@@ -28,4 +29,5 @@ __all__ = (
		"make_secret",
		"redirect",
		"register_pattern",
		"wait",
		)

behave_utils/mysql.py

0 → 100644

+122 −0

Original line number	Diff line number	Diff line
		# Copyright 2021 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		# file, You can obtain one at http://mozilla.org/MPL/2.0/.

		"""
		Management and control for MySQL database fixtures
		"""

		from __future__ import annotations

		from contextlib import contextmanager
		from pathlib import Path
		from time import sleep
		from typing import TYPE_CHECKING
		from typing import Any
		from typing import Iterator
		from typing import Sequence

		from behave import fixture

		from .docker import Cli
		from .docker import Container
		from .docker import Image
		from .docker import Mount
		from .docker import Network
		from .secret import make_secret
		from .utils import wait

		if TYPE_CHECKING:
		from behave.runner import FeatureContext


		INIT_DIRECTORY = Path("/docker-entrypoint-initdb.d")


		class Mysql(Container):
		"""
		Container subclass for a database container
		"""

		def __init__(
		self,
		version: str = "latest",
		init_files: Sequence[Path] = [],
		network: Network\|None = None,
		name: str = "test-db",
		user: str = "test-db-user",
		password: str\|None = None,
		):
		self.name = name
		self.user = user
		self.password = password or make_secret(20)
		volumes: list[Mount] = [(path, INIT_DIRECTORY / path.name) for path in init_files]
		volumes.append(Path("/var/lib/mysql"))
		env = dict(
		MYSQL_DATABASE=name,
		MYSQL_USER=user,
		MYSQL_PASSWORD=self.password,
		)
		Container.__init__(
		self,
		Image.pull(f"mysql/mysql-server:{version}"),
		volumes=volumes,
		env=env,
		network=network,
		)

		def get_location(self) -> str:
		"""
		Return a "host:port" string for connecting to the database from other containers
		"""
		host = self.inspect().path("$.Config.Hostname", str)
		return f"{host}:3306"

		@property
		def mysql(self) -> Cli:
		"""
		Run "mysql" commands
		"""
		return Cli(self, "mysql")

		@property
		def mysqladmin(self) -> Cli:
		"""
		Run "mysqladmin" commands
		"""
		return Cli(self, "mysqladmin")

		@property
		def mysqldump(self) -> Cli:
		"""
		Run "mysqldump" commands
		"""
		return Cli(self, "mysqldump")

		@contextmanager
		def started(self) -> Iterator[Container]:
		"""
		Return a context manager that only enters once the database is initialised
		"""
		with self:
		self.start()
		sleep(20)
		wait(lambda: self.run(['/healthcheck.sh']).returncode == 0)
		yield self


		@fixture
		def snapshot_rollback(context: FeatureContext, /, database: Mysql\|None = None, a: Any, *k: Any) -> Iterator[None]:
		"""
		Manage the state of a database as a revertible fixture

		At the end of the fixture's lifetime it's state at the beginning is restored. This
		allows for faster fixture turn-around than restarting the database.
		"""
		assert database is not None, \
		"'database' is required for snapshot_rollback"
		snapshot = database.mysqldump("--all-databases", deserialiser=bytes)
		yield
		database.mysql(input=snapshot)

behave_utils/utils.py

0 → 100644

+33 −0

Original line number	Diff line number	Diff line
		# Copyright 2021 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# This Source Code Form is subject to the terms of the Mozilla Public
		# License, v. 2.0. If a copy of the MPL was not distributed with this
		# file, You can obtain one at http://mozilla.org/MPL/2.0/.

		"""
		Miscellaneous useful functions
		"""

		from __future__ import annotations

		from time import sleep
		from time import time
		from typing import Callable


		def wait(predicate: Callable[[], bool], timeout: float = 120.0) -> None:
		"""
		Block and periodically call "predictate" until it returns True, or the time limit passes
		"""
		end = time() + timeout
		left = timeout
		while left > 0.0:
		sleep(
		10 if left > 60.0 else
		5 if left > 10.0 else
		1,
		)
		left = end - time()
		if predicate():
		return
		raise TimeoutError