# This file is part of Pynguin.
#
# SPDX-FileCopyrightText: 2019–2026 Pynguin Contributors
#
# SPDX-License-Identifier: MIT
#
"""Pynguin is an automated unit test generation framework for Python.
The framework generates unit tests for a given Python module. For this it
supports various approaches, such as a random approach, similar to Randoop or a
whole-suite approach, based on a genetic algorithm, as implemented in EvoSuite. The
framework allows to export test suites in various styles, i.e., using the `unittest`
library from the Python standard library or tests in the style used by the PyTest
framework.
Pynguin is supposed to be used as a standalone command-line application, but it
can also be used as a library by instantiating this class directly.
"""
from __future__ import annotations
import datetime
import enum
import importlib
import importlib.machinery
import json
import logging
import math
import random
import statistics
import sys
import time
import types
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
try:
import random
from faker import Faker
FANDANGO_FAKER_AVAILABLE = True
except ImportError:
FANDANGO_FAKER_AVAILABLE = False
import weakref
import multiprocess as mp
import pynguin.assertion.assertiongenerator as ag
import pynguin.assertion.llmassertiongenerator as lag
import pynguin.configuration as config
import pynguin.ga.chromosome as chrom
import pynguin.ga.chromosomevisitor as cv
import pynguin.ga.computations as ff
import pynguin.ga.generationalgorithmfactory as gaf
import pynguin.ga.postprocess as pp
import pynguin.ga.testsuitechromosome as tsc
import pynguin.utils.statistics.stats as stat
if config.configuration.pynguinml.ml_testing_enabled or TYPE_CHECKING:
import pynguin.utils.pynguinml.ml_testing_resources as tr
from pynguin.analyses.constants import (
ConstantProvider,
DelegatingConstantProvider,
DynamicConstantProvider,
EmptyConstantProvider,
RestrictedConstantPool,
collect_static_constants,
)
from pynguin.analyses.module import generate_test_cluster
from pynguin.assertion.mutation_analysis.controller import create_mutation_controller
from pynguin.instrumentation.machinery import InstrumentationFinder, install_import_hook
from pynguin.instrumentation.tracer import SubjectProperties
from pynguin.islands.population_allocation import minimum_island_population_size
from pynguin.large_language_model.client import OpenAIClient, extract_python_code
from pynguin.testcase import export
from pynguin.testcase.execution import (
RemoteAssertionExecutionObserver,
SubprocessTestCaseExecutor,
TestCaseExecutor,
)
from pynguin.utils import randomness
from pynguin.utils.exceptions import (
CannotInstrumentCompiledModuleError,
ConfigurationException,
)
from pynguin.utils.report import (
get_coverage_report,
render_coverage_report,
render_xml_coverage_report,
)
from pynguin.utils.statistics.runtimevariable import RuntimeVariable
from pynguin.utils.timeout import TestExecutionTimeoutError, time_limit
if TYPE_CHECKING:
from pynguin.analyses.module import ModuleTestCluster
from pynguin.ga.algorithms.generationalgorithm import GenerationAlgorithm
if config.configuration.pynguinml.ml_testing_enabled or TYPE_CHECKING:
from pynguin.utils.pynguinml import np_rng
[docs]
@enum.unique
class ReturnCode(enum.IntEnum):
"""Return codes for Pynguin to signal result."""
OK = 0
"""Symbolises that the execution ended as expected."""
SETUP_FAILED = 1
"""Symbolises that the execution failed in the setup phase."""
NO_TESTS_GENERATED = 2
"""Symbolises that no test could be generated."""
FINAL_METRICS_TRACKING_FAILED = 3
"""Symbolises that the final metrics tracking failed."""
_LOGGER = logging.getLogger(__name__)
[docs]
def set_configuration(configuration: config.Configuration) -> None:
"""Initialises the test generator with the given configuration.
Args:
configuration: The configuration to use.
"""
config.configuration = configuration
[docs]
def run_pynguin() -> ReturnCode:
"""Run the test generation.
The result of the test generation is indicated by the resulting ReturnCode.
Returns:
See ReturnCode.
Raises:
ConfigurationException: In case the configuration is illegal
"""
try:
_LOGGER.info("Start Pynguin Test Generation…")
if config.configuration.algorithm == config.Algorithm.LLM:
return _run_llm()
return _run()
finally:
_LOGGER.info("Stop Pynguin Test Generation…")
def _setup_test_cluster() -> ModuleTestCluster | None:
try:
test_cluster = generate_test_cluster(
config.configuration.module_name,
allow_c_module=False,
)
except CannotInstrumentCompiledModuleError as ex:
_LOGGER.error("%s", ex)
return None
except ModuleNotFoundError as ex:
_LOGGER.exception(
"""Module %s could not be found. This is likely due to a missing dependency.
It may also be caused by a bug in the SUT, especially if it uses C-modules.
""",
ex.name,
)
return None
except TestExecutionTimeoutError as ex:
_LOGGER.exception("Timeout setting up test cluster: %s", ex)
return None
if test_cluster.num_accessible_objects_under_test() == 0:
_LOGGER.error("SUT contains nothing we can test.")
return None
return test_cluster
def _setup_path() -> bool:
"""Set up the run-time path.
Inserts the path to the SUT into the path list, installs the import hook and
tries to load the SUT.
Returns:
An optional execution tracer, if loading was successful, None otherwise.
"""
if not Path(config.configuration.project_path).is_dir():
_LOGGER.error("%s is not a valid project path", config.configuration.project_path)
return False
_LOGGER.debug("Setting up path for %s", config.configuration.project_path)
sys.path.insert(0, config.configuration.project_path)
return True
def _is_within(file: str | None, directory: Path) -> bool:
if file is None:
return False
try:
Path(file).resolve().relative_to(directory)
except ValueError:
return False
return True
def _evict_shadowing_modules(module_name: str, project_path: str) -> dict[str, types.ModuleType]:
"""Remove an already-imported copy of the SUT's package that lives elsewhere.
Pynguin's own dependencies can import a module with the same top-level name as
the SUT from outside the project path (e.g., setuptools' vendored
``more_itertools``). Without eviction, the SUT import would reuse that copy
instead of the one in the project path.
Args:
module_name: The name of the module under test.
project_path: The project path containing the module under test.
Returns:
The evicted modules by name, so they can be restored if the project's copy
cannot be imported.
"""
top_level = module_name.partition(".")[0]
loaded = sys.modules.get(top_level)
if loaded is None:
return {}
project_dir = Path(project_path).resolve()
project_spec = importlib.machinery.PathFinder.find_spec(top_level, [str(project_dir)])
if project_spec is None:
# The project path does not provide this package, so nothing is shadowed.
return {}
loaded_locations = [getattr(loaded, "__file__", None), *getattr(loaded, "__path__", [])]
if any(_is_within(location, project_dir) for location in loaded_locations):
return {}
evicted = {name: sys.modules.pop(name) for name in _package_module_names(top_level)}
importlib.invalidate_caches()
_LOGGER.warning(
"Module %s was already imported from %s instead of the project path %s; "
"evicted %d module(s) so the project's copy is tested",
top_level,
getattr(loaded, "__file__", None),
project_dir,
len(evicted),
)
return evicted
def _setup_import_hook(
dynamic_constant_provider: DynamicConstantProvider | None,
) -> SubjectProperties:
_LOGGER.debug("Setting up instrumentation for %s", config.configuration.module_name)
subject_properties = SubjectProperties()
install_import_hook(
config.configuration.module_name,
subject_properties,
dynamic_constant_provider=dynamic_constant_provider,
)
return subject_properties
def _package_module_names(top_level: str) -> list[str]:
return [name for name in sys.modules if name == top_level or name.startswith(f"{top_level}.")]
def _restore_modules(top_level: str, modules: dict[str, types.ModuleType]) -> None:
for name in _package_module_names(top_level):
del sys.modules[name]
sys.modules.update(modules)
importlib.invalidate_caches()
def _import_sut(module_name: str, subject_properties: SubjectProperties) -> None:
# We need to activate the tracer so the import trace is recorded.
with (
time_limit(config.configuration.stopping.maximum_module_execution_timeout),
subject_properties.instrumentation_tracer,
):
# If the module is already imported, we need to reload it for the
# ExecutionTracer to successfully register the subject_properties
if module_name in sys.modules:
importlib.reload(sys.modules[module_name])
else:
importlib.import_module(module_name)
def _load_sut(subject_properties: SubjectProperties) -> bool:
module_name = config.configuration.module_name
evicted = _evict_shadowing_modules(module_name, config.configuration.project_path)
try:
_import_sut(module_name, subject_properties)
except (Exception, TestExecutionTimeoutError) as ex:
if not evicted:
# A module could not be imported because some dependencies
# are missing or it is malformed or any error is raised during the import
_LOGGER.exception("Failed to load SUT: %s", ex)
return False
_LOGGER.warning(
"Failed to load SUT from the project path (%s); "
"falling back to the previously imported copy",
ex,
)
_restore_modules(module_name.partition(".")[0], evicted)
try:
_import_sut(module_name, subject_properties)
except (Exception, TestExecutionTimeoutError) as fallback_ex:
_LOGGER.exception("Failed to load SUT: %s", fallback_ex)
return False
return True
def _setup_report_dir() -> bool:
# Report dir only needs to be created when statistics or coverage report is enabled.
if (
config.configuration.statistics_output.statistics_backend != config.StatisticsBackend.NONE
or config.configuration.statistics_output.create_coverage_report
):
report_dir = Path(config.configuration.statistics_output.report_dir).absolute()
try:
report_dir.mkdir(parents=True, exist_ok=True)
except (OSError, FileNotFoundError):
_LOGGER.exception(
"Cannot create report dir %s",
config.configuration.statistics_output.report_dir,
)
return False
return True
def _patch_random() -> None:
"""Patch random.Random.seed to use the Pynguin seed for determinism.
This must be called BEFORE the SUT is imported, because the SUT (or its
dependencies) may create random.Random instances at module level.
All seeded instances are collected and stored as
``random.Random.seed.__pynguin_instances__`` so callers can reseed every
living SUT-related instance before each test-case execution.
"""
if not getattr(random.Random.seed, "__pynguin_patched__", False):
orig_random_seed = random.Random.seed
tracked: weakref.WeakSet[random.Random] = weakref.WeakSet()
def _deterministic_random_seed(self: random.Random, x=None) -> None:
if x is None:
x = config.configuration.seeding.seed
elif type(x).__hash__ is object.__hash__:
# If x is an object that uses the default id-based hash, we replace it with
# a deterministic string to avoid non-determinism from memory addresses.
x = f"{type(x).__module__}.{type(x).__name__}"
orig_random_seed(self, x)
tracked.add(self)
_deterministic_random_seed.__pynguin_patched__ = True # type: ignore[attr-defined]
_deterministic_random_seed.__pynguin_instances__ = tracked # type: ignore[attr-defined]
random.Random.seed = _deterministic_random_seed # type: ignore[assignment,method-assign]
def _setup_random_number_generator() -> None:
"""Setup RNG."""
_LOGGER.info("Using seed %d", config.configuration.seeding.seed)
randomness.RNG.seed(config.configuration.seeding.seed)
random.seed(config.configuration.seeding.seed)
if config.configuration.pynguinml.ml_testing_enabled:
np_rng.init_rng(config.configuration.seeding.seed)
if FANDANGO_FAKER_AVAILABLE:
# Seed Faker
Faker.seed(config.configuration.seeding.seed)
def _setup_constant_seeding() -> tuple[ConstantProvider, DynamicConstantProvider | None]:
"""Collect constants from SUT, if enabled."""
# Use empty provider by default.
wrapped_provider: ConstantProvider = EmptyConstantProvider()
# We need to return the provider used for dynamic values separately,
# because it is later on used to hook up the instrumentation calls.
dynamic_constant_provider: DynamicConstantProvider | None = None
if config.configuration.seeding.constant_seeding:
_LOGGER.info("Collecting static constants from module under test")
module_names = (
[config.configuration.module_name]
if getattr(config.configuration, "module_name", None)
else None
)
constant_pool = collect_static_constants(
config.configuration.project_path, module_names=module_names
)
if len(constant_pool) == 0:
_LOGGER.info("No constants found")
else:
_LOGGER.info("Constants found: %s", len(constant_pool))
# Probability of 1.0 -> if a value is requested and available -> return it.
wrapped_provider = DelegatingConstantProvider(constant_pool, wrapped_provider, 1.0)
if config.configuration.seeding.dynamic_constant_seeding:
_LOGGER.info("Setting up runtime collection of constants")
dynamic_constant_provider = DynamicConstantProvider(
RestrictedConstantPool(max_size=config.configuration.seeding.max_dynamic_pool_size),
wrapped_provider,
config.configuration.seeding.seeded_dynamic_values_reuse_probability,
config.configuration.seeding.max_dynamic_length,
)
wrapped_provider = dynamic_constant_provider
return wrapped_provider, dynamic_constant_provider
def _setup_ml_testing_environment(test_cluster: ModuleTestCluster):
# load resources once so they get cached
tr.get_datatype_mapping()
tr.get_nparray_function(test_cluster)
tr.get_constructor_function(test_cluster)
def _verify_config() -> None:
"""Verify the configuration and raise an exception if something is invalid/not supported."""
# All combinations of algorithms and coverage metrics are supported.
migration_strategy = config.configuration.island.migration_strategy
migration_active = migration_strategy is not config.MigrationStrategy.DISABLED
if migration_active and config.configuration.algorithm not in {
config.Algorithm.DYNAMOSA,
config.Algorithm.LLDYNAMOSA,
}:
raise ConfigurationException(
"Island migration requires algorithm to be DYNAMOSA or LLDYNAMOSA -- "
"migration unlocks children via a goals manager only those two have."
)
if (
config.configuration.island.num_islands > 1
and config.configuration.stopping.maximum_search_time <= 0
):
raise ConfigurationException(
"Island mode requires a positive --maximum_search_time -- it is the only "
"shared stopping condition all islands use."
)
if config.configuration.island.num_islands > 1 and "fork" not in mp.get_all_start_methods():
raise ConfigurationException(
"Island mode requires the fork start method, which this platform does not "
"provide -- the islands and the LLM worker inherit the orchestrator's SUT "
"setup by forking."
)
if migration_strategy in {
config.MigrationStrategy.PERIODIC,
config.MigrationStrategy.COMBINED,
}:
island_config = config.configuration.island
worst_case_population = minimum_island_population_size(
config.configuration.search_algorithm.population,
island_config.population_allocation,
island_config.num_islands,
)
if island_config.periodic_migration_size > worst_case_population:
raise ConfigurationException(
f"periodic_migration_size (K={island_config.periodic_migration_size}) "
f"exceeds the smallest per-island population "
f"({worst_case_population}) under "
f"population_allocation={island_config.population_allocation.value} with "
f"{island_config.num_islands} islands -- periodic migration would have "
"to repeat or fabricate migrants to reach K."
)
if (
island_config.migrant_selection_policy is config.MigrantSelectionPolicy.RANK
and not 1.0 < island_config.migrant_rank_bias <= 2.0
):
raise ConfigurationException(
f"migrant_rank_bias ({island_config.migrant_rank_bias}) must be in the "
"range (1, 2] for rank selection of periodic migrants."
)
_verify_llm_worker_config(migration_active=migration_active)
def _verify_llm_worker_config(*, migration_active: bool) -> None:
"""Verify how LLDYNAMOSA, parallel islands and the LLM worker are combined.
Args:
migration_active: Whether any island migration strategy is enabled.
Raises:
ConfigurationException: If the combination is not supported.
"""
algorithm = config.configuration.algorithm
num_islands = config.configuration.island.num_islands
if not config.configuration.llm_worker.enabled:
if algorithm is config.Algorithm.LLDYNAMOSA and num_islands > 1:
raise ConfigurationException(
"LLDYNAMOSA with parallel islands requires the centralized LLM worker "
"(--llm_worker.enabled True); use DYNAMOSA for parallel island search "
"without the LLM worker."
)
return
if algorithm is not config.Algorithm.LLDYNAMOSA:
raise ConfigurationException(
f"The LLM worker requires algorithm=LLDYNAMOSA, got {algorithm.name}. "
"With the worker enabled, the islands run DYNAMOSA and receive "
"LLM-generated tests from the centralized worker."
)
if num_islands <= 1:
raise ConfigurationException(
"The LLM worker requires island.num_islands > 1 -- standalone "
"single-island use is not supported."
)
if (
config.configuration.llm_worker.immigration_routing is config.ImmigrationRouting.BROADCAST
and not migration_active
):
raise ConfigurationException(
"immigration_routing=BROADCAST requires island.migration_strategy to "
"not be DISABLED -- it delivers results via the same migration "
"mechanism."
)
def _is_random_entity(value: object, random_module: types.ModuleType) -> bool:
try:
if value is random_module:
return True
random_class = getattr(random_module, "Random", None)
if isinstance(random_class, type):
if isinstance(value, type) and issubclass(value, random_class):
return True
if isinstance(value, random_class):
return True
self_obj = getattr(value, "__self__", None)
if self_obj is not None and isinstance(self_obj, random_class):
return True
if getattr(value, "__module__", None) in {"random", "_random"}:
return True
except Exception: # noqa: BLE001
return False
return False
def _check_sut_uses_random(new_module_names: set[str]) -> bool:
"""Return True if any module loaded by the SUT import references Python's random.
Args:
new_module_names: Module names that were added to sys.modules as a side
effect of importing the SUT.
Returns:
True if the SUT or any transitive dependency uses Python's random module.
"""
random_module = sys.modules.get("random")
if random_module is None:
return False
if "random" in new_module_names:
return True
for n in new_module_names:
module = sys.modules.get(n)
if module is None or not isinstance(module, types.ModuleType):
continue
for value in list(vars(module).values()):
if _is_random_entity(value, random_module):
return True
if isinstance(value, type) and getattr(value, "__module__", None) == module.__name__:
for member in list(vars(value).values()):
if _is_random_entity(member, random_module):
return True
return False
def _setup_and_check() -> tuple[TestCaseExecutor, ModuleTestCluster, ConstantProvider] | None:
"""Load the System Under Test (SUT) i.e. the module that is tested.
Perform setup and some sanity checks.
Returns:
An optional tuple of test-case executor and test cluster
"""
if not _setup_path():
return None
wrapped_constant_provider, dynamic_constant_provider = _setup_constant_seeding()
subject_properties = _setup_import_hook(dynamic_constant_provider)
_patch_random()
modules_before_sut = set(sys.modules.keys())
if not _load_sut(subject_properties):
return None
new_sut_module_names = set(sys.modules.keys()) - modules_before_sut
if not _setup_report_dir():
return None
# Analyzing the SUT should not cause any coverage.
with subject_properties.instrumentation_tracer.temporarily_disable():
if (test_cluster := _setup_test_cluster()) is None:
return None
test_cluster.sut_uses_random = _check_sut_uses_random(new_sut_module_names)
# Make alias to make the following lines shorter...
stop = config.configuration.stopping
if config.configuration.subprocess:
executor: TestCaseExecutor = SubprocessTestCaseExecutor(
subject_properties=subject_properties,
maximum_test_execution_timeout=stop.maximum_test_execution_timeout,
test_execution_time_per_statement=stop.test_execution_time_per_statement,
)
else:
executor = TestCaseExecutor(
subject_properties=subject_properties,
maximum_test_execution_timeout=stop.maximum_test_execution_timeout,
test_execution_time_per_statement=stop.test_execution_time_per_statement,
)
_track_sut_data(subject_properties, test_cluster)
_setup_random_number_generator()
_setup_mock_generation(test_cluster)
if config.configuration.pynguinml.ml_testing_enabled:
_setup_ml_testing_environment(test_cluster)
# Detect which LLM strategy is used
stat.track_output_variable(RuntimeVariable.LLMStrategy, _detect_llm_strategy())
return executor, test_cluster, wrapped_constant_provider
def _cluster_candidate_classes(test_cluster: ModuleTestCluster) -> list[type]:
"""External, non-exception classes an untyped parameter could resolve to.
These are the classes Pynguin can generate, filtered to third-party
(non-stdlib) classes, so the mock pipeline can classify the few that match an
untyped parameter's usage.
Args:
test_cluster: The test cluster whose generatable types are inspected.
Returns:
The list of external, non-exception classes.
"""
from pynguin.analyses.typesystem import Instance # noqa: PLC0415
classes: list[type] = []
for typ in test_cluster.get_all_generatable_types():
if not isinstance(typ, Instance):
continue
raw = typ.type.raw_type
if not isinstance(raw, type) or issubclass(raw, BaseException):
continue
top = (getattr(raw, "__module__", "") or "").split(".", 1)[0]
if top and top not in sys.stdlib_module_names and top != "builtins":
classes.append(raw)
return classes
def _untyped_param_bindings(
module_path: Path, candidate_classes: list[type], mock_targets: set[str]
) -> dict[tuple[str, str], str]:
"""Map ``("<module>.<qualname>", param) -> boundary FQN`` for untyped params.
An untyped parameter whose attribute usage matches a class already classified
as a boundary (``mock_targets``) is bound directly, so the test factory can
inject a mock for it without type tracing.
Args:
module_path: Path to the module under test.
candidate_classes: classes an untyped parameter could match.
mock_targets: FQNs classified as boundaries to mock.
Returns:
A mapping from ``(callable key, parameter name)`` to boundary FQN.
"""
from pynguin.mock_generation.untyped_param_analyzer import ( # noqa: PLC0415
match_param_boundaries,
untyped_param_bindings,
)
# Only match against classes that are actual boundaries, so every match is a
# class we intend to mock (and whose FQN form aligns with mock_targets).
boundary_classes = [
cls
for cls in candidate_classes
if f"{getattr(cls, '__module__', '')}.{getattr(cls, '__name__', '')}" in mock_targets
]
if not boundary_classes:
return {}
bindings = untyped_param_bindings(module_path.read_text(encoding="utf-8"))
matched = match_param_boundaries(bindings, boundary_classes)
module_name = config.configuration.module_name
return {(f"{module_name}.{qualname}", param): fqn for (qualname, param), fqn in matched.items()}
def _apply_static_setups( # noqa: C901
module_path: Path,
mock_targets: set[str],
candidate_classes: list[type],
module_name: str,
untyped_bindings: dict[tuple[str, str], str],
) -> None:
"""Attach SUT-derived static setups to each parameter's own mock target.
For every mocked parameter, the branch-constant, iteration-default and
side-effect setups derived from that parameter's usage are attached to the
template of the target it binds to, so two dependencies in one module never
share setups. A bare template is created when none exists, and setups already
present from the proxy-cache are not duplicated.
Args:
module_path: SUT source file.
mock_targets: FQNs that will be mocked.
candidate_classes: classes an untyped parameter could match.
module_name: The module under test, used to build the callable key.
untyped_bindings: injector's ``(callable key, param) -> boundary FQN`` map.
"""
import ast # noqa: PLC0415
import pynguin.testcase.mock_templates_store as _mock_store # noqa: PLC0415
from pynguin.mock_generation import ( # noqa: PLC0415
ast_helpers,
mock_hint_generator,
static_setups,
)
from pynguin.mock_generation.mock_generator import ( # noqa: PLC0415
MockTemplate,
MutableSetup,
)
try:
tree = ast.parse(module_path.read_text(encoding="utf-8"))
except SyntaxError:
return
alias_map = ast_helpers.import_alias_map(tree)
candidates = candidate_classes or []
# Index the injector's untyped bindings by callable key, as generate_templates does.
bindings_by_func: dict[str, dict[str, str]] = {}
for (callable_key, param), fqn in (untyped_bindings or {}).items():
bindings_by_func.setdefault(callable_key, {})[param] = fqn
injector_mode = untyped_bindings is not None
def template_for(fqn: str) -> MockTemplate:
template = _mock_store.MOCK_TEMPLATES_BY_TARGET.get(fqn)
if template is None:
dep = fqn.rsplit(".", maxsplit=1)[-1].lower()
template = MockTemplate(
dependency=dep, import_path=fqn.split(".", maxsplit=1)[0], mock_target=fqn
)
_mock_store.MOCK_TEMPLATES_BY_TARGET[fqn] = template
return template
for qualname, func in mock_hint_generator.iter_named_functions(tree):
func_key = f"{module_name}.{qualname}" if module_name else qualname
per_func = bindings_by_func.get(func_key, {}) if injector_mode else None
params_to_target = mock_hint_generator.mocked_params(
func, alias_map, mock_targets, candidates, per_func
)
if not params_to_target:
continue
per_param = static_setups.function_param_setups(func)
for param, target in params_to_target.items():
setups = per_param.get(param)
if setups is None:
continue
mutable, lines = setups
template = template_for(target)
existing = {s.target for s in template.mutable_setups}
for chain, cands in mutable:
if chain not in existing:
template.mutable_setups.append(MutableSetup(target=chain, candidates=cands))
existing.add(chain)
for line in lines:
if line not in template.setup_lines:
template.setup_lines.append(line)
def _setup_mock_generation(test_cluster: ModuleTestCluster) -> None:
"""Run the mock pipeline and record its statistics.
Thin wrapper around :func:`_run_mock_generation` that always emits the
mock-generation runtime variables, so that baseline/disabled runs report
zeros and the columns stay comparable across configurations in the results.
Args:
test_cluster: The test cluster passed to the mock pipeline.
"""
metrics = _run_mock_generation(test_cluster)
stat.track_output_variable(RuntimeVariable.MockTargetsTotal, metrics["targets_total"])
stat.track_output_variable(RuntimeVariable.MockTargetsFromRules, metrics["from_rules"])
stat.track_output_variable(RuntimeVariable.MockTargetsFromLLM, metrics["from_llm"])
stat.track_output_variable(RuntimeVariable.MockHintTargets, metrics["hint_targets"])
stat.track_output_variable(RuntimeVariable.MockUntypedParams, metrics["untyped_params"])
stat.track_output_variable(RuntimeVariable.MockGenerationTime, metrics["time"])
def _run_mock_generation(test_cluster: ModuleTestCluster) -> dict[str, Any]:
"""Run the mock pipeline and store templates for statement insertion.
Feeds the proxy-cache-generated mock templates to the test factory (via the shared
``mock_templates_store``) so that, when Pynguin needs a value of a
mocked-library type, it inserts a mock statement configured from the
corresponding template (return values / side effects included).
Args:
test_cluster: supplies the classes an untyped parameter may statically
match, so those boundaries can be injected directly.
Returns:
A metrics mapping (``targets_total``, ``from_rules``, ``from_llm``,
``hint_targets``, ``untyped_params``, ``time``); all-zero when mock
generation does not run.
"""
metrics: dict[str, Any] = {
"targets_total": 0,
"from_rules": 0,
"from_llm": 0,
"hint_targets": 0,
"untyped_params": 0,
"time": 0.0,
}
if not config.configuration.mock_generation.mock_generation_enabled:
return metrics
from pynguin.mock_generation.proxy_cache_resolver import ( # noqa: PLC0415
get_proxy_cache_url,
)
if not get_proxy_cache_url():
_LOGGER.warning(
"Mock generation is enabled but no proxy-cache URL found. "
"Set PYNGUIN_PROXY_CACHE_URL or --proxy_cache_url. "
"Falling back to default Pynguin without mocks."
)
return metrics
module_name = config.configuration.module_name
module = sys.modules.get(module_name)
if module is None or not getattr(module, "__file__", None):
_LOGGER.warning("Mock generation: cannot locate file for %r - skipping", module_name)
return metrics
import pynguin.testcase.mock_templates_store as _mock_store # noqa: PLC0415
from pynguin.mock_generation.mock_generator import ( # noqa: PLC0415
MockGenerator,
)
base_rules_cache_id = config.configuration.mock_generation.base_rules_cache_id or None
# Classes an untyped parameter could statically match. Computed unconditionally
# so untyped parameters are mocked directly (no type tracing required).
candidate_classes = _cluster_candidate_classes(test_cluster)
start = time.perf_counter()
try:
gen = MockGenerator(use_proxy=True)
gen.generate_for_module(
Path(str(module.__file__)),
base_rules_cache_id,
candidate_classes=candidate_classes,
)
_mock_store.MOCK_TARGETS = gen.mock_targets
_mock_store.MOCK_UNTYPED_PARAMS = _untyped_param_bindings(
Path(str(module.__file__)), candidate_classes, gen.mock_targets
)
metrics["targets_total"] = len(gen.mock_targets)
metrics["untyped_params"] = len(_mock_store.MOCK_UNTYPED_PARAMS)
metrics["from_rules"] = gen.source_counts.get("rule", 0)
metrics["from_llm"] = gen.source_counts.get("llm", 0)
sources = ", ".join(f"{k}={v}" for k, v in sorted(gen.source_counts.items())) or "none"
_LOGGER.info(
"Mock generation: %d target(s) for %s [by source: %s]: %s",
len(gen.mock_targets),
module_name,
sources,
sorted(gen.mock_targets),
)
if _mock_store.MOCK_UNTYPED_PARAMS:
_LOGGER.info(
"Mock generation: %d untyped parameter(s) bound directly: %s",
len(_mock_store.MOCK_UNTYPED_PARAMS),
sorted(_mock_store.MOCK_UNTYPED_PARAMS),
)
# Return-value hints so mocks reach value-dependent branches.
from pynguin.mock_generation import ( # noqa: PLC0415
mock_hint_generator,
)
templates = mock_hint_generator.generate_templates(
Path(str(module.__file__)),
gen.mock_targets,
candidate_classes,
module_name=module_name,
untyped_bindings=_mock_store.MOCK_UNTYPED_PARAMS,
)
_mock_store.MOCK_TEMPLATES_BY_TARGET = templates
metrics["hint_targets"] = len(templates)
# Static, deterministic return-value setups derived from the SUT itself
# (branch constants + iteration defaults), so bare mocks reach value-
# dependent branches even when the proxy-cache returned no hints.
_apply_static_setups(
Path(str(module.__file__)),
gen.mock_targets,
candidate_classes,
module_name,
_mock_store.MOCK_UNTYPED_PARAMS,
)
for fqn, template in templates.items():
for setup in template.mutable_setups:
_LOGGER.info(
"Mock generation: mutable setup %s.%s candidates=%s",
fqn,
setup.target,
setup.candidates,
)
except Exception as exc: # noqa: BLE001
_LOGGER.warning("Mock generation setup failed: %s - proceeding without mocks", exc)
finally:
metrics["time"] = time.perf_counter() - start
return metrics
def _count_mock_statements(generation_result: tsc.TestSuiteChromosome) -> int:
"""Count the mock statements present in the final generated test suite.
Args:
generation_result: the generated test suite.
Returns:
The number of mock statements across all test cases.
"""
return sum(
statement.mock_info is not None
for chromosome in generation_result.test_case_chromosomes
for statement in chromosome.test_case.statements()
)
def _detect_llm_strategy() -> str:
if config.configuration.large_language_model.hybrid_initial_population:
return (
f"Hybrid-Initial-Population-"
f"{config.configuration.large_language_model.llm_test_case_percentage}"
)
if config.configuration.large_language_model.call_llm_on_stall_detection:
return "LLM-On-Stall-Detection"
if config.configuration.large_language_model.call_llm_for_uncovered_targets:
return "LLM-For-Initial-Uncovered-Targets"
if config.configuration.test_case_output.assertion_generation == config.AssertionGenerator.LLM:
return "LLM-Assertion-Generator"
return ""
def _track_sut_data(subject_properties: SubjectProperties, test_cluster: ModuleTestCluster) -> None:
"""Track data from the SUT.
Args:
subject_properties: The properties of the subject under test.
test_cluster: the test cluster
"""
stat.track_output_variable(
RuntimeVariable.CodeObjects,
len(subject_properties.existing_code_objects),
)
stat.track_output_variable(
RuntimeVariable.Predicates,
len(subject_properties.existing_predicates),
)
stat.track_output_variable(
RuntimeVariable.Lines,
len(subject_properties.existing_lines),
)
cyclomatic_complexities: list[int] = [
code.cfg.cyclomatic_complexity for code in subject_properties.existing_code_objects.values()
]
stat.track_output_variable(
RuntimeVariable.McCabeCodeObject, json.dumps(cyclomatic_complexities)
)
test_cluster.track_statistics_values(stat.track_output_variable)
if config.CoverageMetric.BRANCH in config.configuration.search_algorithm.coverage_metrics:
stat.track_output_variable(
RuntimeVariable.ImportBranchCoverage,
ff.compute_branch_coverage(
subject_properties.instrumentation_tracer.import_trace,
subject_properties,
),
)
if config.CoverageMetric.LINE in config.configuration.search_algorithm.coverage_metrics:
stat.track_output_variable(
RuntimeVariable.ImportLineCoverage,
ff.compute_line_coverage(
subject_properties.instrumentation_tracer.import_trace,
subject_properties,
),
)
def _get_coverage_ff_from_algorithm(
algorithm: GenerationAlgorithm, function_type: type[ff.TestSuiteCoverageFunction]
) -> ff.TestSuiteCoverageFunction:
"""Retrieve the coverage function for a test suite of a given coverage type.
Args:
algorithm: The test generation strategy
function_type: the type of coverage function to receive
Returns:
The coverage function for a test suite for this run of the given type
"""
test_suite_coverage_func = None
for coverage_func in algorithm.test_suite_coverage_functions:
if isinstance(coverage_func, function_type):
test_suite_coverage_func = coverage_func
assert test_suite_coverage_func, "The required coverage function was not initialised"
return test_suite_coverage_func
def _reload_instrumentation_loader(
coverage_metrics: set[config.CoverageMetric],
dynamic_constant_provider: DynamicConstantProvider | None,
subject_properties: SubjectProperties,
):
module_name = config.configuration.module_name
try:
with time_limit(config.configuration.stopping.maximum_module_execution_timeout):
module = importlib.import_module(module_name)
except (Exception, TestExecutionTimeoutError) as ex:
_LOGGER.exception("Failed to import SUT: %s", ex)
return False
first_finder: InstrumentationFinder | None = None
for finder in sys.meta_path:
if isinstance(finder, InstrumentationFinder):
first_finder = finder
break
if first_finder is None:
_LOGGER.error("InstrumentationFinder not found in sys.meta_path")
return False
first_finder.update_instrumentation_metrics(
subject_properties=subject_properties,
coverage_metrics=coverage_metrics,
dynamic_constant_provider=dynamic_constant_provider,
)
try:
with (
time_limit(config.configuration.stopping.maximum_module_execution_timeout),
subject_properties.instrumentation_tracer,
):
importlib.reload(module)
except (Exception, TestExecutionTimeoutError) as ex:
_LOGGER.exception("Failed to reload SUT: %s", ex)
return False
return True
def _reset_cache_for_result(generation_result):
generation_result.invalidate_cache()
for test_case in generation_result.test_case_chromosomes:
test_case.invalidate_cache()
test_case.remove_last_execution_result()
def _track_final_metrics(
algorithm,
executor: TestCaseExecutor,
generation_result: tsc.TestSuiteChromosome,
constant_provider: ConstantProvider,
) -> set[config.CoverageMetric] | None:
"""Track the final coverage metrics.
Re-loads all required instrumentations for metrics that were not already
calculated and tracked during the result generation.
These metrics are then also calculated on the result, which is executed
once again with the new instrumentation.
Args:
algorithm: the used test-generation algorithm
executor: the testcase executor of the run
generation_result: the generated testsuite containing assertions
constant_provider: the constant provider required for the
reloading of the module
Returns:
The set of tracked coverage metrics, including the ones that we optimised for
or None if the tracking failed.
"""
output_variables = config.configuration.statistics_output.output_variables
# Alias for shorter lines
cov_metrics = config.configuration.search_algorithm.coverage_metrics
metrics_for_reinstrumenation: set[config.CoverageMetric] = set(cov_metrics)
to_calculate: list[tuple[RuntimeVariable, ff.TestSuiteCoverageFunction]] = []
add_additional_metrics(
algorithm=algorithm,
cov_metrics=cov_metrics,
executor=executor,
metrics_for_reinstrumentation=metrics_for_reinstrumenation,
output_variables=output_variables,
to_calculate=to_calculate,
)
# Assertion Checked Coverage is special...
if RuntimeVariable.AssertionCheckedCoverage in output_variables:
metrics_for_reinstrumenation.add(config.CoverageMetric.CHECKED)
executor.set_instrument(True)
executor.add_remote_observer(RemoteAssertionExecutionObserver())
assertion_checked_coverage_ff = ff.TestSuiteAssertionCheckedCoverageFunction(executor)
to_calculate.append((
RuntimeVariable.AssertionCheckedCoverage,
assertion_checked_coverage_ff,
))
# re-instrument the files
dynamic_constant_provider = None
if isinstance(constant_provider, DynamicConstantProvider):
dynamic_constant_provider = constant_provider
if not _reload_instrumentation_loader(
metrics_for_reinstrumenation,
dynamic_constant_provider,
executor.subject_properties,
):
# Re-instrumenting the SUT for the final metrics re-executes the
# module's top-level code. Modules that register a name into a
# process-global, duplicate-rejecting registry on import (e.g.
# networkx's dispatch registry or SQLAlchemy's type registry) raise on
# this second execution. Rather than discarding the already-generated
# test suite, keep it and fall back to the metrics that were already
# tracked during generation.
_LOGGER.warning(
"Could not re-instrument the SUT for final metrics; exporting the "
"generated test suite with the metrics tracked during generation "
"instead of discarding it."
)
return set(cov_metrics)
# force new execution of the test cases after new instrumentation
_reset_cache_for_result(generation_result)
# set value for each newly calculated variable
for runtime_variable, coverage_ff in to_calculate:
generation_result.add_coverage_function(coverage_ff)
_LOGGER.info(f"Calculating resulting {runtime_variable.value}") # noqa: G004
stat.track_output_variable(
runtime_variable, generation_result.get_coverage_for(coverage_ff)
)
# Collect other final stats on result
stat.track_output_variable(RuntimeVariable.FinalLength, generation_result.length())
stat.track_output_variable(RuntimeVariable.FinalSize, generation_result.size())
stat.track_output_variable(
RuntimeVariable.MockStatementsInserted, _count_mock_statements(generation_result)
)
# reset whether to instrument tests and assertions as well as the SUT
instrument_test = config.CoverageMetric.CHECKED in cov_metrics
executor.set_instrument(instrument_test)
return metrics_for_reinstrumenation
def add_additional_metrics( # noqa: D103
*,
algorithm,
cov_metrics,
executor,
metrics_for_reinstrumentation,
output_variables,
to_calculate,
):
if (
RuntimeVariable.FinalLineCoverage in output_variables
and config.CoverageMetric.LINE not in cov_metrics
):
metrics_for_reinstrumentation.add(config.CoverageMetric.LINE)
line_cov_ff = ff.TestSuiteLineCoverageFunction(executor)
to_calculate.append((RuntimeVariable.FinalLineCoverage, line_cov_ff))
elif config.CoverageMetric.LINE in cov_metrics:
# If we optimised for lines, we still want to get the final line coverage.
to_calculate.append((
RuntimeVariable.FinalLineCoverage,
_get_coverage_ff_from_algorithm(algorithm, ff.TestSuiteLineCoverageFunction),
))
if (
RuntimeVariable.FinalBranchCoverage in output_variables
and config.CoverageMetric.BRANCH not in cov_metrics
):
metrics_for_reinstrumentation.add(config.CoverageMetric.BRANCH)
branch_cov_ff = ff.TestSuiteBranchCoverageFunction(executor)
to_calculate.append((RuntimeVariable.FinalBranchCoverage, branch_cov_ff))
elif config.CoverageMetric.BRANCH in cov_metrics:
# If we optimised for branches, we still want to get the final branch coverage.
to_calculate.append((
RuntimeVariable.FinalBranchCoverage,
_get_coverage_ff_from_algorithm(algorithm, ff.TestSuiteBranchCoverageFunction),
))
def _add_slicing_observer_if_needed(executor: TestCaseExecutor) -> None:
"""Traces slices for test cases after execution, which checked coverage needs.
Args:
executor: The executor to add the slicing observer to.
"""
if config.CoverageMetric.CHECKED in config.configuration.search_algorithm.coverage_metrics:
from pynguin.slicer.statementslicingobserver import ( # noqa: PLC0415
RemoteStatementSlicingObserver,
)
executor.add_remote_observer(RemoteStatementSlicingObserver())
def _run() -> ReturnCode:
_verify_config()
if (setup_result := _setup_and_check()) is None:
return ReturnCode.SETUP_FAILED
executor, test_cluster, constant_provider = setup_result
_add_slicing_observer_if_needed(executor)
coverage_metrics = config.configuration.search_algorithm.coverage_metrics
algorithm: GenerationAlgorithm = _instantiate_test_generation_strategy(
executor, test_cluster, constant_provider
)
_LOGGER.info("Start generating test cases")
generation_result = algorithm.generate_tests()
if algorithm.resources_left():
_LOGGER.info("Algorithm stopped before using all resources.")
else:
_LOGGER.info("Stopping condition reached")
for stop in algorithm.stopping_conditions:
_LOGGER.info("%s", stop)
_LOGGER.info("Stop generating test cases")
# Executions that happen after this point should not influence the
# search statistics
executor.clear_observers()
executor.clear_remote_observers()
return finalize_generation_result(
algorithm, executor, test_cluster, constant_provider, generation_result, coverage_metrics
)
def _track_pre_refinement_metrics(
algorithm: GenerationAlgorithm,
generation_result: tsc.TestSuiteChromosome,
) -> None:
"""Track pre-refinement metrics before test suite export and LLM refinement.
Args:
algorithm: Generation algorithm used for coverage-related setup.
generation_result: Finalized test suite chromosome.
"""
post_ass_assertions = sum(
len(tc.test_case.get_assertions()) for tc in generation_result.test_case_chromosomes
)
stat.track_output_variable(
RuntimeVariable.PostAssertionGenerationAssertions, post_ass_assertions
)
stat.track_output_variable(RuntimeVariable.PreRefinementAssertions, post_ass_assertions)
stat.track_output_variable(RuntimeVariable.Assertions, post_ass_assertions)
cov_functions = getattr(algorithm, "test_suite_coverage_functions", [])
try:
if cov_functions:
post_ass_coverage = statistics.mean(
generation_result.get_coverage_for(cov_func) for cov_func in cov_functions
)
else:
post_ass_coverage = generation_result.get_coverage()
except Exception: # noqa: BLE001
post_ass_coverage = 0.0
stat.track_output_variable(RuntimeVariable.PostAssertionGenerationCoverage, post_ass_coverage)
stat.track_output_variable(RuntimeVariable.PreRefinementCoverage, post_ass_coverage)
stat.track_output_variable(RuntimeVariable.Coverage, post_ass_coverage)
[docs]
def finalize_generation_result( # noqa: C901, PLR0917
algorithm: GenerationAlgorithm,
executor: TestCaseExecutor,
test_cluster: ModuleTestCluster,
constant_provider: ConstantProvider,
generation_result: tsc.TestSuiteChromosome,
coverage_metrics: list[config.CoverageMetric],
) -> ReturnCode:
"""Finalize a generated test suite and write the resulting artifacts.
This includes assertion generation, minimization, final metric tracking,
export, LLM refinement, and coverage-report generation.
Args:
algorithm: Generation algorithm used for coverage-related setup.
executor: Executor used to run test cases.
test_cluster: Test cluster for the current run.
constant_provider: Provider for generated constants.
generation_result: Generated or aggregated test suite to finalize.
coverage_metrics: Coverage metrics configured for the run.
Returns:
The result of exporting the finalized test suite.
"""
_track_search_metrics(algorithm, generation_result, coverage_metrics)
# Generate assertions FIRST
_generate_assertions(executor, generation_result, test_cluster)
# Minimize assertions if configured (requires re-instrumentation for checked_instructions)
ass_gen = config.configuration.test_case_output.assertion_generation
if (
ass_gen == config.AssertionGenerator.CHECKED_MINIMIZING
and RuntimeVariable.AssertionCheckedCoverage
in config.configuration.statistics_output.output_variables
):
_prepare_for_assertion_minimization(executor, generation_result, constant_provider)
_LOGGER.info("Minimizing assertions")
_minimize_assertions(generation_result)
# Statement minimization LAST (now assertion-aware)
try:
_LOGGER.info("Minimizing test cases")
_minimize(generation_result, algorithm)
except Exception as ex:
_LOGGER.exception("Minimization failed: %s", ex)
if (
tracked_metrics := _track_final_metrics(
algorithm,
executor,
generation_result,
constant_provider,
)
) is None:
return ReturnCode.FINAL_METRICS_TRACKING_FAILED
executor.subject_properties.instrumentation_tracer.disable()
_track_pre_refinement_metrics(algorithm, generation_result)
# Export the generated test suites
if config.configuration.test_case_output.export_strategy == config.ExportStrategy.PY_TEST:
try:
exported_test_file = _export_chromosome(
generation_result,
sut_uses_random=test_cluster.sut_uses_random,
subject_properties=executor.subject_properties,
test_cluster=test_cluster,
)
# Apply LLM-based test refinement if enabled
if config.configuration.llm_refinement.enabled:
from pynguin.refinement.refiner import refine_generated_tests # noqa: PLC0415
test_file_path = exported_test_file
try:
_LOGGER.info("Starting LLM-based test refinement for %s", test_file_path)
# refine_generated_tests reads the LLM model and API key from
# the shared large_language_model configuration and records its
# own statistics via the statistics tracker.
refinement_stats = refine_generated_tests(
test_file_path=test_file_path,
module_name=config.configuration.module_name,
max_repair_iterations=config.configuration.llm_refinement.max_repair_iterations,
max_tests=config.configuration.llm_refinement.max_tests,
subject_properties=executor.subject_properties,
)
_LOGGER.info("Refinement complete: %s", refinement_stats)
except Exception as refinement_ex:
_LOGGER.exception("LLM refinement failed: %s", refinement_ex)
except Exception as ex:
_LOGGER.exception("Export to PyTest failed: %s", ex)
if config.configuration.statistics_output.create_coverage_report:
try:
coverage_report = get_coverage_report(
generation_result,
executor.subject_properties,
tracked_metrics,
)
render_coverage_report(
coverage_report,
Path(config.configuration.statistics_output.report_dir) / "cov_report.html",
datetime.datetime.now(), # noqa: DTZ005
)
render_xml_coverage_report(
coverage_report,
Path(config.configuration.statistics_output.report_dir) / "cov_report.xml",
datetime.datetime.now(), # noqa: DTZ005
)
except Exception as e: # noqa: BLE001
_LOGGER.warning(
"Failed to create coverage report: %s. ",
e,
)
_collect_miscellaneous_statistics(test_cluster)
if not stat.write_statistics():
_LOGGER.error("Failed to write statistics data")
if generation_result.size() == 0:
# not able to generate one test case
return ReturnCode.NO_TESTS_GENERATED
return ReturnCode.OK
def _run_llm() -> ReturnCode:
def load_sut_code() -> str:
project_path = Path(config.configuration.project_path)
module_name = config.configuration.module_name.replace(".", "/") + ".py"
sut_file = project_path / module_name
return sut_file.read_text()
# Imported lazily so that ``import pynguin.generator`` does not require the
# optional ``openai`` extra (and its ``pyyaml`` dependency) to be installed.
from pynguin.large_language_model.prompts.prompt import Prompt # noqa: PLC0415
class GeneratorGeneratePrompt(Prompt):
_resource_name = "generator_generate"
def _template_vars(self):
return ["sut_code"]
prompt = GeneratorGeneratePrompt()
request = prompt.render(sut_code=load_sut_code())
client = OpenAIClient()
response = client.send(request)
if not response:
return ReturnCode.NO_TESTS_GENERATED
code = extract_python_code(response)
module_name = config.configuration.module_name.replace(".", "_")
target_file = (
Path(config.configuration.test_case_output.output_path).resolve() / f"test_{module_name}.py"
)
target_file.write_text(code)
return ReturnCode.OK
def _check_coverage(original_coverages: list[float], minimized_coverages: list[float]) -> bool:
"""Check if the coverage after minimization is the same as before.
Args:
original_coverages: The coverages before minimization
minimized_coverages: The coverages after minimization
Returns:
If the coverage is still the same
"""
is_same = all(map(math.isclose, original_coverages, minimized_coverages))
if is_same:
_LOGGER.info("Coverage after minimization is the same as before: %s", minimized_coverages)
else:
_LOGGER.warning(
"Coverage after minimization changed from %s to %s",
original_coverages,
minimized_coverages,
)
return is_same
def _minimize(generation_result, algorithm=None):
truncation = pp.ExceptionTruncation()
generation_result.accept(truncation)
if config.configuration.test_case_output.post_process:
unused_vars_minimizer = pp.UnusedStatementsTestCaseVisitor()
minimization_strategy = (
config.configuration.test_case_output.minimization.test_case_minimization_strategy
)
if minimization_strategy != config.MinimizationStrategy.NONE and algorithm is not None:
fitness_functions = algorithm.test_suite_coverage_functions
assert len(fitness_functions) > 0, "No test suite coverage functions available"
original_coverages = [
generation_result.get_coverage_for(fitness_function)
for fitness_function in fitness_functions
]
# Save a copy of the original test suite before minimization
original_test_suite = generation_result.clone()
# Select the appropriate minimization visitor based on the strategy
if (
config.configuration.test_case_output.minimization.test_case_minimization_direction
== config.MinimizationDirection.FORWARD
):
iterative_minimizer: pp.IterativeMinimizationVisitor = (
pp.ForwardIterativeMinimizationVisitor(fitness_functions)
)
else:
iterative_minimizer = pp.BackwardIterativeMinimizationVisitor(fitness_functions)
# Check if we should use the combined minimization approach
if (
config.configuration.test_case_output.minimization.test_case_minimization_strategy
== config.MinimizationStrategy.COMBINED
):
combined_minimizer = pp.CombinedMinimizationVisitor(fitness_functions)
generation_result.accept(combined_minimizer)
_LOGGER.info(
"Combined minimization removed %d statement(s)",
combined_minimizer.removed_statements,
)
else:
# Apply traditional test case minimization strategies
test_case_minimizer = pp.TestCasePostProcessor([
unused_vars_minimizer,
iterative_minimizer,
])
generation_result.accept(test_case_minimizer)
_LOGGER.info(
"Removed %d statement(s) from test casesusing %s minimization",
iterative_minimizer.removed_statements,
minimization_strategy.value,
)
# Apply test suite minimization to remove redundant test cases
if (
config.configuration.test_case_output.minimization.test_case_minimization_strategy
== config.MinimizationStrategy.SUITE
):
test_suite_minimizer = pp.TestSuiteMinimizationVisitor(fitness_functions)
generation_result.accept(test_suite_minimizer)
if test_suite_minimizer.removed_test_cases > 0:
_LOGGER.info(
"Removed %d test case(s) from test suite during minimization",
test_suite_minimizer.removed_test_cases,
)
minimized_coverages = [
generation_result.get_coverage_for(fitness_function)
for fitness_function in fitness_functions
]
is_same = _check_coverage(original_coverages, minimized_coverages)
if not is_same:
# Restore unminimized test suite
_LOGGER.info("Restoring unminimized test suite due to coverage loss")
# Replace the current test suite with the original one
generation_result.test_case_chromosomes = [
test.clone() for test in original_test_suite.test_case_chromosomes
]
# Mark the test suite as changed
generation_result.changed = True
# Verify that coverage is restored
restored_coverage = generation_result.get_coverage_for(fitness_functions)
_LOGGER.info("Coverage after restoration: %.4f", restored_coverage)
else:
unused_primitives_removal = pp.TestCasePostProcessor([unused_vars_minimizer])
generation_result.accept(unused_primitives_removal)
# Remove empty test cases after minimization
empty_test_case_remover = pp.EmptyTestCaseRemover()
generation_result.accept(empty_test_case_remover)
def _minimize_assertions(generation_result: tsc.TestSuiteChromosome):
_LOGGER.info("Minimizing assertions based on checked coverage")
assertion_minimizer = pp.AssertionMinimization()
generation_result.accept(assertion_minimizer)
stat.track_output_variable(
RuntimeVariable.Assertions, len(assertion_minimizer.remaining_assertions)
)
stat.track_output_variable(
RuntimeVariable.DeletedAssertions,
len(assertion_minimizer.deleted_assertions),
)
def _prepare_for_assertion_minimization(
executor: TestCaseExecutor,
generation_result: tsc.TestSuiteChromosome,
constant_provider: ConstantProvider,
) -> None:
"""Prepare for assertion minimization by populating checked_instructions.
Re-instruments the module with RemoteAssertionExecutionObserver,
executes the test suite to populate checked_instructions on assertions,
then returns executor to non-instrumented state.
Args:
executor: Test case executor
generation_result: Test suite with assertions to minimize
constant_provider: Constant provider for re-instrumentation
"""
# Add assertion execution observer
executor.set_instrument(True)
executor.add_remote_observer(RemoteAssertionExecutionObserver())
# Re-instrument with CHECKED metric
metrics_for_reinstrumentation = {config.CoverageMetric.CHECKED}
dynamic_constant_provider = None
if isinstance(constant_provider, DynamicConstantProvider):
dynamic_constant_provider = constant_provider
if not _reload_instrumentation_loader(
metrics_for_reinstrumentation,
dynamic_constant_provider,
executor.subject_properties,
):
_LOGGER.warning("Failed to reload instrumentation for assertion minimization")
return
# Force re-execution to populate checked_instructions
_reset_cache_for_result(generation_result)
# Execute to populate checked_instructions
# This happens automatically when coverage is computed
assertion_checked_ff = ff.TestSuiteAssertionCheckedCoverageFunction(executor)
generation_result.add_coverage_function(assertion_checked_ff)
_ = generation_result.get_coverage_for(assertion_checked_ff)
# Return executor to non-instrumented state
executor.set_instrument(False)
executor.clear_observers()
executor.clear_remote_observers()
def _setup_mutation_analysis_assertion_generator(
executor: TestCaseExecutor,
*,
start_time: float | None = None,
maximum_time: float | None = None,
) -> ag.MutationAnalysisAssertionGenerator:
_LOGGER.info("Setup mutation generator")
_LOGGER.info("Import module %s", config.configuration.module_name)
with time_limit(config.configuration.stopping.maximum_module_execution_timeout):
module = importlib.import_module(config.configuration.module_name)
_LOGGER.info("Mutate module %s", module.__name__)
mutation_controller = create_mutation_controller(module)
assertion_generator: ag.MutationAnalysisAssertionGenerator
if config.configuration.test_case_output.assertion_generation is config.AssertionGenerator.LLM:
assertion_generator = lag.MutationAnalysisLLMAssertionGenerator(
executor,
mutation_controller,
filtering_executor=ag.create_filtering_executor(executor),
start_time=start_time,
maximum_time=maximum_time,
)
else:
assertion_generator = ag.MutationAnalysisAssertionGenerator(
executor,
mutation_controller,
filtering_executor=ag.create_filtering_executor(executor),
start_time=start_time,
maximum_time=maximum_time,
)
_LOGGER.info("Generated %d mutants", mutation_controller.mutant_count())
return assertion_generator
def _generate_assertions(executor, generation_result, test_cluster):
ass_gen = config.configuration.test_case_output.assertion_generation
if ass_gen != config.AssertionGenerator.NONE:
_LOGGER.info("Start generating assertions")
generator: cv.ChromosomeVisitor
if ass_gen == config.AssertionGenerator.LLM:
generation_result.accept(lag.LLMAssertionGenerator(test_cluster))
generator = _setup_mutation_analysis_assertion_generator(executor)
elif ass_gen == config.AssertionGenerator.MUTATION_ANALYSIS:
generator = _setup_mutation_analysis_assertion_generator(executor)
else:
generator = ag.AssertionGenerator(
executor, filtering_executor=ag.create_filtering_executor(executor)
)
generation_result.accept(generator)
# Track total number of assertions after generation
total_assertions = sum(
len(test_case_chrom.test_case.get_assertions())
for test_case_chrom in generation_result.test_case_chromosomes
)
stat.track_output_variable(RuntimeVariable.Assertions, total_assertions)
def _track_search_metrics(
algorithm: GenerationAlgorithm,
generation_result: tsc.TestSuiteChromosome,
coverage_metrics: list[config.CoverageMetric],
) -> None:
"""Track multiple set coverage metrics of the generated test suites.
This possibly re-executes the test suites.
Args:
algorithm: The test generation strategy
generation_result: The resulting chromosome of the generation strategy
coverage_metrics: The selected coverage metrics to guide the search
"""
for metric, runtime, fitness_type in [
(
config.CoverageMetric.LINE,
RuntimeVariable.LineCoverage,
ff.TestSuiteLineCoverageFunction,
),
(
config.CoverageMetric.BRANCH,
RuntimeVariable.BranchCoverage,
ff.TestSuiteBranchCoverageFunction,
),
(
config.CoverageMetric.CHECKED,
RuntimeVariable.StatementCheckedCoverage,
ff.TestSuiteStatementCheckedCoverageFunction,
),
]:
if metric in coverage_metrics:
coverage_function: ff.TestSuiteCoverageFunction = _get_coverage_ff_from_algorithm(
algorithm, cast("type", fitness_type)
)
stat.track_output_variable(
runtime, generation_result.get_coverage_for(coverage_function)
)
# Write overall coverage data of result
stat.current_individual(generation_result)
try:
post_search_cov = generation_result.get_coverage()
except Exception: # noqa: BLE001
post_search_cov = 0.0
stat.track_output_variable(RuntimeVariable.PostSearchCoverage, post_search_cov)
post_search_assertions = sum(
len(tc.test_case.get_assertions()) for tc in generation_result.test_case_chromosomes
)
stat.track_output_variable(RuntimeVariable.PostSearchAssertions, post_search_assertions)
def _instantiate_test_generation_strategy(
executor: TestCaseExecutor,
test_cluster: ModuleTestCluster,
constant_provider: ConstantProvider,
) -> GenerationAlgorithm:
factory = gaf.TestSuiteGenerationAlgorithmFactory(executor, test_cluster, constant_provider)
return factory.get_search_algorithm()
def _collect_miscellaneous_statistics(test_cluster: ModuleTestCluster) -> None:
test_cluster.log_cluster_statistics()
stat.track_output_variable(RuntimeVariable.TargetModule, config.configuration.module_name)
stat.track_output_variable(RuntimeVariable.RandomSeed, randomness.RNG.get_seed())
stat.track_output_variable(
RuntimeVariable.ConfigurationId,
config.configuration.statistics_output.configuration_id,
)
stat.track_output_variable(RuntimeVariable.RunId, config.configuration.statistics_output.run_id)
stat.track_output_variable(
RuntimeVariable.ProjectName, config.configuration.statistics_output.project_name
)
for runtime_variable, value in stat.variables_generator:
stat.set_output_variable_for_runtime_variable(runtime_variable, value)
def _export_chromosome(
chromosome: chrom.Chromosome,
*,
sut_uses_random: bool = False,
subject_properties: SubjectProperties | None = None,
test_cluster: ModuleTestCluster | None = None,
) -> Path:
"""Export the given chromosome.
Args:
chromosome: the chromosome to export.
sut_uses_random: Whether the SUT transitively uses Python's random module.
If True, an autouse fixture that reseeds random is emitted.
subject_properties: The subject properties, forwarded to the writer so it can
disable/rebind the instrumentation tracer during per-statement exception
detection (otherwise the tracer's thread-identity guard aborts the
exporter's re-execution and pollutes the suite with bogus
``pytest.raises(TracingAbortedException)`` wrappers).
test_cluster: Optional test cluster supplying registered external imports.
Returns:
The name of the target file
"""
output_dir = Path(config.configuration.test_case_output.output_path).resolve()
generation_result = cast("tsc.TestSuiteChromosome", chromosome)
if test_cluster is not None and hasattr(test_cluster, "external_imports"):
generation_result.external_imports.update(test_cluster.external_imports)
writer = export.TestSuiteWriter(no_xfail=config.configuration.test_case_output.no_xfail)
target_file = writer.write(
generation_result,
config.configuration.module_name,
output_dir,
project_path=config.configuration.project_path,
format_with_black=config.configuration.test_case_output.format_with_black,
seed=config.configuration.seeding.seed if sut_uses_random else None,
subject_properties=subject_properties,
)
_LOGGER.info("Written %i test cases to %s", chromosome.size(), target_file)
return Path(target_file)