cosmic_ray package

Subpackages

Submodules

cosmic_ray.cli module

cosmic_ray.config module

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

cosmic_ray.plugins module

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:

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 version info.

cosmic_ray.work_db module

cosmic_ray.work_item module

Classes for describing work and results.

class cosmic_ray.work_item.MutationSpec(module_path: pathlib.Path, operator_name: str, occurrence: int, start_pos: Tuple[int, int], end_pos: Tuple[int, int], operator_args: Dict[str, Any] = <factory>)

Bases: object

Description of a single mutation.

class cosmic_ray.work_item.StrEnum

Bases: str, enum.Enum

An Enum subclass with str values.

class cosmic_ray.work_item.TestOutcome

Bases: cosmic_ray.work_item.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: Tuple[cosmic_ray.work_item.MutationSpec])

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.

classmethod single(job_id, mutation: cosmic_ray.work_item.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: cosmic_ray.work_item.WorkerOutcome, output: Optional[str] = None, test_outcome: Optional[cosmic_ray.work_item.TestOutcome] = None, diff: Optional[str] = None)

Bases: object

The result of a single mutation and test run.

diff = None
is_killed

Whether the mutation should be considered ‘killed’

output = None
test_outcome = None
class cosmic_ray.work_item.WorkerOutcome

Bases: cosmic_ray.work_item.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.