Source code for pynguin.generator

#  This file is part of Pynguin.
#
#  SPDX-FileCopyrightText: 2019-2023 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 json
import logging
import sys
import threading

from pathlib import Path
from typing import TYPE_CHECKING
from typing import cast

import pynguin.assertion.assertiongenerator as ag
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.statistics as stat

from pynguin.analyses.constants import ConstantProvider
from pynguin.analyses.constants import DelegatingConstantProvider
from pynguin.analyses.constants import DynamicConstantProvider
from pynguin.analyses.constants import EmptyConstantProvider
from pynguin.analyses.constants import RestrictedConstantPool
from pynguin.analyses.constants import collect_static_constants
from pynguin.analyses.module import generate_test_cluster
from pynguin.instrumentation.machinery import InstrumentationFinder
from pynguin.instrumentation.machinery import install_import_hook
from pynguin.slicer.statementslicingobserver import StatementSlicingObserver
from pynguin.testcase import export
from pynguin.testcase.execution import AssertionExecutionObserver
from pynguin.testcase.execution import ExecutionTracer
from pynguin.testcase.execution import TestCaseExecutor
from pynguin.utils import randomness
from pynguin.utils.report import get_coverage_report
from pynguin.utils.report import render_coverage_report
from pynguin.utils.report import render_xml_coverage_report
from pynguin.utils.statistics.runtimevariable import RuntimeVariable


if TYPE_CHECKING:
    from pynguin.analyses.module import ModuleTestCluster
    from pynguin.ga.algorithms.generationalgorithm import GenerationAlgorithm


[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."""
_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…") return _run() finally: _LOGGER.info("Stop Pynguin Test Generation…")
def _setup_test_cluster() -> ModuleTestCluster | None: test_cluster = generate_test_cluster( config.configuration.module_name, config.configuration.type_inference.type_inference_strategy, query_type4py=config.configuration.type_inference.type4py, ) 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 _setup_import_hook( coverage_metrics: set[config.CoverageMetric], dynamic_constant_provider: DynamicConstantProvider | None, ) -> ExecutionTracer: _LOGGER.debug("Setting up instrumentation for %s", config.configuration.module_name) tracer = ExecutionTracer() install_import_hook( config.configuration.module_name, tracer, coverage_metrics=coverage_metrics, dynamic_constant_provider=dynamic_constant_provider, ) return tracer def _load_sut(tracer: ExecutionTracer) -> bool: try: # We need to set the current thread ident so the import trace is recorded. tracer.current_thread_identifier = threading.current_thread().ident importlib.import_module(config.configuration.module_name) except ImportError as ex: # A module could not be imported because some dependencies # are missing or it is malformed _LOGGER.exception("Failed to load SUT: %s", 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 _setup_random_number_generator() -> None: """Setup RNG.""" _LOGGER.info("Using seed %d", config.configuration.seeding.seed) randomness.RNG.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") constant_pool = collect_static_constants(config.configuration.project_path) 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_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() tracer = _setup_import_hook( set(config.configuration.statistics_output.coverage_metrics), dynamic_constant_provider, ) if not _load_sut(tracer): return None if not _setup_report_dir(): return None # Analyzing the SUT should not cause any coverage. tracer.disable() if (test_cluster := _setup_test_cluster()) is None: return None tracer.enable() # Make alias to make the following lines shorter... stop = config.configuration.stopping executor = TestCaseExecutor( tracer, maximum_test_execution_timeout=stop.maximum_test_execution_timeout, test_execution_time_per_statement=stop.test_execution_time_per_statement, ) _track_sut_data(tracer, test_cluster) _setup_random_number_generator() return executor, test_cluster, wrapped_constant_provider def _track_sut_data(tracer: ExecutionTracer, test_cluster: ModuleTestCluster) -> None: """Track data from the SUT. Args: tracer: the execution tracer test_cluster: the test cluster """ stat.track_output_variable( RuntimeVariable.CodeObjects, len(tracer.get_subject_properties().existing_code_objects), ) stat.track_output_variable( RuntimeVariable.Predicates, len(tracer.get_subject_properties().existing_predicates), ) stat.track_output_variable( RuntimeVariable.Lines, len(tracer.get_subject_properties().existing_lines), ) cyclomatic_complexities: list[int] = [ code.original_cfg.cyclomatic_complexity for code in tracer.get_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.statistics_output.coverage_metrics ): stat.track_output_variable( RuntimeVariable.ImportBranchCoverage, ff.compute_branch_coverage( tracer.import_trace, tracer.get_subject_properties() ), ) if ( config.CoverageMetric.LINE in config.configuration.statistics_output.coverage_metrics ): stat.track_output_variable( RuntimeVariable.ImportLineCoverage, ff.compute_line_coverage( tracer.import_trace, tracer.get_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, tracer: ExecutionTracer, ): module_name = config.configuration.module_name module = importlib.import_module(module_name) tracer.current_thread_identifier = threading.current_thread().ident first_finder: InstrumentationFinder | None = None for finder in sys.meta_path: if isinstance(finder, InstrumentationFinder): first_finder = finder break assert first_finder is not None first_finder.update_instrumentation_metrics( tracer=tracer, coverage_metrics=coverage_metrics, dynamic_constant_provider=dynamic_constant_provider, ) importlib.reload(module) 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]: """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. """ output_variables = config.configuration.statistics_output.output_variables # Alias for shorter lines cov_metrics = config.configuration.statistics_output.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_observer(AssertionExecutionObserver(executor.tracer)) 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 _reload_instrumentation_loader( metrics_for_reinstrumenation, dynamic_constant_provider, executor.tracer ) # 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) ) ass_gen = config.configuration.test_case_output.assertion_generation if ( ass_gen == config.AssertionGenerator.CHECKED_MINIMIZING and RuntimeVariable.AssertionCheckedCoverage in output_variables ): _minimize_assertions(generation_result) # Collect other final stats on result stat.track_output_variable(RuntimeVariable.FinalLength, generation_result.length()) stat.track_output_variable(RuntimeVariable.FinalSize, generation_result.size()) # 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 _run() -> ReturnCode: if (setup_result := _setup_and_check()) is None: return ReturnCode.SETUP_FAILED executor, test_cluster, constant_provider = setup_result # traces slices for test cases after execution coverage_metrics = config.configuration.statistics_output.coverage_metrics if config.CoverageMetric.CHECKED in coverage_metrics: executor.add_observer(StatementSlicingObserver(executor.tracer)) 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() _track_search_metrics(algorithm, generation_result, coverage_metrics) _remove_statements_after_exceptions(generation_result) _generate_assertions(executor, generation_result) tracked_metrics = _track_final_metrics( algorithm, executor, generation_result, constant_provider ) # Export the generated test suites if ( config.configuration.test_case_output.export_strategy == config.ExportStrategy.PY_TEST ): _export_chromosome(generation_result) if config.configuration.statistics_output.create_coverage_report: coverage_report = get_coverage_report( generation_result, executor, 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 ) _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 _remove_statements_after_exceptions(generation_result): truncation = pp.ExceptionTruncation() generation_result.accept(truncation) if config.configuration.test_case_output.post_process: unused_primitives_removal = pp.TestCasePostProcessor( [pp.UnusedStatementsTestCaseVisitor()] ) generation_result.accept(unused_primitives_removal) # TODO(fk) add more postprocessing stuff. 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 _generate_assertions(executor, generation_result): ass_gen = config.configuration.test_case_output.assertion_generation if ass_gen != config.AssertionGenerator.NONE: _LOGGER.info("Start generating assertions") if ass_gen == config.AssertionGenerator.MUTATION_ANALYSIS: generator: cv.ChromosomeVisitor = ag.MutationAnalysisAssertionGenerator( executor ) else: generator = ag.AssertionGenerator(executor) generation_result.accept(generator) 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[ff.TestSuiteCoverageFunction], 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) 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, file_name_suffix: str = "", ) -> None: """Export the given chromosome. Args: chromosome: the chromosome to export. file_name_suffix: Suffix that can be added to the file name to distinguish between different results e.g., failing and succeeding test cases. Returns: The name of the target file """ module_name = config.configuration.module_name.replace(".", "_") target_file = ( Path(config.configuration.test_case_output.output_path).resolve() / f"test_{module_name}{file_name_suffix}.py" ) export_visitor = export.PyTestChromosomeToAstVisitor() chromosome.accept(export_visitor) export.save_module_to_file( export_visitor.to_module(), target_file, format_with_black=config.configuration.test_case_output.format_with_black, ) _LOGGER.info("Written %i test cases to %s", chromosome.size(), target_file)