suitcase/utils/tests/conftest.py
def generate_data(RE, detector_list, event_type):  # noqa
    '''A fixture that returns event data for a number of test cases.

    Returns a list of (name, doc) tuples for the plan passed in as an arg.

    Parameters
    ----------
    RE : object
        pytest fixture object imported from `bluesky.test.conftest`
    detector_list : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of detectors
    event_type : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of 'event_types'.
    '''

    def _generate_data_func(plan, skip_tests_with=None, md=None):
        '''Generates data to be used for testing of suitcase.*.export(..)
        functions

        Parameters
        ----------
        plan : the plan to use to generate the test data

        Returns
        -------
        collector : list
            A list of (name, doc) tuple pairs generated by the run engine.
        skip_tests_with : list, optional
            any test having request.param in this list will be skipped
        md : dict, optional
            metadata to be passed to the RunEngine
        '''
        if skip_tests_with is None:
            skip_tests_with = []
        if md is None:
            md = {}

        # define the output lists and an internal list.
        collector = []
        event_list = []

        # define the collector function depending on the event_type
        if event_type(skip_tests_with) == 'event':
            def collect(name, doc):
                collector.append((name, doc))
                if name == 'event':
                    event_list.append(doc)
        elif event_type(skip_tests_with) == 'event_page':
            def collect(name, doc):
                if name == 'event':
                    event_list.append(doc)
                elif name == 'stop':
                    collector.append(('event_page',
                                      event_model.pack_event_page(
                                          *event_list)))
                    collector.append((name, doc))
                else:
                    collector.append((name, doc))
        elif event_type(skip_tests_with) == 'bulk_events':
            def collect(name, doc):
                if name == 'event':
                    event_list.append(doc)
                elif name == 'stop':
                    collector.append(('bulk_events', {'primary': event_list}))
                    collector.append((name, doc))
                else:
                    collector.append((name, doc))
        else:
            raise UnknownEventType('Unknown event_type kwarg passed to '
                                   'suitcase.utils.events_data')

        # collect the documents
        RE.subscribe(collect)
        RE(plan(detector_list(skip_tests_with)), md=md)

        return collector

    return _generate_data_func
A fixture that returns event data for a number of test cases.

    Returns a list of (name, doc) tuples for the plan passed in as an arg.

    Parameters
    ----------
    RE : object
        pytest fixture object imported from `bluesky.test.conftest`
    detector_list : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of detectors
    event_type : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of 'event_types'.
    
A fixture that returns event data for a number of test cases.
Returns a list of (name, doc) tuples for the plan passed in as an arg.

Parameters


A fixture that returns event data for a number of test cases . Returns a list of ( name doc ) tuples for the plan passed in as an arg . Parameters
def generate_data(RE, detector_list, event_type):  
    def _generate_data_func(plan, skip_tests_with=None, md=None):
        if skip_tests_with is None:
            skip_tests_with = []
        if md is None:
            md = {}
        collector = []
        event_list = []
        if event_type(skip_tests_with) == 'event':
            def collect(name, doc):
                collector.append((name, doc))
                if name == 'event':
                    event_list.append(doc)
        elif event_type(skip_tests_with) == 'event_page':
            def collect(name, doc):
                if name == 'event':
                    event_list.append(doc)
                elif name == 'stop':
                    collector.append(('event_page',
                                      event_model.pack_event_page(
                                          *event_list)))
                    collector.append((name, doc))
                else:
                    collector.append((name, doc))
        elif event_type(skip_tests_with) == 'bulk_events':
            def collect(name, doc):
                if name == 'event':
                    event_list.append(doc)
                elif name == 'stop':
                    collector.append(('bulk_events', {'primary': event_list}))
                    collector.append((name, doc))
                else:
                    collector.append((name, doc))
        else:
            raise UnknownEventType('Unknown event_type kwarg passed to '
                                   'suitcase.utils.events_data')
        RE.subscribe(collect)
        RE(plan(detector_list(skip_tests_with)), md=md)
        return collector
    return _generate_data_func
def generate_data ( RE , detector_list , event_type ) : def _generate_data_func ( plan , skip_tests_with = None , md = None ) : '''Generates data to be used for testing of suitcase.*.export(..)
        functions

        Parameters
        ----------
        plan : the plan to use to generate the test data

        Returns
        -------
        collector : list
            A list of (name, doc) tuple pairs generated by the run engine.
        skip_tests_with : list, optional
            any test having request.param in this list will be skipped
        md : dict, optional
            metadata to be passed to the RunEngine
        ''' if skip_tests_with is None : skip_tests_with = [ ] if md is None : md = { } collector = [ ] event_list = [ ] if event_type ( skip_tests_with ) == 'event' : def collect ( name , doc ) : collector . append ( ( name , doc ) ) if name == 'event' : event_list . append ( doc ) elif event_type ( skip_tests_with ) == 'event_page' : def collect ( name , doc ) : if name == 'event' : event_list . append ( doc ) elif name == 'stop' : collector . append ( ( 'event_page' , event_model . pack_event_page ( * event_list ) ) ) collector . append ( ( name , doc ) ) else : collector . append ( ( name , doc ) ) elif event_type ( skip_tests_with ) == 'bulk_events' : def collect ( name , doc ) : if name == 'event' : event_list . append ( doc ) elif name == 'stop' : collector . append ( ( 'bulk_events' , { 'primary' : event_list } ) ) collector . append ( ( name , doc ) ) else : collector . append ( ( name , doc ) ) else : raise UnknownEventType ( 'Unknown event_type kwarg passed to ' 'suitcase.utils.events_data' ) RE . subscribe ( collect ) RE ( plan ( detector_list ( skip_tests_with ) ) , md = md ) return collector return _generate_data_func
A fixture that returns event data for a number of test cases.
A fixture that returns event data for a number of test cases .
# noqa '''A fixture that returns event data for a number of test cases.

    Returns a list of (name, doc) tuples for the plan passed in as an arg.

    Parameters
    ----------
    RE : object
        pytest fixture object imported from `bluesky.test.conftest`
    detector_list : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of detectors
    event_type : list
        pytest fixture defined in `suitcase.utils.conftest` which returns a
        list of 'event_types'.
    ''' '''Generates data to be used for testing of suitcase.*.export(..)
        functions

        Parameters
        ----------
        plan : the plan to use to generate the test data

        Returns
        -------
        collector : list
            A list of (name, doc) tuple pairs generated by the run engine.
        skip_tests_with : list, optional
            any test having request.param in this list will be skipped
        md : dict, optional
            metadata to be passed to the RunEngine
        ''' # define the output lists and an internal list. # define the collector function depending on the event_type # collect the documents
RE
detector_list
event_type