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

63.96

/src/tests4py/api/default.py

import importlib.resources
import os
import shutil
import subprocess
from pathlib import Path
from typing import List, Union, Optional

from tabulate import tabulate

from tests4py.api.cache import (
    check_cache_exists,
    copy_cached,
    cache,
    check_cache_exists_env,
    copy_cached_env,
    cache_venv,
)
from tests4py.api.config import load_config
from tests4py.api.report import CheckoutReport, CompileReport, InfoReport, TestReport
from tests4py.api.test import get_pytest_result, get_test_results
from tests4py.api.utils import get_work_dir, load_project
from tests4py.constants import (
    DEFAULT_WORK_DIR,
    FIX_FILES,
    INFO_FILE,
    REQUIREMENTS_FILE,
    HARNESS_FILE,
    PATCH_FILE,
    EXPLANATION_FILE,
    DEFAULT_UNITTESTS_DIVERSITY_PATH,
    DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
    UNITTEST_TOTAL_PATTERN,
    UNITTEST_FAILED_PATTERN,
    GLOBAL_PROJECTS,
    PYENV_EXISTS,
    PYTHON,
)
from tests4py.environment import (
    env_on,
    create_venv,
    activate_venv,
    update_env,
    sflkit_env,
    install_pyenv,
)
from tests4py.logger import LOGGER
from tests4py.projects import (
    resources,
    TestingFramework,
    get_number_of_bugs,
    get_project_names,
    TestStatus,
    Project,
    Status,
    get_project,
    get_matching_projects,
)
from tests4py.tests.utils import get_pytest_skip


def checkout(
    project: Project | str,
    work_dir: Path = DEFAULT_WORK_DIR,
    update: bool = False,
    force: bool = False,
    report: Optional[CheckoutReport] = None,
    verbose: bool = False,
) -> CheckoutReport:
    """
    Performs the checkout operation for the given project.

    Args:
        project (Project | str): The project to be checked out.
        work_dir (Path): The working directory for the checkout operation. Defaults to DEFAULT_WORK_DIR.
        update (bool): Indicates whether to update the project. Defaults to False.
        force (bool): Indicates whether to force the checkout. Defaults to False.
        report (Optional[CheckoutReport]): The checkout report. Defaults to None.
        verbose (bool): Indicates whether to display verbose output. Defaults to False.

    Returns:
        CheckoutReport: The report of the checkout operation.
    """
    if report is None:
        report = CheckoutReport()
    config = load_config()
    try:
        if isinstance(project, str):
            matching_projects = get_matching_projects(project)
            if matching_projects:
                project = matching_projects[0]
            else:
                raise ValueError(f"Project {project} not found")
        report.project = project

        check_further = project.status is Status.OK

        work_location = work_dir / project.get_identifier()
        report.location = work_location
        if update and work_location.exists():
            project_verify = load_project(work_location, only_project=True)
            project.compiled = project_verify.compiled
        else:
            tmp_location = (
                work_dir / f"tmp_{project.project_name}_{project.bug_id}"
            ).absolute()

            if check_further:
                if force or not config.cache or not check_cache_exists(project):
                    LOGGER.info(
                        f"Cloning {project.github_url} into {work_location}... "
                    )
                    if work_location.exists():
                        shutil.rmtree(work_location, ignore_errors=True)
                    subprocess.run(
                        ["git", "clone", project.github_url, work_location],
                        stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                    )
                    if config.cache:
                        cache(project, work_location)
                else:
                    LOGGER.info(
                        f"Copying {project.github_url} from {GLOBAL_PROJECTS / project.project_name} "
                        f"into {work_location}... "
                    )
                    copy_cached(project, work_location)
            else:
                raise AttributeError("Not status=OK")

            try:
                LOGGER.info(
                    f"Resetting git at {work_location} to {project.fixed_commit_id}"
                )
                subprocess.run(
                    ["git", "reset", "--hard", project.fixed_commit_id],
                    stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                    cwd=work_location,
                )

                LOGGER.info(f"Creating tmp location at {tmp_location}")
                os.makedirs(tmp_location, exist_ok=True)

                LOGGER.info(f"Copying required files to {tmp_location}")
                # Determine which files the fix touched. A normal fix commit is
                # described by `git show --name-only`. A MERGE fix commit lists no
                # files there (the diff is empty against its first parent), which
                # silently left the fix un-applied; detect that case and fall back
                # to the full buggy->fixed delta so merge-carried fixes apply.
                parents = (
                    subprocess.run(
                        ["git", "rev-list", "--parents", "-n", "1", "HEAD"],
                        stdout=subprocess.PIPE,
                        stderr=subprocess.DEVNULL,
                        cwd=work_location,
                    )
                    .stdout.decode("utf-8")
                    .split()
                )
                is_merge_fix = len(parents) > 2  # <commit> <parent1> <parent2>...
                if is_merge_fix:
                    LOGGER.info(
                        "Fixed commit is a merge; using buggy->fixed diff to "
                        "locate changed files."
                    )
                    name_only_cmd = [
                        "git",
                        "diff",
                        "--name-only",
                        project.buggy_commit_id,
                        project.fixed_commit_id,
                    ]
                else:
                    name_only_cmd = ["git", "show", "--name-only"]
                result = subprocess.run(
                    name_only_cmd,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    cwd=work_location,
                ).stdout.decode("utf-8")
                change_file_all = list()

                for line in result.split("\n"):
                    if line:
                        line_path = work_location / line
                        if line_path.exists() and line_path.is_file():
                            change_file_all.append(f"{line}")
                            if not project.buggy:
                                shutil.copy(
                                    line_path,
                                    tmp_location / line.replace(os.path.sep, "_"),
                                )

                for test_file in project.test_files:
                    if (work_location / test_file).is_dir():
                        shutil.copytree(
                            work_location / test_file,
                            tmp_location / str(test_file).replace(os.path.sep, "_"),
                            dirs_exist_ok=True,
                        )
                    else:
                        shutil.copy(
                            work_location / test_file,
                            tmp_location / str(test_file).replace(os.path.sep, "_"),
                        )

                LOGGER.info(f"Checkout buggy commit id {project.buggy_commit_id}")
                subprocess.run(
                    ["git", "reset", "--hard", project.buggy_commit_id],
                    stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                    cwd=work_location,
                )
                subprocess.run(
                    ["git", "clean", "-f", "-d"],
                    stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                    cwd=work_location,
                )

                LOGGER.info(f"Copying required files from {tmp_location}")

                for test_file in project.test_files:
                    dest = tmp_location / str(test_file).replace(os.path.sep, "_")
                    if dest.is_dir():
                        shutil.copytree(
                            dest, work_location / test_file, dirs_exist_ok=True
                        )
                    else:
                        shutil.copy(dest, work_location / test_file)

                patch_fix_all = list()
                # Copy other change file from fixed to buggy if version is fixed commit
                for change_file in change_file_all:
                    change_file_path = tmp_location / change_file.replace(
                        os.path.sep, "_"
                    )
                    if change_file_path.exists():
                        patch_fix_all.append(change_file)
                        if not project.buggy:
                            shutil.move(change_file_path, work_location / change_file)

                project.patch(work_location)
            finally:
                shutil.rmtree(tmp_location, ignore_errors=True)

            LOGGER.info(f"Create info file")
            with open(work_location / FIX_FILES, "w") as fp:
                fp.write(";".join(patch_fix_all))

        # Move information about bug to clone project folder
        project.write_bug_info(work_location / INFO_FILE)

        LOGGER.info(f"Copying resources for {project.get_identifier()}")
        project_resources = importlib.resources.files(
            getattr(getattr(resources, project.project_name), f"bug_{project.bug_id}")
        )
        with importlib.resources.as_file(
            project_resources.joinpath(
                "requirements.txt",
            )
        ) as resource:
            if resource.exists():
                shutil.copy(resource, work_location / REQUIREMENTS_FILE)
            else:
                with importlib.resources.as_file(
                    importlib.resources.files(
                        getattr(resources, project.project_name)
                    ).joinpath("requirements.txt")
                ) as default_resource:
                    if default_resource.exists():
                        shutil.copy(default_resource, work_location / REQUIREMENTS_FILE)
        with importlib.resources.as_file(
            project_resources.joinpath("harness.py"),
        ) as resource:
            if resource.exists():
                shutil.copy(resource, work_location / HARNESS_FILE)
            else:
                with importlib.resources.as_file(
                    importlib.resources.files(
                        getattr(resources, project.project_name)
                    ).joinpath("harness.py")
                ) as default_resource:
                    if default_resource.exists():
                        shutil.copy(default_resource, work_location / HARNESS_FILE)

        with importlib.resources.as_file(
            project_resources.joinpath(
                "fix.patch",
            )
        ) as resource:
            if resource.exists():
                shutil.copy(resource, work_location / PATCH_FILE)
        with importlib.resources.as_file(
            project_resources.joinpath(
                "explanation.md",
            )
        ) as resource:
            if resource.exists():
                shutil.copy(resource, work_location / EXPLANATION_FILE)

        if project.unittests:
            with importlib.resources.as_file(
                project_resources.joinpath(
                    "unittests.py",
                )
            ) as resource:
                if resource.exists():
                    shutil.copy(
                        resource, work_location / DEFAULT_UNITTESTS_DIVERSITY_PATH
                    )

        if project.systemtests:
            module = getattr(
                getattr(
                    getattr(resources, project.project_name), f"bug_{project.bug_id}"
                ),
                "systemtests",
            )
            module.TestsPassing().write(
                work_location / DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
                grammar=project.grammar,
            )
            module.TestsFailing().write(
                work_location / DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
                grammar=project.grammar,
            )
        config.last_workdir = work_location.absolute()
        config.write()
        report.successful = True
    except BaseException as e:
        report.raised = e
        report.successful = False
    return report


def build(
    work_dir_or_project: Optional[Union[os.PathLike, Project]] = None,
    rebuild: bool = False,
    force: bool = False,
    report: Optional[CompileReport] = None,
    sfl: bool = False,
    verbose: bool = False,
) -> CompileReport:
    if not PYENV_EXISTS:
        install_pyenv()
    report = report or CompileReport()
    config = load_config()
    work_dir = get_work_dir(work_dir_or_project)
    report.location = work_dir
    try:
        project, t4p_info, t4p_requirements = load_project(work_dir)
        report.project = project
        if project.compiled and not rebuild:
            LOGGER.info(f"{project} already compiled")
            report.successful = True
            return report

        if project.project_name == "pandas":
            if shutil.which("gcc-10") is None:
                raise ValueError("GCC 10 is required to build pandas")
            try:
                subprocess.run(
                    ["gcc-10", "--version"],
                    stdout=subprocess.PIPE,
                    check=True,
                )
            except (subprocess.CalledProcessError, FileNotFoundError) as exc:
                raise ValueError("GCC 10 is required to build pandas") from exc

        environ = env_on(project)
        if force or not config.cache or not check_cache_exists_env(project):
            create_venv(work_dir, environ)
            env_exists = False
        else:
            copy_cached_env(project, work_dir)
            env_exists = True

        environ = activate_venv(work_dir, environ)
        report.env = environ

        if not env_exists:
            LOGGER.info("Installing utilities")
            update_env(environ, force=force)

            LOGGER.info("Installing requirements")
            subprocess.check_call(
                [PYTHON, "-m", "pip", "install", "-r", t4p_requirements],
                stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                env=environ,
            )

            LOGGER.info("Checking and installing test requirements")
            test_requirements = work_dir / "test_requirements.txt"
            if test_requirements.exists():
                subprocess.check_call(
                    [PYTHON, "-m", "pip", "install", "-r", test_requirements],
                    stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                    env=environ,
                )

        if config.cache:
            cache_venv(project, work_dir)

        if sfl:
            sflkit_env(environ)

        LOGGER.info("Run setup")
        if project.setup_env:
            environ.update(project.setup_env)

        for command in project.setup:
            subprocess.check_call(
                command,
                env=environ,
                stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
                cwd=work_dir,
            )

        LOGGER.info("Set compiled flag")
        project.compiled = True
        project.write_bug_info(t4p_info)

        report.successful = True
    except BaseException as e:
        report.raised = e
        report.successful = False
    return report


def info(
    project_name: Optional[str] = None,
    bug_id: Optional[int] = None,
    report: Optional[InfoReport] = None,
):
    report = report or InfoReport()
    try:
        if not project_name:
            print("The existing subjects in Tests4Py:")
            print()
            print(
                tabulate(
                    map(lambda p: [p, get_number_of_bugs(p)], get_project_names()),
                    headers=["project", "# bugs"],
                    tablefmt="fancy_grid",
                )
            )
            print()
            print("Set a project to get further information.")
            report.successful = True
            return report
        project_name = project_name.lower()
        if bug_id is None:
            project = get_project(project_name, 1)
            report.example = True
            report.project = project
            description = project.project_name
        else:
            project = get_project(project_name, bug_id)
            report.project = project
            description = f"{project.project_name} with bug id {project.bug_id}"

        data = [
            ("Name", project.project_name),
            ("URL", project.github_url),
            ("Status", project.status),
        ]
        if bug_id is None:
            data.append(("Bugs", get_number_of_bugs(project.project_name)))
        else:
            data += [
                ("ID", project.bug_id),
                ("Python Version", project.python_version),
                ("Python Path", project.python_path),
                ("Buggy Commit", project.buggy_commit_id),
                ("Fixed Commit", project.fixed_commit_id),
                ("Test Files", "\n".join(map(str, project.test_files))),
                ("Test Cases", "\n".join(project.test_cases)),
                ("Unit Tests", project.unittests is not None),
                ("System Tests", project.systemtests is not None),
            ]

        print(f"Information for project {description}:")
        print()
        print(
            tabulate(
                data,
                tablefmt="fancy_grid",
            )
        )
        print()

        if bug_id is not None:
            if project.test_status_buggy == TestStatus.PASSING:
                print("WARNING: The tests do not fail on the buggy commit.")
                print("Hence, this subject is not useful.")
                print()
            if project.test_status_fixed == TestStatus.FAILING:
                print("WARNING: The tests do fail on the fixed commit.")
                print(
                    "The subject could still be used and generated tests and the oracle reveal to correct fault."
                )
                print()

        report.successful = True
    except BaseException as e:
        report.raised = e
        report.successful = False
    return report


def test(
    work_dir_or_project: Optional[Union[os.PathLike, Project]] = None,
    single_test: Optional[Union[List[str], str]] = None,
    relevant_tests: bool = False,
    all_tests: bool = False,
    xml_output: Optional[Path] = None,
    coverage: bool = False,
    report: Optional[TestReport] = None,
) -> TestReport:
    if report is None:
        report = TestReport()
    work_dir = get_work_dir(work_dir_or_project)
    report.location = work_dir
    try:
        project = load_project(work_dir, only_project=True)
        report.project = project
        if not project.compiled:
            raise ValueError(
                f"Project {project.project_name} at {work_dir} was not compiled"
            )

        if project.buggy and project.test_status_buggy == TestStatus.PASSING:
            LOGGER.warning(
                f"The tests will pass on this buggy version {project.get_identifier()}"
            )
        elif not project.buggy and project.test_status_fixed == TestStatus.FAILING:
            LOGGER.warning(
                f"The tests will fail on this fixed version {project.get_identifier()}"
            )

        environ = env_on(project)
        environ = activate_venv(work_dir, environ)

        command = [PYTHON, "-m"]

        if coverage:
            subprocess.run(
                [PYTHON, "-m", "pip", "install", "coverage"], env=environ, cwd=work_dir
            )
            command += ["coverage", "run", "-m"]

        if project.testing_framework == TestingFramework.PYTEST:
            command.append(TestingFramework.PYTEST.value)
            if project.set_rootdir:
                command.append(f"--rootdir={work_dir}")
            if xml_output:
                command.append(f"--junit-xml={xml_output.absolute()}")
        elif project.testing_framework == TestingFramework.UNITTEST:
            if xml_output:
                subprocess.run(
                    [PYTHON, "-m", "pip", "install", "unittest-xml-reporting"],
                    env=environ,
                    cwd=work_dir,
                )
                command += ["xmlrunner", "--output-file", xml_output.absolute()]
            else:
                command.append(TestingFramework.UNITTEST.value)
        else:
            raise NotImplementedError(
                f"No command found for {project.testing_framework.value}"
            )
        if project.test_command_arguments:
            command += project.test_command_arguments
        skips = []
        tests = []
        if not relevant_tests and not all_tests and not single_test:
            LOGGER.info(f"Run relevant tests {project.test_cases}")
            tests += project.test_cases
        elif all_tests:
            if project.test_base:
                tests.append(project.test_base)
        elif relevant_tests:
            tests += project.relevant_test_files
            if project.skip_tests:
                skips = [
                    "-k",
                    get_pytest_skip(project.skip_tests),
                ]
        elif single_test:
            if isinstance(single_test, str):
                tests.append(single_test)
            else:
                tests += single_test

        command += skips
        command += tests

        LOGGER.info(f"Run tests with command {command}")
        process = subprocess.run(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=environ,
            cwd=work_dir,
        )
        output = process.stdout or process.stderr
        LOGGER.info(output.decode("utf-8"))

        successful = False
        if project.testing_framework == TestingFramework.PYTEST:
            (
                successful,
                report.total,
                report.failing,
                report.passing,
            ) = get_pytest_result(output)
        elif project.testing_framework == TestingFramework.UNITTEST:
            number_match = UNITTEST_TOTAL_PATTERN.search(output)
            failed_match = UNITTEST_FAILED_PATTERN.search(output)
            if number_match:
                report.total = int(number_match.group("n"))
                if failed_match:
                    failed = failed_match.group("f")
                    errors = failed_match.group("e")
                    report.failing = 0
                    if failed:
                        report.failing += int(failed)
                    if errors:
                        report.failing += int(errors)
                else:
                    report.failing = 0
                report.passing = report.total - report.failing
                successful = True
        else:
            raise NotImplementedError(
                f"No command found for {project.testing_framework.value}"
            )

        if xml_output:
            report.results = get_test_results(project, work_dir, xml_output)
        LOGGER.info(f"Ran {report.total} tests")
        LOGGER.info(f"{report.passing} passed --- {report.failing} failed")
        report.successful = successful
    except BaseException as e:
        report.raised = e
        report.successful = False
    return report

1	import importlib.resources	1✔
2	import os	1✔
3	import shutil	1✔
4	import subprocess	1✔
5	from pathlib import Path	1✔
6	from typing import List, Union, Optional	1✔
7
8	from tabulate import tabulate	1✔
9
10	from tests4py.api.cache import (	1✔
11	check_cache_exists,
12	copy_cached,
13	cache,
14	check_cache_exists_env,
15	copy_cached_env,
16	cache_venv,
17	)
18	from tests4py.api.config import load_config	1✔
19	from tests4py.api.report import CheckoutReport, CompileReport, InfoReport, TestReport	1✔
20	from tests4py.api.test import get_pytest_result, get_test_results	1✔
21	from tests4py.api.utils import get_work_dir, load_project	1✔
22	from tests4py.constants import (	1✔
23	DEFAULT_WORK_DIR,
24	FIX_FILES,
25	INFO_FILE,
26	REQUIREMENTS_FILE,
27	HARNESS_FILE,
28	PATCH_FILE,
29	EXPLANATION_FILE,
30	DEFAULT_UNITTESTS_DIVERSITY_PATH,
31	DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
32	UNITTEST_TOTAL_PATTERN,
33	UNITTEST_FAILED_PATTERN,
34	GLOBAL_PROJECTS,
35	PYENV_EXISTS,
36	PYTHON,
37	)
38	from tests4py.environment import (	1✔
39	env_on,
40	create_venv,
41	activate_venv,
42	update_env,
43	sflkit_env,
44	install_pyenv,
45	)
46	from tests4py.logger import LOGGER	1✔
47	from tests4py.projects import (	1✔
48	resources,
49	TestingFramework,
50	get_number_of_bugs,
51	get_project_names,
52	TestStatus,
53	Project,
54	Status,
55	get_project,
56	get_matching_projects,
57	)
58	from tests4py.tests.utils import get_pytest_skip	1✔
59
60
61	def checkout(	1✔
62	project: Project \| str,
63	work_dir: Path = DEFAULT_WORK_DIR,
64	update: bool = False,
65	force: bool = False,
66	report: Optional[CheckoutReport] = None,
67	verbose: bool = False,
68	) -> CheckoutReport:
69	"""
70	Performs the checkout operation for the given project.
71
72	Args:
73	project (Project \| str): The project to be checked out.
74	work_dir (Path): The working directory for the checkout operation. Defaults to DEFAULT_WORK_DIR.
75	update (bool): Indicates whether to update the project. Defaults to False.
76	force (bool): Indicates whether to force the checkout. Defaults to False.
77	report (Optional[CheckoutReport]): The checkout report. Defaults to None.
78	verbose (bool): Indicates whether to display verbose output. Defaults to False.
79
80	Returns:
81	CheckoutReport: The report of the checkout operation.
82	"""
83	if report is None:	1✔
84	report = CheckoutReport()	1✔
85	config = load_config()	1✔
86	try:	1✔
87	if isinstance(project, str):	1✔
88	matching_projects = get_matching_projects(project)	×
89	if matching_projects:	×
90	project = matching_projects[0]	×
91	else:
92	raise ValueError(f"Project {project} not found")	×
93	report.project = project	1✔
94
95	check_further = project.status is Status.OK	1✔
96
97	work_location = work_dir / project.get_identifier()	1✔
98	report.location = work_location	1✔
99	if update and work_location.exists():	1✔
100	project_verify = load_project(work_location, only_project=True)	×
101	project.compiled = project_verify.compiled	×
102	else:
103	tmp_location = (	1✔
104	work_dir / f"tmp_{project.project_name}_{project.bug_id}"
105	).absolute()
106
107	if check_further:	1✔
108	if force or not config.cache or not check_cache_exists(project):	1✔
109	LOGGER.info(	1✔
110	f"Cloning {project.github_url} into {work_location}... "
111	)
112	if work_location.exists():	1✔
113	shutil.rmtree(work_location, ignore_errors=True)	1✔
114	subprocess.run(	1✔
115	["git", "clone", project.github_url, work_location],
116	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
117	)
118	if config.cache:	1✔
119	cache(project, work_location)	1✔
120	else:
121	LOGGER.info(	1✔
122	f"Copying {project.github_url} from {GLOBAL_PROJECTS / project.project_name} "
123	f"into {work_location}... "
124	)
125	copy_cached(project, work_location)	1✔
126	else:
127	raise AttributeError("Not status=OK")	×
128
129	try:	1✔
130	LOGGER.info(	1✔
131	f"Resetting git at {work_location} to {project.fixed_commit_id}"
132	)
133	subprocess.run(	1✔
134	["git", "reset", "--hard", project.fixed_commit_id],
135	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
136	cwd=work_location,
137	)
138
139	LOGGER.info(f"Creating tmp location at {tmp_location}")	1✔
140	os.makedirs(tmp_location, exist_ok=True)	1✔
141
142	LOGGER.info(f"Copying required files to {tmp_location}")	1✔
143	# Determine which files the fix touched. A normal fix commit is
144	# described by `git show --name-only`. A MERGE fix commit lists no
145	# files there (the diff is empty against its first parent), which
146	# silently left the fix un-applied; detect that case and fall back
147	# to the full buggy->fixed delta so merge-carried fixes apply.
148	parents = (	1✔
149	subprocess.run(
150	["git", "rev-list", "--parents", "-n", "1", "HEAD"],
151	stdout=subprocess.PIPE,
152	stderr=subprocess.DEVNULL,
153	cwd=work_location,
154	)
155	.stdout.decode("utf-8")
156	.split()
157	)
158	is_merge_fix = len(parents) > 2 # <commit> <parent1> <parent2>...	1✔
159	if is_merge_fix:	1✔
NEW 160	LOGGER.info(	×
161	"Fixed commit is a merge; using buggy->fixed diff to "
162	"locate changed files."
163	)
NEW 164	name_only_cmd = [	×
165	"git",
166	"diff",
167	"--name-only",
168	project.buggy_commit_id,
169	project.fixed_commit_id,
170	]
171	else:
172	name_only_cmd = ["git", "show", "--name-only"]	1✔
173	result = subprocess.run(	1✔
174	name_only_cmd,
175	stdout=subprocess.PIPE,
176	stderr=subprocess.STDOUT,
177	cwd=work_location,
178	).stdout.decode("utf-8")
179	change_file_all = list()	1✔
180
181	for line in result.split("\n"):	1✔
182	if line:	1✔
183	line_path = work_location / line	1✔
184	if line_path.exists() and line_path.is_file():	1✔
185	change_file_all.append(f"{line}")	1✔
186	if not project.buggy:	1✔
187	shutil.copy(	1✔
188	line_path,
189	tmp_location / line.replace(os.path.sep, "_"),
190	)
191
192	for test_file in project.test_files:	1✔
193	if (work_location / test_file).is_dir():	1✔
194	shutil.copytree(	1✔
195	work_location / test_file,
196	tmp_location / str(test_file).replace(os.path.sep, "_"),
197	dirs_exist_ok=True,
198	)
199	else:
200	shutil.copy(	1✔
201	work_location / test_file,
202	tmp_location / str(test_file).replace(os.path.sep, "_"),
203	)
204
205	LOGGER.info(f"Checkout buggy commit id {project.buggy_commit_id}")	1✔
206	subprocess.run(	1✔
207	["git", "reset", "--hard", project.buggy_commit_id],
208	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
209	cwd=work_location,
210	)
211	subprocess.run(	1✔
212	["git", "clean", "-f", "-d"],
213	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
214	cwd=work_location,
215	)
216
217	LOGGER.info(f"Copying required files from {tmp_location}")	1✔
218
219	for test_file in project.test_files:	1✔
220	dest = tmp_location / str(test_file).replace(os.path.sep, "_")	1✔
221	if dest.is_dir():	1✔
222	shutil.copytree(	1✔
223	dest, work_location / test_file, dirs_exist_ok=True
224	)
225	else:
226	shutil.copy(dest, work_location / test_file)	1✔
227
228	patch_fix_all = list()	1✔
229	# Copy other change file from fixed to buggy if version is fixed commit
230	for change_file in change_file_all:	1✔
231	change_file_path = tmp_location / change_file.replace(	1✔
232	os.path.sep, "_"
233	)
234	if change_file_path.exists():	1✔
235	patch_fix_all.append(change_file)	1✔
236	if not project.buggy:	1✔
237	shutil.move(change_file_path, work_location / change_file)	1✔
238
239	project.patch(work_location)	1✔
240	finally:
241	shutil.rmtree(tmp_location, ignore_errors=True)	1✔
242
243	LOGGER.info(f"Create info file")	1✔
244	with open(work_location / FIX_FILES, "w") as fp:	1✔
245	fp.write(";".join(patch_fix_all))	1✔
246
247	# Move information about bug to clone project folder
248	project.write_bug_info(work_location / INFO_FILE)	1✔
249
250	LOGGER.info(f"Copying resources for {project.get_identifier()}")	1✔
251	project_resources = importlib.resources.files(	1✔
252	getattr(getattr(resources, project.project_name), f"bug_{project.bug_id}")
253	)
254	with importlib.resources.as_file(	1✔
255	project_resources.joinpath(
256	"requirements.txt",
257	)
258	) as resource:
259	if resource.exists():	1✔
260	shutil.copy(resource, work_location / REQUIREMENTS_FILE)	1✔
261	else:
262	with importlib.resources.as_file(	1✔
263	importlib.resources.files(
264	getattr(resources, project.project_name)
265	).joinpath("requirements.txt")
266	) as default_resource:
267	if default_resource.exists():	1✔
268	shutil.copy(default_resource, work_location / REQUIREMENTS_FILE)	1✔
269	with importlib.resources.as_file(	1✔
270	project_resources.joinpath("harness.py"),
271	) as resource:
272	if resource.exists():	1✔
273	shutil.copy(resource, work_location / HARNESS_FILE)	1✔
274	else:
275	with importlib.resources.as_file(	1✔
276	importlib.resources.files(
277	getattr(resources, project.project_name)
278	).joinpath("harness.py")
279	) as default_resource:
280	if default_resource.exists():	1✔
281	shutil.copy(default_resource, work_location / HARNESS_FILE)	1✔
282
283	with importlib.resources.as_file(	1✔
284	project_resources.joinpath(
285	"fix.patch",
286	)
287	) as resource:
288	if resource.exists():	1✔
289	shutil.copy(resource, work_location / PATCH_FILE)	1✔
290	with importlib.resources.as_file(	1✔
291	project_resources.joinpath(
292	"explanation.md",
293	)
294	) as resource:
295	if resource.exists():	1✔
296	shutil.copy(resource, work_location / EXPLANATION_FILE)	1✔
297
298	if project.unittests:	1✔
299	with importlib.resources.as_file(	1✔
300	project_resources.joinpath(
301	"unittests.py",
302	)
303	) as resource:
304	if resource.exists():	1✔
305	shutil.copy(	1✔
306	resource, work_location / DEFAULT_UNITTESTS_DIVERSITY_PATH
307	)
308
309	if project.systemtests:	1✔
310	module = getattr(	1✔
311	getattr(
312	getattr(resources, project.project_name), f"bug_{project.bug_id}"
313	),
314	"systemtests",
315	)
316	module.TestsPassing().write(	1✔
317	work_location / DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
318	grammar=project.grammar,
319	)
320	module.TestsFailing().write(	1✔
321	work_location / DEFAULT_SYSTEMTESTS_DIVERSITY_PATH,
322	grammar=project.grammar,
323	)
324	config.last_workdir = work_location.absolute()	1✔
325	config.write()	1✔
326	report.successful = True	1✔
327	except BaseException as e:	×
328	report.raised = e	×
329	report.successful = False	×
330	return report	1✔
331
332
333	def build(	1✔
334	work_dir_or_project: Optional[Union[os.PathLike, Project]] = None,
335	rebuild: bool = False,
336	force: bool = False,
337	report: Optional[CompileReport] = None,
338	sfl: bool = False,
339	verbose: bool = False,
340	) -> CompileReport:
341	if not PYENV_EXISTS:	1✔
342	install_pyenv()	×
343	report = report or CompileReport()	1✔
344	config = load_config()	1✔
345	work_dir = get_work_dir(work_dir_or_project)	1✔
346	report.location = work_dir	1✔
347	try:	1✔
348	project, t4p_info, t4p_requirements = load_project(work_dir)	1✔
349	report.project = project	1✔
350	if project.compiled and not rebuild:	1✔
351	LOGGER.info(f"{project} already compiled")	×
352	report.successful = True	×
353	return report	×
354
355	if project.project_name == "pandas":	1✔
356	if shutil.which("gcc-10") is None:	×
357	raise ValueError("GCC 10 is required to build pandas")	×
358	try:	×
359	subprocess.run(	×
360	["gcc-10", "--version"],
361	stdout=subprocess.PIPE,
362	check=True,
363	)
364	except (subprocess.CalledProcessError, FileNotFoundError) as exc:	×
365	raise ValueError("GCC 10 is required to build pandas") from exc	×
366
367	environ = env_on(project)	1✔
368	if force or not config.cache or not check_cache_exists_env(project):	1✔
369	create_venv(work_dir, environ)	1✔
370	env_exists = False	1✔
371	else:
372	copy_cached_env(project, work_dir)	1✔
373	env_exists = True	1✔
374
375	environ = activate_venv(work_dir, environ)	1✔
376	report.env = environ	1✔
377
378	if not env_exists:	1✔
379	LOGGER.info("Installing utilities")	1✔
380	update_env(environ, force=force)	1✔
381
382	LOGGER.info("Installing requirements")	1✔
383	subprocess.check_call(	1✔
384	[PYTHON, "-m", "pip", "install", "-r", t4p_requirements],
385	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
386	env=environ,
387	)
388
389	LOGGER.info("Checking and installing test requirements")	1✔
390	test_requirements = work_dir / "test_requirements.txt"	1✔
391	if test_requirements.exists():	1✔
392	subprocess.check_call(	1✔
393	[PYTHON, "-m", "pip", "install", "-r", test_requirements],
394	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
395	env=environ,
396	)
397
398	if config.cache:	1✔
399	cache_venv(project, work_dir)	1✔
400
401	if sfl:	1✔
402	sflkit_env(environ)	1✔
403
404	LOGGER.info("Run setup")	1✔
405	if project.setup_env:	1✔
406	environ.update(project.setup_env)	×
407
408	for command in project.setup:	1✔
409	subprocess.check_call(	1✔
410	command,
411	env=environ,
412	stdout=subprocess.STDOUT if verbose else subprocess.DEVNULL,
413	cwd=work_dir,
414	)
415
416	LOGGER.info("Set compiled flag")	1✔
417	project.compiled = True	1✔
418	project.write_bug_info(t4p_info)	1✔
419
420	report.successful = True	1✔
421	except BaseException as e:	×
422	report.raised = e	×
423	report.successful = False	×
424	return report	1✔
425
426
427	def info(	1✔
428	project_name: Optional[str] = None,
429	bug_id: Optional[int] = None,
430	report: Optional[InfoReport] = None,
431	):
432	report = report or InfoReport()	×
433	try:	×
434	if not project_name:	×
435	print("The existing subjects in Tests4Py:")	×
436	print()	×
437	print(	×
438	tabulate(
439	map(lambda p: [p, get_number_of_bugs(p)], get_project_names()),
440	headers=["project", "# bugs"],
441	tablefmt="fancy_grid",
442	)
443	)
444	print()	×
445	print("Set a project to get further information.")	×
446	report.successful = True	×
447	return report	×
448	project_name = project_name.lower()	×
449	if bug_id is None:	×
450	project = get_project(project_name, 1)	×
451	report.example = True	×
452	report.project = project	×
453	description = project.project_name	×
454	else:
455	project = get_project(project_name, bug_id)	×
456	report.project = project	×
457	description = f"{project.project_name} with bug id {project.bug_id}"	×
458
459	data = [	×
460	("Name", project.project_name),
461	("URL", project.github_url),
462	("Status", project.status),
463	]
464	if bug_id is None:	×
465	data.append(("Bugs", get_number_of_bugs(project.project_name)))	×
466	else:
467	data += [	×
468	("ID", project.bug_id),
469	("Python Version", project.python_version),
470	("Python Path", project.python_path),
471	("Buggy Commit", project.buggy_commit_id),
472	("Fixed Commit", project.fixed_commit_id),
473	("Test Files", "\n".join(map(str, project.test_files))),
474	("Test Cases", "\n".join(project.test_cases)),
475	("Unit Tests", project.unittests is not None),
476	("System Tests", project.systemtests is not None),
477	]
478
479	print(f"Information for project {description}:")	×
480	print()	×
481	print(	×
482	tabulate(
483	data,
484	tablefmt="fancy_grid",
485	)
486	)
487	print()	×
488
489	if bug_id is not None:	×
490	if project.test_status_buggy == TestStatus.PASSING:	×
491	print("WARNING: The tests do not fail on the buggy commit.")	×
492	print("Hence, this subject is not useful.")	×
493	print()	×
494	if project.test_status_fixed == TestStatus.FAILING:	×
495	print("WARNING: The tests do fail on the fixed commit.")	×
496	print(	×
497	"The subject could still be used and generated tests and the oracle reveal to correct fault."
498	)
499	print()	×
500
501	report.successful = True	×
502	except BaseException as e:	×
503	report.raised = e	×
504	report.successful = False	×
505	return report	×
506
507
508	def test(	1✔
509	work_dir_or_project: Optional[Union[os.PathLike, Project]] = None,
510	single_test: Optional[Union[List[str], str]] = None,
511	relevant_tests: bool = False,
512	all_tests: bool = False,
513	xml_output: Optional[Path] = None,
514	coverage: bool = False,
515	report: Optional[TestReport] = None,
516	) -> TestReport:
517	if report is None:	1✔
518	report = TestReport()	×
519	work_dir = get_work_dir(work_dir_or_project)	1✔
520	report.location = work_dir	1✔
521	try:	1✔
522	project = load_project(work_dir, only_project=True)	1✔
523	report.project = project	1✔
524	if not project.compiled:	1✔
525	raise ValueError(	×
526	f"Project {project.project_name} at {work_dir} was not compiled"
527	)
528
529	if project.buggy and project.test_status_buggy == TestStatus.PASSING:	1✔
530	LOGGER.warning(	×
531	f"The tests will pass on this buggy version {project.get_identifier()}"
532	)
533	elif not project.buggy and project.test_status_fixed == TestStatus.FAILING:	1✔
534	LOGGER.warning(	×
535	f"The tests will fail on this fixed version {project.get_identifier()}"
536	)
537
538	environ = env_on(project)	1✔
539	environ = activate_venv(work_dir, environ)	1✔
540
541	command = [PYTHON, "-m"]	1✔
542
543	if coverage:	1✔
544	subprocess.run(	×
545	[PYTHON, "-m", "pip", "install", "coverage"], env=environ, cwd=work_dir
546	)
547	command += ["coverage", "run", "-m"]	×
548
549	if project.testing_framework == TestingFramework.PYTEST:	1✔
550	command.append(TestingFramework.PYTEST.value)	1✔
551	if project.set_rootdir:	1✔
552	command.append(f"--rootdir={work_dir}")	1✔
553	if xml_output:	1✔
554	command.append(f"--junit-xml={xml_output.absolute()}")	×
555	elif project.testing_framework == TestingFramework.UNITTEST:	×
556	if xml_output:	×
557	subprocess.run(	×
558	[PYTHON, "-m", "pip", "install", "unittest-xml-reporting"],
559	env=environ,
560	cwd=work_dir,
561	)
562	command += ["xmlrunner", "--output-file", xml_output.absolute()]	×
563	else:
564	command.append(TestingFramework.UNITTEST.value)	×
565	else:
566	raise NotImplementedError(
567	f"No command found for {project.testing_framework.value}"
568	)
569	if project.test_command_arguments:	1✔
570	command += project.test_command_arguments	×
571	skips = []	1✔
572	tests = []	1✔
573	if not relevant_tests and not all_tests and not single_test:	1✔
574	LOGGER.info(f"Run relevant tests {project.test_cases}")	1✔
575	tests += project.test_cases	1✔
576	elif all_tests:	×
577	if project.test_base:	×
578	tests.append(project.test_base)	×
579	elif relevant_tests:	×
580	tests += project.relevant_test_files	×
581	if project.skip_tests:	×
582	skips = [	×
583	"-k",
584	get_pytest_skip(project.skip_tests),
585	]
586	elif single_test:	×
587	if isinstance(single_test, str):	×
588	tests.append(single_test)	×
589	else:
590	tests += single_test	×
591
592	command += skips	1✔
593	command += tests	1✔
594
595	LOGGER.info(f"Run tests with command {command}")	1✔
596	process = subprocess.run(	1✔
597	command,
598	stdout=subprocess.PIPE,
599	stderr=subprocess.PIPE,
600	env=environ,
601	cwd=work_dir,
602	)
603	output = process.stdout or process.stderr	1✔
604	LOGGER.info(output.decode("utf-8"))	1✔
605
606	successful = False	1✔
607	if project.testing_framework == TestingFramework.PYTEST:	1✔
608	(	1✔
609	successful,
610	report.total,
611	report.failing,
612	report.passing,
613	) = get_pytest_result(output)
614	elif project.testing_framework == TestingFramework.UNITTEST:	×
615	number_match = UNITTEST_TOTAL_PATTERN.search(output)	×
616	failed_match = UNITTEST_FAILED_PATTERN.search(output)	×
617	if number_match:	×
618	report.total = int(number_match.group("n"))	×
619	if failed_match:	×
620	failed = failed_match.group("f")	×
621	errors = failed_match.group("e")	×
622	report.failing = 0	×
623	if failed:	×
624	report.failing += int(failed)	×
625	if errors:	×
626	report.failing += int(errors)	×
627	else:
628	report.failing = 0	×
629	report.passing = report.total - report.failing	×
630	successful = True	×
631	else:
632	raise NotImplementedError(
633	f"No command found for {project.testing_framework.value}"
634	)
635
636	if xml_output:	1✔
637	report.results = get_test_results(project, work_dir, xml_output)	×
638	LOGGER.info(f"Ran {report.total} tests")	1✔
639	LOGGER.info(f"{report.passing} passed --- {report.failing} failed")	1✔
640	report.successful = successful	1✔
641	except BaseException as e:	×
642	report.raised = e	×
643	report.successful = False	×
644	return report	1✔

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