def test_table_editor_select_cells():
    gui = GUI()
    object_list = ObjectListWithSelection(
        values=[ListItem(value=str(i**2)) for i in range(10)]
    )
    object_list.selected_cells = [
        (object_list.values[5], 'value'),
        (object_list.values[6], 'other value'),
        (object_list.values[8], 'value'),
    ]

    with store_exceptions_on_all_threads():
        ui = object_list.edit_traits(view=select_cells_view)
        editor = ui.get_editors('values')[0]
        gui.process_events()
        if is_current_backend_qt4():
            selected = editor.selected
        elif is_current_backend_wx():
            selected = editor.selected_cells

        press_ok_button(ui)
        gui.process_events()

    assert selected == [
        (object_list.values[5], 'value'),
        (object_list.values[6], 'other value'),
        (object_list.values[8], 'value'),
    ]

def test_progress_column():
    from traitsui.extras.progress_column import ProgressColumn
    progress_view = View(
        Item(
            'values',
            show_label=False,
            editor=TableEditor(
                columns=[
                    ObjectColumn(name='value'),
                    ProgressColumn(name='other_value'),
                ],
            )
        ),
        buttons=['OK'],
    )
    gui = GUI()
    object_list = ObjectList(
        values=[ListItem(value=str(i**2)) for i in range(10)]
    )

    with store_exceptions_on_all_threads():
        ui = object_list.edit_traits(view=progress_view)
        gui.process_events()
        press_ok_button(ui)
        gui.process_events()

 def _wait_for_data(self):
     next_poll = self._handle_data()
     socks = dict(self._poller.poll(timeout=self.poll_timeout))
     if self._handshake_socket in socks and \
             socks[self._handshake_socket] == zmq.POLLIN:
         self._handle_connection()
     GUI.invoke_after(next_poll, self._wait_for_data)

    def _execute_seek(self, controller, pattern):

        duration = pattern.duration
        total_duration = pattern.total_duration

        lm = self.laser_manager
        sm = lm.stage_manager
        ld = sm.lumen_detector

        ld.mask_kind = pattern.mask_kind
        ld.custom_mask = pattern.custom_mask_radius

        osdp = sm.canvas.show_desired_position
        sm.canvas.show_desired_position = False

        st = time.time()
        self.debug('Pre seek delay {}'.format(pattern.pre_seek_delay))
        time.sleep(pattern.pre_seek_delay)

        self.debug('starting seek')
        self.debug('total duration {}'.format(total_duration))
        self.debug('dwell duration {}'.format(duration))

        if pattern.kind == 'DragonFlyPeakPattern':
            try:
                self._dragonfly_peak(st, pattern, lm, controller)
            except BaseException as e:
                self.critical('Dragonfly exception. {}'.format(e))
        else:
            self._hill_climber(st, controller, pattern)

        sm.canvas.show_desired_position = osdp

        from pyface.gui import GUI
        GUI.invoke_later(self._info.dispose)

def do_callback(dispatch, callback, *args):
    """Invoke the callback with a suitable dispatch.
    """
    if dispatch == 'ui':
        from pyface.gui import GUI
        GUI.invoke_later(callback, *args)
    else:
        callback(*args)

 def progress(self, value):
     """ Set the progress of the operation (0 <= value <= 1). """
     if self.dispatch == 'ui':
         from pyface.gui import GUI
         GUI.set_trait_later(self.promise, '_progress', value)
     else:
         with self.promise._lock:
             self.promise._progress = value

 def test_styles(self):
     # simple smoke test of buttons
     gui = GUI()
     button_text_edit = ButtonTextEdit()
     with store_exceptions_on_all_threads():
         ui = button_text_edit.edit_traits()
         self.addCleanup(ui.dispose)
         gui.process_events()

 def execute(self):
     url = 'http://' + self._host + self._url % self._tile_args
     try:
         r = requests.get(url)
         if r.status_code == 200:
             GUI.invoke_later(self.handler, self._tile_args, r.content)
     except requests.exceptions.RequestException as ex:
         print("Exception in request '{}': {}".format(self, ex))

def test_table_editor():
    gui = GUI()
    object_list = ObjectList(values=[ListItem(value=str(i ** 2)) for i in range(10)])

    with store_exceptions_on_all_threads():
        ui = object_list.edit_traits(view=simple_view)
        gui.process_events()
        press_ok_button(ui)
        gui.process_events()

 def error(self, value):
     """ Complete the deferred with failure and specified result. """
     if self.dispatch == 'ui':
         from pyface.gui import GUI
         promise = self.promise
         GUI.set_trait_later(promise, '_error', value)
         GUI.set_trait_later(promise, '_status', 'error')
     else:
         with self.promise._lock:
             self.promise._error = value
             self.promise._status = 'error'

 def _play_button_changed(self, play_pressed):
     '''
     Triggered when play button is selected
     :param play_pressed:
     :return:
     '''
     if play_pressed:
         if not self._play_thread:
             self._play_thread = True
             GUI.invoke_after(1, self._play_func)
     else:
         self._play_thread = False

    def _play_func(self):
        '''
        Called while play button is selected
        :return:
        '''
        if self._play_thread:
            if self._last_clicked_direction is not None:
                self._last_clicked_direction()
            else:
                self.move_forward()

            GUI.invoke_after(1000, self._play_func)

    def start(self):
        self._zmq_context = zmq.Context()
        self._poller = zmq.Poller()

        self._handshake_socket = self._zmq_context.socket(zmq.REP)
        self._handshake_socket.bind(self.handshake_string)
        self._data_socket = self._zmq_context.socket(zmq.SUB)
        self._data_socket.setsockopt(zmq.SUBSCRIBE, '')
        self._data_socket.connect(self.data_string)

        self._poller.register(self._handshake_socket, zmq.POLLIN)
        self._poller.register(self._data_socket, zmq.POLLIN)

        GUI.invoke_after(self.poll_period, self._wait_for_data)

def test_filtered_table_editor():
    gui = GUI()
    object_list = ObjectListWithSelection(
        values=[ListItem(value=str(i**2)) for i in range(10)]
    )

    with store_exceptions_on_all_threads():
        ui = object_list.edit_traits(view=filtered_view)
        gui.process_events()

        filter = ui.get_editors('values')[0].filter

        press_ok_button(ui)
        gui.process_events()

    assert filter is not None

 def setUp(self):
     self.gui = GUI()
     self.parent = Window()
     self.parent._create()
     self.parent.open()
     self.addCleanup(self._destroy_parent)
     self.gui.process_events()

    def check_button_text_update(self, view):
        gui = GUI()
        button_text_edit = ButtonTextEdit()

        with store_exceptions_on_all_threads():
            ui = button_text_edit.edit_traits(view=view)
            self.addCleanup(ui.dispose)

            gui.process_events()
            editor, = ui.get_editors("play_button")
            button = editor.control

            self.assertEqual(get_button_text(button), "I'm a play button")

            button_text_edit.play_button_label = "New Label"
            self.assertEqual(get_button_text(button), "New Label")

    def check_enum_text_update(self, view):
        gui = GUI()
        enum_edit = EnumModel()

        with store_exceptions_on_all_threads():
            ui = enum_edit.edit_traits(view=view)
            self.addCleanup(ui.dispose)

            gui.process_events()
            editor = ui.get_editors("value")[0]
            combobox = editor.control

            self.assertEqual(get_combobox_text(combobox), "one")

            enum_edit.value = "two"
            gui.process_events()

            self.assertEqual(get_combobox_text(combobox), "two")

def test_table_editor_select_column_index():
    gui = GUI()
    object_list = ObjectListWithSelection(
        values=[ListItem(value=str(i**2)) for i in range(10)]
    )
    object_list.selected_index = 1

    with store_exceptions_on_all_threads():
        ui = object_list.edit_traits(view=select_column_index_view)
        editor = ui.get_editors('values')[0]
        gui.process_events()
        if is_current_backend_qt4():
            selected = editor.selected_indices
        elif is_current_backend_wx():
            selected = editor.selected_column_index

        press_ok_button(ui)
        gui.process_events()

    assert selected == 1

    def setUp(self):
        from pyface.qt.QtGui import QApplication
        from pyface.ui.qt4.util.event_loop_helper import EventLoopHelper

        qt_app = QApplication.instance()
        if qt_app is None:
            qt_app = QApplication([])
        self.qt_app = qt_app

        if NativeScrollBar is None:
            raise unittest.SkipTest("Qt4 NativeScrollbar not available.")
        self.gui = GUI()
        self.event_loop_helper = EventLoopHelper(gui=self.gui, qt_app=qt_app)
        self.container = Container(position=[0, 0], bounds=[600, 600])
        self.window = Window(None, size=(600, 600), component=self.container)

    def start(self):
        """ Start the application, setting up things that are required

        Subclasses should open at least one ApplicationWindow or subclass in
        their start method, and should call the superclass start() method
        before doing any work themselves.
        """
        from pyface.gui import GUI

        ok = super(GUIApplication, self).start()
        if ok:
            # create the GUI so that the splash screen comes up first thing
            if self.gui is Undefined:
                self.gui = GUI(splash_screen=self.splash_screen)

            # create the initial windows to show
            self._create_windows()

        return ok