30040549735

Committed 23 Jul 2026 08:04PM UTC coverage: 23.663% (-21.5%) from 45.174%

Build # 30040549735

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #106 from smythi93/dev

Release 1.0.0 — complete benchmark coverage (328/328 reproducible bugs)

Coverage Stats

11882 of 64778 new or added lines in 1205 files covered. (18.34%)

46 existing lines in 14 files now uncovered.

18085 of 76426 relevant lines covered (23.66%)

0.24 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

17.97

/src/tests4py/api/docker.py

"""Optional containerised backend for Tests4Py.

Everything here is *additive*: the default host/pyenv backend is untouched and
this module is only exercised through ``t4p docker ...`` or when
``config.mode == "docker"``. It drives a layered image build

    environment image  (tests4py-env)          -- OS + pyenv + Tests4Py
        -> project image (tests4py-project-<p>) -- one bug warmed, deps cached
            -> instance image (tests4py-instance-<p>-<b>) -- one bug ready to run

so that the expensive layers (toolchain, interpreter, dependencies) are built
once and reused by Docker's layer cache across bugs.

The functions shell out to the ``docker`` CLI; they never import a Docker SDK so
there is no new dependency. Each returns a :class:`DockerReport`.
"""

import os
import subprocess
from pathlib import Path
from typing import List, Optional, Sequence

from tests4py.api.config import config_set, load_config
from tests4py.api.report import DockerReport
from tests4py.constants import (
    DOCKER_ENV,
    DOCKER_INSTANCE,
    DOCKER_MODE,
    DOCKER_PROJECT,
    DOCKER_RUN,
)
from tests4py.logger import LOGGER

# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ image naming ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #

ENV_IMAGE = "tests4py-env:latest"


def project_image(project_name: str) -> str:
    return f"tests4py-project-{project_name}:latest"


def instance_image(project_name: str, bug_id: int) -> str:
    return f"tests4py-instance-{project_name}-{bug_id}:latest"


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ helpers ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #


def repo_root() -> Path:
    """Repository root (the Docker build context for the env image).

    Walks up from this package until a ``pyproject.toml`` is found. Works for an
    editable/source install; for a wheel install pass an explicit ``context``.
    """
    here = Path(__file__).resolve()
    for parent in here.parents:
        if (parent / "pyproject.toml").exists():
            return parent
    # Fall back to <site-packages>/tests4py/.. two levels up from this file.
    return here.parent.parent.parent


def docker_available(timeout: float = 8.0) -> bool:
    """True iff the ``docker`` CLI exists and its daemon answers quickly."""
    from shutil import which

    if which("docker") is None:
        return False
    try:
        subprocess.run(
            ["docker", "version", "--format", "{{.Server.Version}}"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=timeout,
            check=True,
        )
        return True
    except (subprocess.SubprocessError, OSError):
        return False


def image_exists(tag: str) -> bool:
    try:
        return (
            subprocess.run(
                ["docker", "image", "inspect", tag],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
            ).returncode
            == 0
        )
    except OSError:
        return False


def _dockerfile(name: str) -> Path:
    from tests4py.docker import (
        DOCKERFILE_ENV,
        DOCKERFILE_INSTANCE,
        DOCKERFILE_PROJECT,
    )

    return {
        "env": DOCKERFILE_ENV,
        "project": DOCKERFILE_PROJECT,
        "instance": DOCKERFILE_INSTANCE,
    }[name]


def _run(
    cmd: Sequence[str], report: DockerReport, verbose: bool = False
) -> DockerReport:
    LOGGER.info("docker: %s", " ".join(str(c) for c in cmd))
    try:
        proc = subprocess.run(
            list(cmd),
            stdout=None if verbose else subprocess.DEVNULL,
            stderr=None if verbose else subprocess.STDOUT,
        )
        report.returncode = proc.returncode
        report.successful = proc.returncode == 0
        if not report.successful:
            report.raised = RuntimeError(
                f"`{' '.join(str(c) for c in cmd[:3])} ...` exited with "
                f"{proc.returncode}"
            )
    except BaseException as e:  # noqa: BLE001 -- surfaced via report.raised
        report.raised = e
        report.successful = False
    return report


def _guard_available(report: DockerReport) -> bool:
    if docker_available():
        return True
    report.successful = False
    report.raised = RuntimeError(
        "Docker is not available (the `docker` CLI is missing or its daemon is "
        "not running). Start Docker and retry, or use the default pyenv backend."
    )
    return False


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ image builds ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #


def build_env(
    tag: str = ENV_IMAGE,
    context: Optional[os.PathLike] = None,
    default_python: str = "3.10.9",
    verbose: bool = False,
    report: Optional[DockerReport] = None,
) -> DockerReport:
    """Build the reusable environment image (OS + pyenv + Tests4Py)."""
    report = report or DockerReport(DOCKER_ENV)
    if not _guard_available(report):
        return report
    context = Path(context) if context else repo_root()
    cmd = [
        "docker",
        "build",
        "-f",
        str(_dockerfile("env")),
        "--build-arg",
        f"DEFAULT_PYTHON={default_python}",
        "-t",
        tag,
        str(context),
    ]
    _run(cmd, report, verbose=verbose)
    report.image = tag if report.successful else None
    return report


def build_project(
    project_name: str,
    warm_bug: int = 1,
    env_image: str = ENV_IMAGE,
    tag: Optional[str] = None,
    context: Optional[os.PathLike] = None,
    build_env_if_missing: bool = True,
    verbose: bool = False,
    report: Optional[DockerReport] = None,
) -> DockerReport:
    """Build a project image warmed by one representative bug."""
    report = report or DockerReport(DOCKER_PROJECT)
    if not _guard_available(report):
        return report
    if build_env_if_missing and not image_exists(env_image):
        env_report = build_env(tag=env_image, context=context, verbose=verbose)
        if not env_report.successful:
            report.successful = False
            report.raised = env_report.raised
            return report
    tag = tag or project_image(project_name)
    context = Path(context) if context else repo_root()
    cmd = [
        "docker",
        "build",
        "-f",
        str(_dockerfile("project")),
        "--build-arg",
        f"ENV_IMAGE={env_image}",
        "--build-arg",
        f"PROJECT={project_name}",
        "--build-arg",
        f"WARM_BUG={warm_bug}",
        "-t",
        tag,
        str(context),
    ]
    _run(cmd, report, verbose=verbose)
    report.image = tag if report.successful else None
    return report


def build_instance(
    project_name: str,
    bug_id: int,
    project_img: Optional[str] = None,
    tag: Optional[str] = None,
    context: Optional[os.PathLike] = None,
    build_project_if_missing: bool = True,
    warm_bug: int = 1,
    verbose: bool = False,
    report: Optional[DockerReport] = None,
) -> DockerReport:
    """Build an instance image for one specific bug, ready to run."""
    report = report or DockerReport(DOCKER_INSTANCE)
    if not _guard_available(report):
        return report
    project_img = project_img or project_image(project_name)
    if build_project_if_missing and not image_exists(project_img):
        proj_report = build_project(
            project_name,
            warm_bug=warm_bug,
            tag=project_img,
            context=context,
            verbose=verbose,
        )
        if not proj_report.successful:
            report.successful = False
            report.raised = proj_report.raised
            return report
    tag = tag or instance_image(project_name, bug_id)
    context = Path(context) if context else repo_root()
    cmd = [
        "docker",
        "build",
        "-f",
        str(_dockerfile("instance")),
        "--build-arg",
        f"PROJECT_IMAGE={project_img}",
        "--build-arg",
        f"PROJECT={project_name}",
        "--build-arg",
        f"BUG={bug_id}",
        "-t",
        tag,
        str(context),
    ]
    _run(cmd, report, verbose=verbose)
    report.image = tag if report.successful else None
    return report


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ run a slice ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #


def run_slice(
    project_name: str,
    bug_id: int,
    command: Optional[List[str]] = None,
    build_if_missing: bool = True,
    docker_run_args: Optional[List[str]] = None,
    verbose: bool = True,
    report: Optional[DockerReport] = None,
) -> DockerReport:
    """Run a Tests4Py command inside a bug's instance container.

    ``command`` is the ``t4p`` argument vector (default: run the failing tests).
    Example: ``run_slice("middle", 1, ["systemtest", "generate", "-n", "4"])``.
    """
    report = report or DockerReport(DOCKER_RUN)
    if not _guard_available(report):
        return report
    tag = instance_image(project_name, bug_id)
    if not image_exists(tag):
        if not build_if_missing:
            report.successful = False
            report.raised = RuntimeError(
                f"Instance image {tag} not found (build it first with "
                f"`t4p docker instance -p {project_name} -i {bug_id}`)."
            )
            return report
        inst = build_instance(project_name, bug_id, verbose=verbose)
        if not inst.successful:
            report.successful = False
            report.raised = inst.raised
            return report
    work = f"/work/{project_name}_{bug_id}"
    t4p_cmd = command or ["test", "-w", work]
    cmd = (
        ["docker", "run", "--rm"]
        + (docker_run_args or [])
        + [tag, "t4p"]
        + t4p_cmd
    )
    _run(cmd, report, verbose=verbose)
    report.image = tag
    return report


# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ mode toggle ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #


def set_mode(mode: str, report: Optional[DockerReport] = None) -> DockerReport:
    """Persist the execution backend (``pyenv`` or ``docker``)."""
    report = report or DockerReport(DOCKER_MODE)
    if mode not in ("pyenv", "docker"):
        report.successful = False
        report.raised = ValueError(f"mode must be 'pyenv' or 'docker', got {mode!r}")
        return report
    config_set("mode", mode)
    report.successful = True
    return report


def current_mode() -> str:
    return load_config().mode

1	"""Optional containerised backend for Tests4Py.
2
3	Everything here is additive: the default host/pyenv backend is untouched and
4	this module is only exercised through ``t4p docker ...`` or when
5	``config.mode == "docker"``. It drives a layered image build
6
7	environment image (tests4py-env) -- OS + pyenv + Tests4Py
8	-> project image (tests4py-project-<p>) -- one bug warmed, deps cached
9	-> instance image (tests4py-instance-<p>-<b>) -- one bug ready to run
10
11	so that the expensive layers (toolchain, interpreter, dependencies) are built
12	once and reused by Docker's layer cache across bugs.
13
14	The functions shell out to the ``docker`` CLI; they never import a Docker SDK so
15	there is no new dependency. Each returns a :class:`DockerReport`.
16	"""
17
18	import os	1✔
19	import subprocess	1✔
20	from pathlib import Path	1✔
21	from typing import List, Optional, Sequence	1✔
22
23	from tests4py.api.config import config_set, load_config	1✔
24	from tests4py.api.report import DockerReport	1✔
25	from tests4py.constants import (	1✔
26	DOCKER_ENV,
27	DOCKER_INSTANCE,
28	DOCKER_MODE,
29	DOCKER_PROJECT,
30	DOCKER_RUN,
31	)
32	from tests4py.logger import LOGGER	1✔
33
34	# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ image naming ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
35
36	ENV_IMAGE = "tests4py-env:latest"	1✔
37
38
39	def project_image(project_name: str) -> str:	1✔
NEW 40	return f"tests4py-project-{project_name}:latest"	×
41
42
43	def instance_image(project_name: str, bug_id: int) -> str:	1✔
NEW 44	return f"tests4py-instance-{project_name}-{bug_id}:latest"	×
45
46
47	# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ helpers ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
48
49
50	def repo_root() -> Path:	1✔
51	"""Repository root (the Docker build context for the env image).
52
53	Walks up from this package until a ``pyproject.toml`` is found. Works for an
54	editable/source install; for a wheel install pass an explicit ``context``.
55	"""
NEW 56	here = Path(__file__).resolve()	×
NEW 57	for parent in here.parents:	×
NEW 58	if (parent / "pyproject.toml").exists():	×
NEW 59	return parent	×
60	# Fall back to <site-packages>/tests4py/.. two levels up from this file.
NEW 61	return here.parent.parent.parent	×
62
63
64	def docker_available(timeout: float = 8.0) -> bool:	1✔
65	"""True iff the ``docker`` CLI exists and its daemon answers quickly."""
NEW 66	from shutil import which	×
67
NEW 68	if which("docker") is None:	×
NEW 69	return False	×
NEW 70	try:	×
NEW 71	subprocess.run(	×
72	["docker", "version", "--format", "{{.Server.Version}}"],
73	stdout=subprocess.DEVNULL,
74	stderr=subprocess.DEVNULL,
75	timeout=timeout,
76	check=True,
77	)
NEW 78	return True	×
NEW 79	except (subprocess.SubprocessError, OSError):	×
NEW 80	return False	×
81
82
83	def image_exists(tag: str) -> bool:	1✔
NEW 84	try:	×
NEW 85	return (	×
86	subprocess.run(
87	["docker", "image", "inspect", tag],
88	stdout=subprocess.DEVNULL,
89	stderr=subprocess.DEVNULL,
90	).returncode
91	== 0
92	)
NEW 93	except OSError:	×
NEW 94	return False	×
95
96
97	def _dockerfile(name: str) -> Path:	1✔
NEW 98	from tests4py.docker import (	×
99	DOCKERFILE_ENV,
100	DOCKERFILE_INSTANCE,
101	DOCKERFILE_PROJECT,
102	)
103
NEW 104	return {	×
105	"env": DOCKERFILE_ENV,
106	"project": DOCKERFILE_PROJECT,
107	"instance": DOCKERFILE_INSTANCE,
108	}[name]
109
110
111	def _run(	1✔
112	cmd: Sequence[str], report: DockerReport, verbose: bool = False
113	) -> DockerReport:
NEW 114	LOGGER.info("docker: %s", " ".join(str(c) for c in cmd))	×
NEW 115	try:	×
NEW 116	proc = subprocess.run(	×
117	list(cmd),
118	stdout=None if verbose else subprocess.DEVNULL,
119	stderr=None if verbose else subprocess.STDOUT,
120	)
NEW 121	report.returncode = proc.returncode	×
NEW 122	report.successful = proc.returncode == 0	×
NEW 123	if not report.successful:	×
NEW 124	report.raised = RuntimeError(	×
125	f"`{' '.join(str(c) for c in cmd[:3])} ...` exited with "
126	f"{proc.returncode}"
127	)
NEW 128	except BaseException as e: # noqa: BLE001 -- surfaced via report.raised	×
NEW 129	report.raised = e	×
NEW 130	report.successful = False	×
NEW 131	return report	×
132
133
134	def _guard_available(report: DockerReport) -> bool:	1✔
NEW 135	if docker_available():	×
NEW 136	return True	×
NEW 137	report.successful = False	×
NEW 138	report.raised = RuntimeError(	×
139	"Docker is not available (the `docker` CLI is missing or its daemon is "
140	"not running). Start Docker and retry, or use the default pyenv backend."
141	)
NEW 142	return False	×
143
144
145	# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ image builds ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
146
147
148	def build_env(	1✔
149	tag: str = ENV_IMAGE,
150	context: Optional[os.PathLike] = None,
151	default_python: str = "3.10.9",
152	verbose: bool = False,
153	report: Optional[DockerReport] = None,
154	) -> DockerReport:
155	"""Build the reusable environment image (OS + pyenv + Tests4Py)."""
NEW 156	report = report or DockerReport(DOCKER_ENV)	×
NEW 157	if not _guard_available(report):	×
NEW 158	return report	×
NEW 159	context = Path(context) if context else repo_root()	×
NEW 160	cmd = [	×
161	"docker",
162	"build",
163	"-f",
164	str(_dockerfile("env")),
165	"--build-arg",
166	f"DEFAULT_PYTHON={default_python}",
167	"-t",
168	tag,
169	str(context),
170	]
NEW 171	_run(cmd, report, verbose=verbose)	×
NEW 172	report.image = tag if report.successful else None	×
NEW 173	return report	×
174
175
176	def build_project(	1✔
177	project_name: str,
178	warm_bug: int = 1,
179	env_image: str = ENV_IMAGE,
180	tag: Optional[str] = None,
181	context: Optional[os.PathLike] = None,
182	build_env_if_missing: bool = True,
183	verbose: bool = False,
184	report: Optional[DockerReport] = None,
185	) -> DockerReport:
186	"""Build a project image warmed by one representative bug."""
NEW 187	report = report or DockerReport(DOCKER_PROJECT)	×
NEW 188	if not _guard_available(report):	×
NEW 189	return report	×
NEW 190	if build_env_if_missing and not image_exists(env_image):	×
NEW 191	env_report = build_env(tag=env_image, context=context, verbose=verbose)	×
NEW 192	if not env_report.successful:	×
NEW 193	report.successful = False	×
NEW 194	report.raised = env_report.raised	×
NEW 195	return report	×
NEW 196	tag = tag or project_image(project_name)	×
NEW 197	context = Path(context) if context else repo_root()	×
NEW 198	cmd = [	×
199	"docker",
200	"build",
201	"-f",
202	str(_dockerfile("project")),
203	"--build-arg",
204	f"ENV_IMAGE={env_image}",
205	"--build-arg",
206	f"PROJECT={project_name}",
207	"--build-arg",
208	f"WARM_BUG={warm_bug}",
209	"-t",
210	tag,
211	str(context),
212	]
NEW 213	_run(cmd, report, verbose=verbose)	×
NEW 214	report.image = tag if report.successful else None	×
NEW 215	return report	×
216
217
218	def build_instance(	1✔
219	project_name: str,
220	bug_id: int,
221	project_img: Optional[str] = None,
222	tag: Optional[str] = None,
223	context: Optional[os.PathLike] = None,
224	build_project_if_missing: bool = True,
225	warm_bug: int = 1,
226	verbose: bool = False,
227	report: Optional[DockerReport] = None,
228	) -> DockerReport:
229	"""Build an instance image for one specific bug, ready to run."""
NEW 230	report = report or DockerReport(DOCKER_INSTANCE)	×
NEW 231	if not _guard_available(report):	×
NEW 232	return report	×
NEW 233	project_img = project_img or project_image(project_name)	×
NEW 234	if build_project_if_missing and not image_exists(project_img):	×
NEW 235	proj_report = build_project(	×
236	project_name,
237	warm_bug=warm_bug,
238	tag=project_img,
239	context=context,
240	verbose=verbose,
241	)
NEW 242	if not proj_report.successful:	×
NEW 243	report.successful = False	×
NEW 244	report.raised = proj_report.raised	×
NEW 245	return report	×
NEW 246	tag = tag or instance_image(project_name, bug_id)	×
NEW 247	context = Path(context) if context else repo_root()	×
NEW 248	cmd = [	×
249	"docker",
250	"build",
251	"-f",
252	str(_dockerfile("instance")),
253	"--build-arg",
254	f"PROJECT_IMAGE={project_img}",
255	"--build-arg",
256	f"PROJECT={project_name}",
257	"--build-arg",
258	f"BUG={bug_id}",
259	"-t",
260	tag,
261	str(context),
262	]
NEW 263	_run(cmd, report, verbose=verbose)	×
NEW 264	report.image = tag if report.successful else None	×
NEW 265	return report	×
266
267
268	# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ run a slice ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
269
270
271	def run_slice(	1✔
272	project_name: str,
273	bug_id: int,
274	command: Optional[List[str]] = None,
275	build_if_missing: bool = True,
276	docker_run_args: Optional[List[str]] = None,
277	verbose: bool = True,
278	report: Optional[DockerReport] = None,
279	) -> DockerReport:
280	"""Run a Tests4Py command inside a bug's instance container.
281
282	``command`` is the ``t4p`` argument vector (default: run the failing tests).
283	Example: ``run_slice("middle", 1, ["systemtest", "generate", "-n", "4"])``.
284	"""
NEW 285	report = report or DockerReport(DOCKER_RUN)	×
NEW 286	if not _guard_available(report):	×
NEW 287	return report	×
NEW 288	tag = instance_image(project_name, bug_id)	×
NEW 289	if not image_exists(tag):	×
NEW 290	if not build_if_missing:	×
NEW 291	report.successful = False	×
NEW 292	report.raised = RuntimeError(	×
293	f"Instance image {tag} not found (build it first with "
294	f"`t4p docker instance -p {project_name} -i {bug_id}`)."
295	)
NEW 296	return report	×
NEW 297	inst = build_instance(project_name, bug_id, verbose=verbose)	×
NEW 298	if not inst.successful:	×
NEW 299	report.successful = False	×
NEW 300	report.raised = inst.raised	×
NEW 301	return report	×
NEW 302	work = f"/work/{project_name}_{bug_id}"	×
NEW 303	t4p_cmd = command or ["test", "-w", work]	×
NEW 304	cmd = (	×
305	["docker", "run", "--rm"]
306	+ (docker_run_args or [])
307	+ [tag, "t4p"]
308	+ t4p_cmd
309	)
NEW 310	_run(cmd, report, verbose=verbose)	×
NEW 311	report.image = tag	×
NEW 312	return report	×
313
314
315	# ~~~~~~~~~~~~~~~~~~~~~~~~~~~ mode toggle ~~~~~~~~~~~~~~~~~~~~~~~~~~~ #
316
317
318	def set_mode(mode: str, report: Optional[DockerReport] = None) -> DockerReport:	1✔
319	"""Persist the execution backend (``pyenv`` or ``docker``)."""
NEW 320	report = report or DockerReport(DOCKER_MODE)	×
NEW 321	if mode not in ("pyenv", "docker"):	×
NEW 322	report.successful = False	×
NEW 323	report.raised = ValueError(f"mode must be 'pyenv' or 'docker', got {mode!r}")	×
NEW 324	return report	×
NEW 325	config_set("mode", mode)	×
NEW 326	report.successful = True	×
NEW 327	return report	×
328
329
330	def current_mode() -> str:	1✔
NEW 331	return load_config().mode	×

smythi93 / Tests4Py / 30040549735

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous