lsst.pipe.base.testUtils Namespace Reference

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

Function Documentation

assertValidInitOutput()

def lsst.pipe.base.testUtils.assertValidInitOutput

(

task

)

Test that a constructed task conforms to its own init-connections.

Parameters
----------
task : `lsst.pipe.base.PipelineTask`
    The task whose connections need validation.

Raises
------
AssertionError:
    Raised if ``task`` does not have the state expected from ``task's``
    connections.

Definition at line 315 of file testUtils.py.

def assertValidInitOutput(task):
    """Test that a constructed task conforms to its own init-connections.
 
    Parameters
    ----------
    task : `lsst.pipe.base.PipelineTask`
        The task whose connections need validation.
 
    Raises
    ------
    AssertionError:
        Raised if ``task`` does not have the state expected from ``task's``
        connections.
    """
    connections = task.config.ConnectionsClass(config=task.config)
 
    for name in connections.initOutputs:
        connection = connections.__getattribute__(name)
        _assertAttributeMatchesConnection(task, name, connection)
 
 

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 290 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)
 
    for name in connections.outputs:
        connection = connections.__getattribute__(name)
        _assertAttributeMatchesConnection(result, name, connection)
 
 

getInitInputs()

def lsst.pipe.base.testUtils.getInitInputs	(		butler,
			config
	)

Return the initInputs object that would have been passed to a
`~lsst.pipe.base.PipelineTask` constructor.

Parameters
----------
butler : `lsst.daf.butler.Butler`
    The repository to search for input datasets. Must have
    pre-configured collections.
config : `lsst.pipe.base.PipelineTaskConfig`
    The config for the task to be constructed.

Returns
-------
initInputs : `dict` [`str`]
    A dictionary of objects in the format of the ``initInputs`` parameter
    to `lsst.pipe.base.PipelineTask`.

Definition at line 336 of file testUtils.py.

def getInitInputs(butler, config):
    """Return the initInputs object that would have been passed to a
    `~lsst.pipe.base.PipelineTask` constructor.
 
    Parameters
    ----------
    butler : `lsst.daf.butler.Butler`
        The repository to search for input datasets. Must have
        pre-configured collections.
    config : `lsst.pipe.base.PipelineTaskConfig`
        The config for the task to be constructed.
 
    Returns
    -------
    initInputs : `dict` [`str`]
        A dictionary of objects in the format of the ``initInputs`` parameter
        to `lsst.pipe.base.PipelineTask`.
    """
    connections = config.connections.ConnectionsClass(config=config)
    initInputs = {}
    for name in connections.initInputs:
        attribute = getattr(connections, name)
        # Get full dataset type to check for consistency problems
        dsType = DatasetType(attribute.name, butler.registry.dimensions.extract(set()),
                             attribute.storageClass)
        # All initInputs have empty data IDs
        initInputs[name] = butler.get(dsType)
 
    return initInputs

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 40 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 214 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