lsst.pipe.base.testUtils Namespace Reference

Functions
def	makeQuantum (task, butler, dataId, ioDataIds)
def	runTestQuantum (task, butler, quantum, mockRun=True)
def	assertValidOutput (task, result)

Function Documentation

assertValidOutput()

def lsst.pipe.base.testUtils.assertValidOutput	(		task,
			result
	)

Test that the output of a call to ``run`` conforms to its own
connections.

Parameters
----------
task : `lsst.pipe.base.PipelineTask`
    The task whose connections need validation. This is a fully-configured
    task object to support features such as optional outputs.
result : `lsst.pipe.base.Struct`
    A result object produced by calling ``task.run``.

Raises
-------
AssertionError:
    Raised if ``result`` does not match what's expected from ``task's``
    connections.

Definition at line 245 of file testUtils.py.

def assertValidOutput(task, result):
    """Test that the output of a call to ``run`` conforms to its own
    connections.
 
    Parameters
    ----------
    task : `lsst.pipe.base.PipelineTask`
        The task whose connections need validation. This is a fully-configured
        task object to support features such as optional outputs.
    result : `lsst.pipe.base.Struct`
        A result object produced by calling ``task.run``.
 
    Raises
    -------
    AssertionError:
        Raised if ``result`` does not match what's expected from ``task's``
        connections.
    """
    connections = task.config.ConnectionsClass(config=task.config)
    recoveredOutputs = result.getDict()
 
    for name in connections.outputs:
        connection = connections.__getattribute__(name)
        # name
        try:
            output = recoveredOutputs[name]
        except KeyError:
            raise AssertionError(f"No such output: {name}")
        # multiple
        if connection.multiple:
            if not isinstance(output, collections.abc.Sequence):
                raise AssertionError(f"Expected {name} to be a sequence, got {output} instead.")
        else:
            # use lazy evaluation to not use StorageClassFactory unless
            # necessary
            if isinstance(output, collections.abc.Sequence) \
                    and not issubclass(
                        StorageClassFactory().getStorageClass(connection.storageClass).pytype,
                        collections.abc.Sequence):
                raise AssertionError(f"Expected {name} to be a single value, got {output} instead.")
        # no test for storageClass, as I'm not sure how much persistence
        # depends on duck-typing

makeQuantum()

def lsst.pipe.base.testUtils.makeQuantum	(		task,
			butler,
			dataId,
			ioDataIds
	)

Create a Quantum for a particular data ID(s).

Parameters
----------
task : `lsst.pipe.base.PipelineTask`
    The task whose processing the quantum represents.
butler : `lsst.daf.butler.Butler`
    The collection the quantum refers to.
dataId: any data ID type
    The data ID of the quantum. Must have the same dimensions as
    ``task``'s connections class.
ioDataIds : `collections.abc.Mapping` [`str`]
    A mapping keyed by input/output names. Values must be data IDs for
    single connections and sequences of data IDs for multiple connections.

Returns
-------
quantum : `lsst.daf.butler.Quantum`
    A quantum for ``task``, when called with ``dataIds``.

Definition at line 35 of file testUtils.py.

def makeQuantum(task, butler, dataId, ioDataIds):
    """Create a Quantum for a particular data ID(s).
 
    Parameters
    ----------
    task : `lsst.pipe.base.PipelineTask`
        The task whose processing the quantum represents.
    butler : `lsst.daf.butler.Butler`
        The collection the quantum refers to.
    dataId: any data ID type
        The data ID of the quantum. Must have the same dimensions as
        ``task``'s connections class.
    ioDataIds : `collections.abc.Mapping` [`str`]
        A mapping keyed by input/output names. Values must be data IDs for
        single connections and sequences of data IDs for multiple connections.
 
    Returns
    -------
    quantum : `lsst.daf.butler.Quantum`
        A quantum for ``task``, when called with ``dataIds``.
    """
    connections = task.config.ConnectionsClass(config=task.config)
 
    try:
        inputs = defaultdict(list)
        outputs = defaultdict(list)
        for name in itertools.chain(connections.inputs, connections.prerequisiteInputs):
            connection = connections.__getattribute__(name)
            _checkDataIdMultiplicity(name, ioDataIds[name], connection.multiple)
            ids = _normalizeDataIds(ioDataIds[name])
            for id in ids:
                ref = _refFromConnection(butler, connection, id)
                inputs[ref.datasetType].append(ref)
        for name in connections.outputs:
            connection = connections.__getattribute__(name)
            _checkDataIdMultiplicity(name, ioDataIds[name], connection.multiple)
            ids = _normalizeDataIds(ioDataIds[name])
            for id in ids:
                ref = _refFromConnection(butler, connection, id)
                outputs[ref.datasetType].append(ref)
        quantum = Quantum(taskClass=type(task),
                          dataId=dataId,
                          inputs=inputs,
                          outputs=outputs)
        return quantum
    except KeyError as e:
        raise ValueError("Mismatch in input data.") from e
 
 

runTestQuantum()

def lsst.pipe.base.testUtils.runTestQuantum	(		task,
			butler,
			quantum,
			mockRun = True
	)

Run a PipelineTask on a Quantum.

Parameters
----------
task : `lsst.pipe.base.PipelineTask`
    The task to run on the quantum.
butler : `lsst.daf.butler.Butler`
    The collection to run on.
quantum : `lsst.daf.butler.Quantum`
    The quantum to run.
mockRun : `bool`
    Whether or not to replace ``task``'s ``run`` method. The default of
    `True` is recommended unless ``run`` needs to do real work (e.g.,
    because the test needs real output datasets).

Returns
-------
run : `unittest.mock.Mock` or `None`
    If ``mockRun`` is set, the mock that replaced ``run``. This object can
    be queried for the arguments ``runQuantum`` passed to ``run``.

Definition at line 209 of file testUtils.py.

def runTestQuantum(task, butler, quantum, mockRun=True):
    """Run a PipelineTask on a Quantum.
 
    Parameters
    ----------
    task : `lsst.pipe.base.PipelineTask`
        The task to run on the quantum.
    butler : `lsst.daf.butler.Butler`
        The collection to run on.
    quantum : `lsst.daf.butler.Quantum`
        The quantum to run.
    mockRun : `bool`
        Whether or not to replace ``task``'s ``run`` method. The default of
        `True` is recommended unless ``run`` needs to do real work (e.g.,
        because the test needs real output datasets).
 
    Returns
    -------
    run : `unittest.mock.Mock` or `None`
        If ``mockRun`` is set, the mock that replaced ``run``. This object can
        be queried for the arguments ``runQuantum`` passed to ``run``.
    """
    _resolveTestQuantumInputs(butler, quantum)
    butlerQc = ButlerQuantumContext(butler, quantum)
    connections = task.config.ConnectionsClass(config=task.config)
    inputRefs, outputRefs = connections.buildDatasetRefs(quantum)
    if mockRun:
        with unittest.mock.patch.object(task, "run") as mock, \
                unittest.mock.patch("lsst.pipe.base.ButlerQuantumContext.put"):
            task.runQuantum(butlerQc, inputRefs, outputRefs)
            return mock
    else:
        task.runQuantum(butlerQc, inputRefs, outputRefs)
        return None