cosmic_ray package

Subpackages

Submodules

cosmic_ray.cli module

This is the command-line program for cosmic ray.

Here we manage command-line parsing and launching of the internal machinery that does mutation testing.

cosmic_ray.cli.main(argv=None)

Invoke the cosmic ray evaluation.

Parameters:

argv – the command line arguments

cosmic_ray.config module

Configuration module.

class cosmic_ray.config.ConfigDict

Bases: dict

A dictionary subclass that contains the application configuration.

property distributor_config

The configuration for the named distributor.

property distributor_name

The name of the distributor to use.

property operators_config

The configuration for specified operators.

This is a dict mapping operator names to dicts which represent keyword-arguments for parameterizing an operator. Each keyword arg dict is a single parameterization of the operator, and each parameterized operator will be executed once for each parameterization.

sub(*segments)

Get a sub-configuration.

property test_command

The command to run to execute tests.

property timeout

The timeout (seconds) for tests.

exception cosmic_ray.config.ConfigError

Bases: Exception

Base class for exceptions raised by ConfigDict.

exception cosmic_ray.config.ConfigKeyError

Bases: ConfigError, KeyError

KeyError subclass raised by ConfigDict.

exception cosmic_ray.config.ConfigValueError

Bases: ConfigError, ValueError

ValueError subclass raised by ConfigDict.

cosmic_ray.config.deserialize_config(sz) → ConfigDict

Parse a serialized config into a ConfigDict.

cosmic_ray.config.load_config(filename=None)

Load a configuration from a file or stdin.

If filename is None or “-”, then configuration gets read from stdin.

Returns: A ConfigDict.

Raises: ConfigError: If there is an error loading the config.

cosmic_ray.config.serialize_config(config)

Return the serialized form of config.

cosmic_ray.exceptions module

exception cosmic_ray.exceptions.CosmicRayTestingException

Bases: Exception

Exception that we use for exception replacement.

cosmic_ray.modules module

Functions related to finding modules for testing.

cosmic_ray.modules.filter_paths(paths, excluded_paths)

Filter out path matching one of excluded_paths glob

Parameters:

  • paths – path to filter.

  • excluded_paths – List for glob of modules to exclude.

Returns:

An iterable of paths Python modules (i.e. *py files).

cosmic_ray.modules.find_modules(module_paths)

Find all modules in the module (possibly package) represented by module_path.

Parameters:

module_paths – A list of pathlib.Path to Python packages or modules.

Returns:

An iterable of paths Python modules (i.e. *py files).

cosmic_ray.mutating module

Support for making mutations to source code.

class cosmic_ray.mutating.MutationVisitor(occurrence, operator)

Bases: Visitor

Visitor that mutates a module with the specific occurrence of an operator.

This will perform at most one mutation in a walk of an AST. If this performs a mutation as part of the walk, it will store the mutated node in the mutant attribute. If the walk does not result in any mutation, mutant will be None.

Note that mutant is just the specifically mutated node. It will generally be a part of the larger AST which is returned from walk().

classmethod mutate_code(source, operator, occurence)
classmethod mutate_path(module_path, operator, occurrence)

Mutate a module in place on disk.

Parameters:

  • module_path (Path) – The path to the module file.

  • operator (Operator) – The operator to apply.

  • occurrence (int) – The occurrence of the operator to apply.

Returns:

The original code and the mutated code (or None)

Return type:

tuple[str, str|None]

property mutation_applied

Whether this visitor has applied a mutation.

visit(node)

Called for each node in the walk.

This should return a node that will replace the node argument in the AST. This can be the node argument itself, a new node, or None. If None is returned, then the node is removed from the tree.

Parameters:

node – The node currently being visited.

Returns:

A node or None.

cosmic_ray.mutating.apply_mutation(module_path, operator, occurrence)

Apply a specific mutation to a file on disk.

Parameters:

  • module_path – The path to the module to mutate.

  • operator – The operator instance to use.

  • occurrence – The occurrence of the operator to apply.

Returns:

A (unmutated-code, mutated-code) tuple to the with-block. If there was no mutation performed, the mutated-code is None.

cosmic_ray.mutating.mutate_and_test(mutations: Iterable[MutationSpec], test_command, timeout) → WorkResult

Apply a sequence of mutations, run thest tests, and reports the results.

This is fundamentally the mutation(s)-and-test-run implementation at the heart of Cosmic Ray.

There are three high-level ways that a worker can finish. First, it could fail exceptionally, meaning that some uncaught exception made its way from some part of the operation to terminate the function. This function will intercept all exceptions and return it in a non-exceptional structure.

Second, the mutation machinery may determines that - for any of the mutations - there is no mutation to be made (e.g. the ‘occurrence’ is too high). In this case there is no way to report a test result (i.e. killed, survived, or incompetent) so a special value is returned indicating that no mutation is possible.

Finally, and hopefully normally, the worker will find that it can run a test. It will do so and report back the result - killed, survived, or incompetent - in a structured way.

Parameters:

  • mutations – An iterable of MutationSpecs describing the mutations to make.

  • test_command – The command to execute to run the tests

  • timeout – The maximum amount of time (seconds) to let the tests run

Returns:

A WorkResult.

Raises:

  • This will generally not raise any exceptions. Rather, exceptions will be reported using the 'exception' –

  • result-type in the return value. –

cosmic_ray.mutating.mutate_code(code, operator, occurrence)

Apply a specific mutation to a code string.

Parameters:

  • code – The code to mutate.

  • operator – The operator instance to use.

  • occurrence – The occurrence of the operator to apply.

Returns:

The mutated code, or None if no mutation was applied.

cosmic_ray.mutating.use_mutation(module_path, operator, occurrence)

A context manager that applies a mutation for the duration of a with-block.

This applies a mutation to a file on disk, and after the with-block it put the unmutated code back in place.

Parameters:

  • module_path – The path to the module to mutate.

  • operator – The Operator instance to use.

  • occurrence – The occurrence of the operator to apply.

Yields:

A (unmutated-code, mutated-code) tuple to the with-block. If there was no mutation performed, the mutated-code is None.

cosmic_ray.plugins module

Query and retrieve the various plugins in Cosmic Ray.

cosmic_ray.plugins.distributor_names()

Get all distributor plugin names.

Returns:

A sequence of distributor names.

cosmic_ray.plugins.get_distributor(name)

Get the distributor by name.

cosmic_ray.plugins.get_operator(name)

Get an operator class from a provider plugin.

Parameters:

name – The name of the operator class.

Returns:

The operator class object (i.e. not an instance).

cosmic_ray.plugins.operator_names()

Get all operator names.

Returns:

A sequence of operator names.

cosmic_ray.progress module

Management of progress reporting.

The design of this subsystem is such that progress reporting is decoupled from updating the current progress. This allows for progress reporting functions which can be invoked from another context, such as a SIGINFO signal handler.

As such, reporter callables are responsible only for displaying current progress, and must be capable of retrieving the latest progress state from elsewhere when invoked. This may be global state or reached through references that the reporter callable was constructed with. Typically the latest progress state will be updated by the routine whose progress is being monitored.

To report the current progress invoke report_progress().

To manage installation and deinstallation of progress reporting functions use the reports_progress() decorator for whole-function contexts, or the progress_reporter() context manager for narrower contexts.

It is the responsibility of the client to manage any thread-safety issues. You should assume that progress reporting functions can be called asynchronously, at any time, from the main thread.

Example:

_progress_message = ""

def _update_foo_progress(i, n):
    global _progress_message
    _progress_message = "{i} of {n} complete".format(i=i, n=n)

def _report_foo_progress(stream):
    print(_progress_message, file=stream)

@reports_progress(_report_foo_progress)
def foo(n):
for i in range(n):
    _update_foo_progress(i, n)

# ...

signal.signal(signal.SIGINFO,
              lambda *args: report_progress())
cosmic_ray.progress.install_progress_reporter(reporter)

Install a progress reporter.

Where possible prefer to use the progress_reporter() context manager or reports_progress() decorator factory.

Parameters:

reporter – A zero-argument callable to report progress. The callable provided should have the means to both retrieve and display current progress information.

cosmic_ray.progress.progress_reporter(reporter)

A context manager to install and remove a progress reporting function.

Parameters:

reporter – A zero-argument callable to report progress. The callable provided should have the means to both retrieve and display current progress information.

cosmic_ray.progress.report_progress(stream=None)

Report progress from any currently installed reporters.

Parameters:

stream – The text stream (default: sys.stderr) to which progress will be reported.

cosmic_ray.progress.reports_progress(reporter)

A decorator factory to mark functions which report progress.

Parameters:

reporter – A zero-argument callable to report progress. The callable provided should have the means to both retrieve and display current progress information.

cosmic_ray.progress.uninstall_progress_reporter(reporter)

Uninstall a progress reporter.

Where possible prefer to use the progress_reporter() context manager or reports_progress() decorator factory.

Parameters:

reporter – A callable previously installed by install_progress_reporter().

cosmic_ray.testing module

Support for running tests in a subprocess.

cosmic_ray.testing.run_tests(command, timeout)

Run test command in a subprocess.

If the command exits with status 0, then we assume that all tests passed. If it exits with any other code, we assume a test failed. If the call to launch the subprocess throws an exception, we consider the test ‘incompetent’.

Tests which time out are considered ‘killed’ as well.

Parameters:

  • command (str) – The command to execute.

  • timeout (number) – The maximum number of seconds to allow the tests to run.

Return: A tuple (TestOutcome, output) where the output is a string

containing the output of the command.

cosmic_ray.timing module

Support for timing the execution of functions.

This is primarily intended to support baselining, but it’s got some reasonable generic functionality.

class cosmic_ray.timing.Timer

Bases: object

A simple context manager for timing events.

Generally use it like this:

with Timer() as t:
    do_something()
print(t.elapsed())
property elapsed

Get the elapsed time between the last call to reset and now.

Returns a datetime.timedelta object.

reset()

Set the elapsed time back to 0.

cosmic_ray.version module

cosmic_ray.work_db module

Implementation of the WorkDB.

class cosmic_ray.work_db.MutationSpecStorage(**kwargs)

Bases: Base

Database model for MutationSpecs

definition_name
end_pos_col
end_pos_row
job_id
module_path
occurrence
operator_args
operator_name
start_pos_col
start_pos_row
work_item
class cosmic_ray.work_db.WorkDB(path, mode)

Bases: object

WorkDB is the database that keeps track of mutation testing work progress.

Essentially, there’s a row in the DB for each mutation that needs to be executed in some run. These initially start off with no results, and results are added as they’re completed.

class Mode(enums: Type[Enum], **kw: Any)
class Mode(*enums: str, **kw: Any)

Bases: Enum

Modes in which a WorkDB may be opened.

create = 1
open = 2
add_work_item(work_item)

Add a WorkItem.

Parameters:

work_item – A WorkItem.

add_work_items(work_items)

Add multiple WorkItems.

Parameters:

work_items – an iterable of WorkItem.

clear()

Clear all work items from the session.

This removes any associated results as well.

close()

Close the database.

property completed_work_items

Iterable of (work-item, result)s for all completed items.

name()

A name for this database.

Derived from the constructor arguments.

property num_results

The number of results.

property num_work_items

The number of work items.

property pending_work_items

Iterable of all pending work items. In random order.

property results

An iterable of all (job-id, WorkResult)s.

set_multiple_results(job_ids, result)

Set the result for all job IDs.

This will overwrite any existing results for the jobs.

Parameters:

  • job_id – The IDs of the WorkItem to set the results for.

  • result – A WorkResult indicating the result of the job.

Raises:

KeyError – If there is no work-item with a matching job-id.

set_result(job_id, result)

Set the result for a job.

This will overwrite any existing results for the job.

Parameters:

  • job_id – The ID of the WorkItem to set the result for.

  • result – A WorkResult indicating the result of the job.

Raises:

KeyError – If there is no work-item with a matching job-id.

property work_items

An iterable of all of WorkItems in the db.

This includes both WorkItems with and without results.

class cosmic_ray.work_db.WorkItemStorage(**kwargs)

Bases: Base

Database model for WorkItem.

job_id
mutations
class cosmic_ray.work_db.WorkResultStorage(**kwargs)

Bases: Base

Database model for WorkResult.

diff
job_id
output
test_outcome
worker_outcome
cosmic_ray.work_db.use_db(path, mode=1)

Open a DB in file path in mode mode as a context manager.

On exiting the context the DB will be automatically closed.

Parameters:

path – The path to the DB file.

Raises:

FileNotFoundError – If mode is Mode.open and path does not exist.

cosmic_ray.work_item module

Classes for describing work and results.

class cosmic_ray.work_item.MutationSpec(module_path, operator_name: str, occurrence, start_pos: tuple[int, int], end_pos: tuple[int, int], operator_args: dict[str, Any] = NOTHING, definition_name: str | None = None)

Bases: object

Description of a single mutation.

definition_name: str | None
end_pos: tuple[int, int]
module_path: Path
occurrence: int
operator_args: dict[str, Any]
operator_name: str
start_pos: tuple[int, int]
class cosmic_ray.work_item.StrEnum(new_class_name, /, names, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

An Enum subclass with str values.

class cosmic_ray.work_item.TestOutcome(*values)

Bases: StrEnum

A enum of the possible outcomes for any mutant test run.

INCOMPETENT = 'incompetent'
KILLED = 'killed'
SURVIVED = 'survived'
class cosmic_ray.work_item.WorkItem(job_id: str, mutations)

Bases: object

A collection (possibly empty) of mutations to perform for a single test.

This ability to perform more than one mutation for a single test run is how we support higher-order mutations.

job_id: str
mutations: tuple[MutationSpec, ...]
classmethod single(job_id, mutation: MutationSpec)

Construct a WorkItem with a single mutation.

Parameters:

  • job_id – The ID of the job.

  • mutation – The single mutation for the WorkItem.

Returns:

A new WorkItem instance.

class cosmic_ray.work_item.WorkResult(worker_outcome: WorkerOutcome, output: str | None = None, test_outcome: TestOutcome | None = None, diff: str | None = None)

Bases: object

The result of a single mutation and test run.

diff: str | None
property is_killed

Whether the mutation should be considered ‘killed’

output: str | None
test_outcome: TestOutcome | None
worker_outcome: WorkerOutcome
class cosmic_ray.work_item.WorkerOutcome(*values)

Bases: StrEnum

Possible outcomes for a worker.

ABNORMAL = 'abnormal'
EXCEPTION = 'exception'
NORMAL = 'normal'
NO_TEST = 'no-test'
SKIPPED = 'skipped'

Module contents

Cosmic Ray is a mutation testing tool for Python.