# Copyright (c) 2007, 2021, Oracle and/or its affiliates. # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License, version 2.0, # as published by the Free Software Foundation. # # This program is also distributed with certain software (including # but not limited to OpenSSL) that is licensed under separate terms, as # designated in a particular file or component or in included license # documentation. The authors of MySQL hereby grant you an additional # permission to link the program and your derivative works with the # separately licensed software that they have included with MySQL. # This program is distributed in the hope that it will be useful, but # WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See # the GNU General Public License, version 2.0, for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software Foundation, Inc., # 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA import os, copy, sys from mforms import newTabView, newBox, newTable, newPanel, TitledBoxPanel, newScrollPanel, newCheckBox, newTextEntry, newTextBox from mforms import newLabel, HFillFlag, VFillFlag, HExpandFlag, Utilities, newSelector from mforms import newButton, SmallHelpTextStyle, FileChooser, OpenDirectory, OpenFile, SaveFile from mforms import SelectorPopup, SelectorCombobox, Form, newImageBox import mforms import wb_admin_config_file_be from wb_common import dprint_ex, debug_level, PermissionDeniedError, InvalidPasswordError, OperationCancelledError, Users from wb_admin_config_file_be import multi_separator from workbench.utils import server_os_path from wb_admin_utils import WbAdminTabBase, WbAdminValidationConfigFile, WbAdminValidationConnection, WbAdminValidationBase, WbAdminValidationRemoteAccess #from grt.modules.WBAdmin import openRemoteFileSelector CATOPTS = os.getenv("WB_CATOPTS") if CATOPTS is not None: cat_sec = ('General', 'Advanced', 'MyISAM Parameters', 'Performance', 'Logging', 'Security', 'InnoDB Parameters', 'Networking', 'Replication') cat_grp = ('Networking', 'Advanced log options', 'Slave replication objects', 'Slave default connection values', 'Activate Logging', 'Memory', 'Fulltext search', 'Data / Memory size', 'Datafiles', 'Localization', 'Thread specific settings', 'Advanced', 'Advanced Settings', 'Various', 'Binlog Options', 'Memory usage', 'Directories', 'Logfiles', 'Relay Log', 'Master', 'General slave', 'Security', 'Activate InnoDB', 'Slave Identification', 'Query cache', 'General', 'Insert delayed settings', 'Slow query log options', 'Naming', 'Timeout Settings') def handle_cat_opt(cat, grp, enabled): print("CATOPT", "'" + enabled + "', '" + cat.get_string_value() + "', '" + grp.get_string_value()) # The list of versions to show to user when detected/given version is not supported supported_versions = [5.5, 5.6, 5.7, 8.0] #=============================================================================== def verify_selected_version(version_selector, set_back): ver = str(version_selector.get_string_value()) new_ver = "" if type(ver) is str: for vch in ver: if vch.isdigit() or vch == '.': new_ver += vch if set_back and new_ver != ver: version_selector.set_value(new_ver) return new_ver #=============================================================================== def run_version_select_form(version): form = Form(Form.main_form()) top_vbox = newBox(False) top_vbox.set_padding(16) top_vbox.set_spacing(16) info_hbox = newBox(True) info_hbox.set_spacing(16) img_box = newImageBox() img_box.set_image("warning_icon.png") right_vbox = newBox(False) right_vbox.set_spacing(12) warn_label = newLabel("Server version %s is not supported by Workbench\nconfiguration file management tool." % ".".join([str(x) for x in version])) right_vbox.add(warn_label, False, True) warn_label = newLabel("Although, you can select different server version\nfor the tool to use. Suggested version " "is given\nbelow. You can either pick version or type one." ) right_vbox.add(warn_label, False, True) warn_label = newLabel("Valid version formats are X.Y.ZZ or X.Y.\nAll other variants will resort to default - 5.1.") right_vbox.add(warn_label, False, True) if (type(version) is not tuple): version = (5,1) dprint_ex(1, "Given version is not a valid tuple object") try: version_maj = int(version[0]) + int(version[1]) / 10.0 except (ValueError, IndexError): version_maj = 5.1 guessed_version = 5.1 versions = copy.copy(supported_versions) for v in versions: if version_maj >= v: guessed_version = v else: break version_selector = newSelector(SelectorCombobox) versions.reverse() version_selector.add_items([str(x) for x in versions]) version_selector.set_value(str(guessed_version)) right_vbox.add(version_selector, False, True) version_selector.add_changed_callback(lambda: verify_selected_version(version_selector, True)) info_hbox.add(img_box, False, True) info_hbox.add(right_vbox, True, True) top_vbox.add(info_hbox, True, True) ok = newButton() ok.set_text("Ok") button_box = newBox(True) button_box.add_end(ok, False, True) top_vbox.add(button_box, False, True) form.set_content(top_vbox) form.run_modal(ok, None) guessed_version = verify_selected_version(version_selector, False) # False - do not set value back to selector if guessed_version == '' or len(guessed_version) == 1: guessed_version = "5.1" try: newver = [] splitted = guessed_version.split(".") for vpart in splitted: newver.append(int(vpart)) guessed_version = tuple(newver) except ValueError: guessed_version = (5,1) return guessed_version #=============================================================================== class Page(object): def __init__(self, page_name, page_content, var_column_width): self.page_name = page_name self.page_content = page_content self.var_column_width = var_column_width self.panel = None self.created = False self.update_cb = None def set_update_cb(self, update_cb): self.update_cb = update_cb class WbAdminValidationRequiredRemoteAdmin(WbAdminValidationBase): def __init__(self, instance_info): super().__init__("MySQL Workbench requires an SSH connection to support managing Option File remotely.") self._instance_info = instance_info def validate(self): return self._instance_info.is_local or self._instance_info.remote_admin_enabled class WbAdminConfigFileUI(WbAdminTabBase): def __init__(self, ctrl_be, instance_info, main_view, version="5.1"): WbAdminTabBase.__init__(self, ctrl_be, instance_info, main_view) self.add_validation(WbAdminValidationRemoteAccess(instance_info)) self.add_validation(WbAdminValidationConfigFile(instance_info)) self.add_validation(WbAdminValidationRequiredRemoteAdmin(instance_info)) self.search_panel = self.create_search_panel() self.set_header(self.create_standard_header("title_options.png", instance_info.name, "Options File", self.search_panel)) self.tab_view = newTabView() self.myopts = None self.opt2ctrl_map = {} self.not_multiline_options = set() self.loading = True self.section = None self.version = version self.version = version self.dir_dict = {} # Will be filled later when loading default values self.pack_to_top() self.bottom_box = newBox(True) accept_btn = newButton() accept_btn.set_text("Apply...") accept_btn.set_name("Apply") accept_btn.add_clicked_callback(self.config_apply_changes_clicked) discard_btn = newButton() discard_btn.set_text("Discard") discard_btn.add_clicked_callback(self.config_discard_changes_clicked) self.bottom_box.add(newLabel("Configuration File:"), False, True) self.bottom_box.add(self.file_name_ctrl, True, True) self.bottom_box.add(self.section_ctrl, False, True) if sys.platform.lower() == "darwin": self.bottom_box.add(accept_btn, False, True) self.bottom_box.add(discard_btn, False, True) else: self.bottom_box.add(discard_btn, False, True) self.bottom_box.add(accept_btn, False, True) self.set_footer(self.bottom_box) self.enable_ui(False) # Default is to false, in case there will be some error, # update_ui and create_ui will take care of enabling it. @classmethod def wba_register(cls, admin_context): admin_context.register_page(cls, "Instance", "Options File", "Options File", True) @classmethod def identifier(cls): return "admin_option_file" #--------------------------------------------------------------------------- def update_ui(self): on = bool(self._instance_info.admin_enabled) self.enable_ui(on) def enable_ui(self, state): self.file_name_ctrl.set_enabled(state) self.section_ctrl.set_enabled(state) self.bottom_box.set_enabled(state) self.tab_view.set_enabled(state) self.search_panel.set_enabled(state) #--------------------------------------------------------------------------- def create_search_panel(self): search_label = newLabel("Locate option:") self.option_lookup_entry = newTextEntry() self.option_lookup_entry.set_size(200,-1) self.search_btn = newButton() self.search_btn.set_text("Find") self.search_btn.set_size(70, -1) search_box = newBox(True) search_box.set_padding(2) search_box.set_spacing(4) #search_box.set_size(300, -1) search_box.add(search_label, False, True) search_box.add(self.option_lookup_entry, False, True) search_box.add(self.search_btn, False, True) search_panel = newPanel(mforms.FilledPanel) search_panel.add(search_box) def lookup_option_wrapper(): self.locate_option(self.option_lookup_entry.get_string_value()) self.search_btn.add_clicked_callback(lookup_option_wrapper) return search_panel #--------------------------------------------------------------------------- def create_page(self, page_number): self.loading = True if page_number < 0 or page_number == 0: self.loading = False return if page_number not in self.pages: print("Unknown page number ", page_number) self.loading = False return # page is a stored page data stored in self.pages in create_ui page = self.pages[page_number] if page.created == True: self.loading = False return page_content = page.page_content box = newBox(False) box.set_spacing(8) # Actual option values from config file parsed by cfg_be, in form of list of tuples (, ) options = self.cfg_be.get_options(self.section_ctrl.get_string_value()) opts_map = dict(options) # Fill the dict with dirs to be used in defaults values: self.dir_dict = {'datadir' : self._get_datadir(), 'basedir' : self._get_basedir(), 'configdir' : os.path.dirname(self.server_profile.config_file_path), } for group in page_content['groups']: controls = group['controls'] number_of_controls = len(controls) if number_of_controls == 0: continue table = newTable() table.set_row_spacing(10) table.set_column_spacing(20) table.set_padding(5) table.set_homogeneous(False) table.set_row_count(number_of_controls) table.set_column_count(3) table_row = -1 # Counter to address table rows, as we may skip some control_idx. table.suspend_layout() for control_idx in range(0, number_of_controls): ctrl_def = controls[control_idx] table_row += 1 name = ctrl_def['name'] # Handle aliases like server_id == server-id. We have only one form in # opts.opts, and that form comes from documentaion team's xml file. # However if user config file contains alias, we substitute the alias # instead of official option name to use for this WBA session. From now # on all operation with the option are done through alias. names = self.cfg_be.option_alt_names(name) #(name, name.replace("-","_"), name.replace("_","-")) right_name = [x for x in names if x in opts_map] if len(right_name) > 0: right_name = right_name[0] caption = ctrl_def.get('caption') if caption and name in caption: ctrl_def['caption'] = caption.replace(name, right_name) name = right_name ctrl_def['name'] = name ctrl_tuple = self.place_control(ctrl_def, table, table_row, page.var_column_width) label = newLabel(ctrl_def['description']) label.set_size(500, -1) label.set_wrap_text(True) label.set_style(SmallHelpTextStyle) table.add(label, 2, 3, table_row, table_row + 1, HFillFlag | VFillFlag) ctrl = ctrl_tuple[1] ctrl_def = ctrl_tuple[2] #load default value into control if ctrl is not None and ctrl_def is not None: ctrl[0].set_active(False) self.enabled_checkbox_click(name, False) if 'default' in ctrl_def: default = ctrl_def['default'] if default is not None: self.set_string_value_to_control(ctrl_tuple, str(default)) else: self.set_string_value_to_control(ctrl_tuple, "") #load control with values from config if name in opts_map: value = opts_map[name] self.enabled_checkbox_click(name, True) self.set_string_value_to_control(ctrl_tuple, value) table.resume_layout() # Remove empty rows table.set_row_count(table_row+1)#number_of_controls - (number_of_controls - table_row)) # Important for good performance: add the table *after* all content was added to its container. So we have only one layout call! panel = newPanel(TitledBoxPanel) panel.add(table) panel.set_title(group['caption']) box.add(panel, False, True) page.panel.add(box) page.created = True self.loading = False #--------------------------------------------------------------------------- def locate_option(self, option_name_fragment): for opt in self.cfg_be.get_options_containing(option_name_fragment): (tab_name, group_name) = self.cfg_be.get_option_location(opt) if tab_name is not None: for page_idx, page in self.pages.items(): if page.page_name == tab_name: self.create_page(page_idx) self.tab_view.set_active_tab(page_idx - 1) page.panel.flush_events() ctrl_tuple = self.opt2ctrl_map.get(opt) # ctrl_tuple is in for of ('', , ) if ctrl_tuple is not None: ctrl = ctrl_tuple[1] # in this ctrl we have format of (enabled_checkbox, ...) page.panel.scroll_to_view(ctrl[0]) # scroll to enabled checkbox return #--------------------------------------------------------------------------- def tab_changed(self): page_number = self.tab_view.get_active_tab() + 1 page = self.pages.get(page_number) if page is not None and page.update_cb is not None: page.update_cb(page) else: self.create_page(page_number) #--------------------------------------------------------------------------- def create_ui(self): self.loading = True self.cfg_be = wb_admin_config_file_be.WbAdminConfigFileBE(self.server_profile, self.ctrl_be) self.enable_ui(True) sys_config_path = self.server_profile.config_file_path if sys_config_path is None: sys_config_path = "" self.file_name_ctrl.set_text(sys_config_path or "not configured") self.section_ctrl.add_changed_callback(self.clear_and_load) try: self.myopts = self.cfg_be.get_possible_options() option_stats = self.cfg_be.get_option_set_stats() dprint_ex(1, "Options stats: '%s'" % str(option_stats)) if option_stats and type(option_stats) is dict: added = option_stats.get("added", None) if added is not None and added < 10: user_selected_version = run_version_select_form(option_stats["version"]) self.server_profile.set_server_version(".".join([str(x) for x in user_selected_version])) self.cfg_be.reload_possible_options() self.myopts = self.cfg_be.get_possible_options() option_stats = self.cfg_be.get_option_set_stats() dprint_ex(1, "Options stats after version correction: '%s'" % str(option_stats)) except KeyError: Utilities.show_error("Error", "Wrong version '" + self.version + "'given to admin plugin", "Close", None, None) self.load_options_from_cfg() #build ordered list of pages. Initially only skeleton pages are created, means only names. # Values into pages will be load as soon as page is switched to. self.pages = {} for page_name, page_content in self.myopts.items(): self.pages[int(page_content['position'])] = Page(page_name, page_content, page_content['width']) # False means page not created # page key is its position in UI. As we can have pages positions set like (1,2,4,5) # the position set needs to be sorted so pages appear in specified order page_positions = list(self.pages.keys()) page_positions.sort() # Create dummy pages according to assigned position for page_pos in page_positions: page = self.pages[page_pos] page.panel = newScrollPanel(mforms.ScrollPanelNoFlags) self.tab_view.add_page(page.panel, page.page_name) if debug_level > 0: # Create file page page = Page("File", None) page.panel = newTextBox(mforms.BothScrollBars) page.set_update_cb(self.update_file_content_tab) self.pages[max(self.pages.keys()) + 1] = page self.tab_view.add_page(page.panel, page.page_name) # Create first page, so we display something from start self.create_page(1) self.loading = True # create_page resets loading flag self.tab_view.add_tab_changed_callback(self.tab_changed) self.loading = False return self.tab_view #--------------------------------------------------------------------------- def update_file_content_tab(self, page): page.panel.clear() for line in self.cfg_be.get_file_content(): page.panel.append_text(line) #--------------------------------------------------------------------------- def create_textedit(self, name, ctrl_def): te = newTextEntry() te.set_enabled(False) te.add_changed_callback(lambda: self.control_action(name)) te.set_tooltip("To convert option to a multi-line one, use " + multi_separator + " to separate values. The symbol " + multi_separator + " should not be at the start or end of the value") return te #--------------------------------------------------------------------------- def create_dir_file_edit(self, name, ctrl_def): dir_box = newBox(True) dir_box.set_spacing(4) te = newTextEntry() te.set_tooltip("To convert option to a multi-line one, use " + multi_separator + " to separate values. The symbol " + multi_separator + " should not be at the start or end of the value") btn = newButton() btn.set_text("...") btn.enable_internal_padding(False) btn.add_clicked_callback(lambda: self.open_file_chooser(OpenDirectory, te, name)) dir_box.add(te, True, True) dir_box.add(btn, False, True) te.set_enabled(False) btn.set_enabled(False) return (dir_box, te, btn) #--------------------------------------------------------------------------- def _get_datadir(self): if self.server_profile.datadir: return self.server_profile.datadir elif self.cfg_be.original_opts.get('datadir'): return self.cfg_be.original_opts['datadir'].val(0) return '' #--------------------------------------------------------------------------- def _get_basedir(self): if self.server_profile.basedir: return self.server_profile.basedir elif self.cfg_be.original_opts.get('basedir'): return self.cfg_be.original_opts['basedir'].val(0) return '' #--------------------------------------------------------------------------- def create_fileedit(self, name, ctrl_def): def download_file_from_server_cb(te): ospath = server_os_path(self.server_profile) remote_path = te.get_string_value().strip('" ') if not remote_path: res = mforms.Utilities.show_error('Specify remote location', 'No remote location available. Please specify the path to the file in the server using ' 'the line edit control at the left of the download button and try again. Or you can click ' 'on the "Use Suggested" button to let Workbench use an appropriate value for the remote ' 'location', 'OK', 'Use Suggested', '') if res == mforms.ResultCancel: # Use ctrl_def default if available before attempting to build a default value: if ctrl_def.get('default'): remote_path = ctrl_def['default'] % self.dir_dict else: remote_path = self._get_datadir() if remote_path: remote_path = ospath.abspath(ospath.join(remote_path, '%s.txt' % name)) else: mforms.Utilities.show_warning('Could not find server data directory', 'Please type the full path to the file in the server before attempting to upload it', 'OK', '', '') return te.set_value(remote_path) else: return elif not ospath.isabs(remote_path): # file is relative to datadir dirname = self._get_datadir() if dirname: remote_path = ospath.abspath(ospath.join(dirname, remote_path)) file_exists = False # Check if remote_path exists in the server before attempting to download it: as_user, user_password = Users.CURRENT, None try: file_exists = self.ctrl_be.server_helper.file_exists(remote_path) except Exception: # Try again as admin user_password = None while True: try: user_password = self.ctrl_be.password_handler.get_password_for("file") if user_password is not None: as_user = Users.ADMIN file_exists = self.ctrl_be.server_helper.file_exists(remote_path, as_user, user_password) except InvalidPasswordError as e: # Wrong pwd, retry: self.ctrl_be.password_handler.reset_password_for("file") pass except OperationCancelledError: # The user chose Cancel in pwd input dialog return else: break if not file_exists: mforms.Utilities.show_error('File does not exist', 'The file "%s" does not exist in %s' % (remote_path, self.server_profile.host_name), 'OK', '', '') return filechooser = FileChooser(SaveFile) filechooser.set_directory(os.path.expanduser('~')) if filechooser.run_modal(): content = None try: content = self.ctrl_be.server_helper.get_file_content(remote_path, as_user, user_password) except PermissionDeniedError as e: try: user_password = self.ctrl_be.password_handler.get_password_for("file") except Exception as e: return if user_password is not None: as_user = Users.ADMIN try: content = self.ctrl_be.server_helper.get_file_content(remote_path, as_user, user_password) except InvalidPasswordError as err: self.ctrl_be.password_handler.reset_password_for("file") raise err except Exception as e: mforms.Utilities.show_error('Error while reading from "%s"' % remote_path, str(e), 'OK', '', '') return if content is not None: local_path = filechooser.get_path() try: local_file = open(local_path, 'w') except IOError as e: mforms.Utilities.show_error('Error while opening "%s" for writing' % local_path, str(e), 'OK', '', '') return local_file.write(content) mforms.App.get().set_status_text('File %s in %s successfully downloaded to %s' % (remote_path, self.server_profile.host_name, local_path) ) def upload_file_to_server_cb(te): ospath = server_os_path(self.server_profile) remote_path = te.get_string_value().strip('" ') if not remote_path: res = mforms.Utilities.show_error('Specify remote location', 'No remote location available. Please specify the path to the file in the server using ' 'the line edit control at the left of the upload button and try again. Or you can click ' 'on the "Use Suggested" button to let Workbench use an appropriate value for the remote ' 'location', 'OK', 'Use Suggested', '') if res == mforms.ResultCancel: # Use ctrl_def default if available before attempting to build a default value: if ctrl_def.get('default'): remote_path = ctrl_def['default'] % self.dir_dict else: remote_path = self._get_datadir() if remote_path: remote_path = ospath.abspath(ospath.join(remote_path, '%s.txt' % name)) te.set_value(remote_path) else: mforms.Utilities.show_warning('Could not find server data directory', 'Please type the full path to the file in the server before attempting to upload it', 'OK', '', '') return te.set_value(remote_path) else: return elif not ospath.isabs(remote_path): # file is relative to datadir dirname = self._get_datadir() if dirname: remote_path = ospath.abspath(ospath.join(dirname, remote_path)) filechooser = FileChooser(OpenFile) filechooser.set_directory(os.path.expanduser('~')) if filechooser.run_modal(): local_path = filechooser.get_path() try: local_file = open(local_path) except IOError as e: mforms.Utilities.show_error('Error while opening "%s"' % local_path, str(e), 'OK', '', '') return try: content = local_file.read() self.ctrl_be.server_helper.set_file_content(remote_path, content, as_user=Users.CURRENT, user_password=None) except PermissionDeniedError as e: try: user_password = self.ctrl_be.password_handler.get_password_for("file") except Exception as e: return if user_password is not None: as_user = Users.ADMIN try: self.ctrl_be.server_helper.set_file_content(remote_path, content, as_user, user_password) except InvalidPasswordError as err: self.ctrl_be.password_handler.reset_password_for("file") raise err except Exception as e: mforms.Utilities.show_error('Error while uploading the file to "%s"' % remote_path, str(e), 'OK', '', '') return mforms.App.get().set_status_text('File %s successfully uploaded to %s as %s' % (local_path, self.server_profile.host_name, remote_path) ) dir_box = newBox(True) dir_box.set_spacing(4) te = newTextEntry() te.set_tooltip('The path to the file in your server') dir_box.add(te, True, True) te.set_enabled(False) btn = newButton() btn.set_text("...") btn.set_tooltip('Browse a file in your server...') btn.enable_internal_padding(False) btn.add_clicked_callback(lambda: self.open_file_chooser(OpenFile, te, name)) dir_box.add(btn, False, True) btn_dwn = None if ctrl_def['type'] in ('filedownload', 'fileedit'): btn_dwn = newButton(mforms.ToolButton) btn_dwn.set_icon(mforms.App.get().get_resource_path('record_export.png')) btn_dwn.set_tooltip('Download this file to your local computer') btn_dwn.enable_internal_padding(False) btn_dwn.add_clicked_callback(lambda: download_file_from_server_cb(te)) dir_box.add(btn_dwn, False, True) btn_dwn.set_enabled(False) btn_upl = None if ctrl_def['type'] == 'fileedit': btn_upl = newButton(mforms.ToolButton) btn_upl.set_icon(mforms.App.get().get_resource_path('record_import.png')) btn_upl.set_tooltip('Upload a file from your local computer to the server') btn_upl.enable_internal_padding(False) btn_upl.add_clicked_callback(lambda: upload_file_to_server_cb(te)) dir_box.add(btn_upl, False, True) btn_upl.set_enabled(False) return (dir_box, te, btn, btn_dwn, btn_upl) #--------------------------------------------------------------------------- def create_numeric(self, name, ctrl_def): te = newTextEntry() #spin_box.add(te, True, True) te.set_enabled(False) te.set_tooltip("For numeric values, you may specify a K, M or G size suffix, if appropriate.") te.add_changed_callback(lambda: self.control_action(name)) #unit = None #if ctrl_def.has_key('unitcontrol'): # unit = ctrl_def['unitcontrol'] #unitcontrol = None #unit_items = None #if unit is not None: # unitcontrol = newSelector() # unit_items = unit.split(";") # for item in unit_items: # unitcontrol.add_item(item) # # spin_box.add(unitcontrol, False, False) # unitcontrol.set_enabled(False) # unitcontrol.add_changed_callback(lambda: self.control_action(name)) #return (spin_box, te, unitcontrol, unit_items) return te #--------------------------------------------------------------------------- def create_dropdownbox(self, name, ctrl_def, ctype): items = None if 'choice' in ctrl_def: items = ctrl_def['choice'] style = SelectorPopup if ctype == 'dropdownboxentry': style = SelectorCombobox dropbox = newSelector(style) if type(items) is str: if items in wb_admin_config_file_be.pysource: code = wb_admin_config_file_be.pysource[items] result = eval(code) items = [] for item in result.split(','): item = item.strip(" \t") items.append(item) for i in items: dropbox.add_item(i) dropbox.set_enabled(False) if 'default' in ctrl_def: default = ctrl_def['default'] idx = 0 for i in items: if i == default: dropbox.set_selected(idx) idx += 1 dropbox.add_changed_callback(lambda: self.control_action(name)) return (dropbox, items) #--------------------------------------------------------------------------- def place_control(self, ctrl_def, table, row, val_column_width): ctrl = None ctype = ctrl_def['type'] name = ctrl_def['name'] enabled = newCheckBox() enabled.set_text(ctrl_def['caption']) enabled.set_size(val_column_width, -1) # Use a fixed fix to make all tables align their columns properly. Must be larger than the largest text, to make it work. enabled.set_tooltip("%s\n\n%s" % (ctrl_def['name'], ctrl_def['description'])) # place_control creates control as ctrl_def describes. Reference to a created control is placed # to map of controls. That is done in order to access controls via option name if ctype == "checkbox" or ctype == "boolean": ctrl = ('chk', (enabled, enabled), ctrl_def) self.opt2ctrl_map[name] = ctrl enabled.set_active(False) #label = newLabel(" ") #table.add(label, 1, 2, row, row+1, HExpandFlag | HFillFlag) elif ctype == 'textedit' or ctype == 'string' or ctype == 'set': te = self.create_textedit(name, ctrl_def) table.add(te, 1, 2, row, row+1, VFillFlag | HExpandFlag | HFillFlag) ctrl = ('txt', (enabled, te), ctrl_def) self.opt2ctrl_map[name] = ctrl elif ctype in ['filename', 'fileedit', 'filebrowse', 'filedownload']: if name.endswith('dir'): (dir_box, te, btn) = self.create_dir_file_edit(name, ctrl_def) ctrl = ('dir', (enabled, te, btn), ctrl_def) else: (dir_box, te, btn, btn_dwn, btn_upl) = self.create_fileedit(name, ctrl_def) ctrl = ('fed', (enabled, te, btn, btn_dwn, btn_upl), ctrl_def) table.add(dir_box, 1, 2, row, row + 1, VFillFlag | HExpandFlag | HFillFlag) te.add_changed_callback(lambda: self.control_action(name)) self.opt2ctrl_map[name] = ctrl elif ctype in ['directory', 'dirname']: (dir_box, te, btn) = self.create_dir_file_edit(name, ctrl_def) ctrl = ('dir', (enabled, te, btn), ctrl_def) table.add(dir_box, 1, 2, row, row + 1, VFillFlag | HExpandFlag | HFillFlag) te.add_changed_callback(lambda: self.control_action(name)) self.opt2ctrl_map[name] = ctrl elif ctype == "numeric" or ctype == "spinedit" or ctype == "integer": #(spin_box, te, unitcontrol, unit_items) = self.create_numeric(name, ctrl_def) #ctrl = ('spn', (enabled, te, unitcontrol, unit_items), ctrl_def) te = self.create_numeric(name, ctrl_def) #ctrl = ('spn', (enabled, te), ctrl_def) ctrl = ('txt', (enabled, te), ctrl_def) self.opt2ctrl_map[name] = ctrl #table.add(spin_box, 1, 2, row, row + 1, HExpandFlag | HFillFlag) table.add(te, 1, 2, row, row + 1, VFillFlag | HExpandFlag | HFillFlag) elif ctype == "dropdownbox" or ctype == 'dropdownboxentry' or ctype == "enum": if 'choice' not in ctrl_def: te = newTextEntry() te.set_enabled(False) te.add_changed_callback(lambda: self.control_action(name)) table.add(te, 1, 2, row, row+1, VFillFlag | HExpandFlag | HFillFlag) ctrl = ('txt', (enabled, te), ctrl_def) self.opt2ctrl_map[name] = ctrl else: (dropbox, items) = self.create_dropdownbox(name, ctrl_def, ctype) table.add(dropbox, 1, 2, row, row + 1, VFillFlag | HExpandFlag | HFillFlag) ctrl = ('drp', (enabled, dropbox, items), ctrl_def) self.opt2ctrl_map[name] = ctrl else: raise NotImplementedError("Control type %s not implemented for config file editor"%ctype) if CATOPTS is None: table.add(enabled, 0, 1, row, row + 1, VFillFlag | HFillFlag) enabled.add_clicked_callback(lambda: self.enabled_checkbox_click(name)) else: catbox = newBox(True) cat = newSelector(mforms.SelectorCombobox) for item in cat_sec: cat.add_item(item) grp = newSelector(mforms.SelectorCombobox) for item in cat_grp: grp.add_item(item) cat.add_changed_callback(lambda : handle_cat_opt(cat, grp, ctrl_def['name'])) grp.add_changed_callback(lambda : handle_cat_opt(cat, grp, ctrl_def['name'])) catbox.add(cat, True, True) catbox.add(grp, True, True) catbox.add(enabled, True, True) table.add(catbox, 0, 1, row, row + 1, HExpandFlag | HFillFlag) return ctrl #--------------------------------------------------------------------------- def open_file_chooser(self, file_chooser_type, textfield, name): filename = None if self.server_profile.is_local: filechooser = FileChooser(file_chooser_type) filechooser.set_directory(textfield.get_string_value()) if filechooser.run_modal(): filename = filechooser.get_directory() if file_chooser_type is OpenDirectory else filechooser.get_path() else: from wba_ssh_ui import remote_file_selector filename = remote_file_selector(self.server_profile, self.ctrl_be.password_handler, self.ctrl_be.ssh) if filename and (type(filename) is str or type(filename) is str): filename = filename.replace("\\", "/") # TODO: Check for backslashed spaces and so on textfield.set_value('"' + filename + '"') self.control_action(name) #--------------------------------------------------------------------------- def enabled_checkbox_click(self, name, force_enabled = None): if name in self.opt2ctrl_map: ctrl = self.opt2ctrl_map[name] def control(idx): return ctrl[1][idx] tag = ctrl[0] # tuple ctrl holds tag at index 0, the rest is control def. Exact format # of control tuple(the one that goes after tag is defined by the type of control if force_enabled is not None: enabled = bool(force_enabled) control(0).set_active(enabled) else: enabled = control(0).get_active() if tag == "txt": control(1).set_enabled(enabled) elif tag == "spn": control(1).set_enabled(enabled) if control(2) is not None: control(2).set_enabled(enabled) elif tag == "drp": control(1).set_enabled(enabled) elif tag == "dir": control(1).set_enabled(enabled) control(2).set_enabled(enabled) elif tag == "fed": control(1).set_enabled(enabled) control(2).set_enabled(enabled) try: control(3).set_enabled(enabled) control(4).set_enabled(enabled) except Exception: pass if not self.loading: # Notify config BE about change if enabled: self.cfg_be.option_added(name, self.get_string_value_from_control(name, ctrl), self.section) else: self.cfg_be.option_removed(name, self.section) #--------------------------------------------------------------------------- def control_action(self, name): if self.loading: return if name in self.opt2ctrl_map: ctrl = self.opt2ctrl_map[name] if not self.loading: self.cfg_be.option_changed(name, self.get_string_value_from_control(name, ctrl), self.section) #--------------------------------------------------------------------------- def get_string_value_from_control(self, option_name, ctrl): value = "" # ctrl is a tuple from map tag = ctrl[0] def control(idx): return ctrl[1][idx] is_multiple = False control_name = control(1).getInternalName() if control_name == "Multiple": is_multiple = True if tag == "txt": value = (control(1).get_string_value(),) elif tag == "spn": # (enabled, te, unitcontrol, unit_items)). Note! unitcontrol and unit_items may be None value = control(1).get_string_value().strip(" \r\n\t") if control(2) is not None: value += control(2).get_string_value() elif tag == "drp": value = control(1).get_string_value() elif tag == "dir": value = control(1).get_string_value() if is_multiple: value = value.split(multi_separator) elif tag == "fed": value = control(1).get_string_value() if is_multiple: value = value.split(multi_separator) elif tag == "chk": value = (control(0).get_active(),) # Here we detect if value has signs of multi line option. # For example, user entered separator char. # It is enough to ensure that the first item in tuple is string is_string = False # We only can detect multi-line in strings (rework is scheduled for 5.3) has_separator = False if type(value) is tuple: value_len = len(value) if value_len == 1: # Check only single item tuples is_string = type(value[0]) is str or type(value[0]) is str if is_string: has_separator = value[0].find(multi_separator) > 0 else: is_string = type(value) is str or type(value) is str has_separator = value.find(multi_separator) > 0 if is_multiple == False and is_string and has_separator and not self.loading and option_name not in self.not_multiline_options: answer = Utilities.show_message("Multi Line Option" ,"The %s sequence was detected in the string, which is used for separating multi-line options. Would you like to split the value to multiple options?" % multi_separator , "Convert", "Keep as Is", "") if answer == mforms.ResultOk: control(1).set_name("Multiple") else: self.not_multiline_options.add(option_name) # some controls return values in form of one-item tuples # so we need to extract that item for processing below if has_separator and is_string and not option_name in self.not_multiline_options: if type(value) is tuple: if len(value) == 1: # Only extract values from one-item tuples value = value[0] # If tuple has more items it already has been converted to multi-line # skip multi line values - no need to convert. Also skip non-string option values if type(value) is not tuple: value = [x.strip(multi_separator) for x in value.split(multi_separator)] if type(value) is not list and type(value) is not tuple: value = (value,) return value #--------------------------------------------------------------------------- #Value is a an integer with def parse_spin_string(self, unit_items, value): value = value.strip(" \r\t\n") longest_suffix = 0 if unit_items is not None: for item in unit_items: l = len(item) if (l > longest_suffix): longest_suffix = l suffix = "" def get_unit(sfx): sfx2 = sfx.lower() ret_item = None for item in unit_items: if sfx2 == item.lower(): ret_item = item break return ret_item if longest_suffix > 0: value_len = len(value) for suffix_length in range(1,longest_suffix + 1): if suffix_length < value_len: cur_sfx = value[-suffix_length:] cur_unit = get_unit(cur_sfx) if cur_unit is not None: suffix = cur_unit value = value[:-suffix_length] return ([x for x in value if x.isdigit() or x == '-'], suffix) #--------------------------------------------------------------------------- def set_string_value_to_control(self, ctrl, value): tag = ctrl[0] #ctrl is a tuple from map def control(idx): return ctrl[1][idx] if type(value) is tuple and len(value) > 1: control(1).set_name("Multiple") else: control(1).set_name("Single") if tag in ['txt', 'dir', 'fed']: if value is None or not isinstance(value, str): # TODO: Add config file error message value = "" value = value.strip(" \r\n\t") try: value = value.format(**self.dir_dict) except Exception: pass control(1).set_value(value) elif tag == "spn": if value is None: value = "" elif type(value) is not str: # TODO: Add Warning pass value = value.strip(" \r\n\t") (value,suffix) = self.parse_spin_string(control(3), value) control(1).set_value(value) if control(2) is not None and suffix is not None: try: idx = control(3).index(suffix) control(2).set_selected(idx) except ValueError: pass elif tag == "drp": #search for value try: items = control(2) if items is not None: lowcase_value = value.lower() for (i, item) in enumerate(items): if item.lower() == lowcase_value: control(1).set_selected(i) except ValueError: pass elif tag == "chk": value = self.cfg_be.normalize_bool(value) control(1).set_active(value) else: print("Error") #--------------------------------------------------------------------------- def config_apply_changes_clicked(self): errors = self.cfg_be.validate_changes(self.myopts) if errors: mforms.Utilities.show_warning("Configuration Error", "The following errors were found in the changes you have made.\n" "Please correct them before applying:\n"+errors, "OK", "", "") else: self.cfg_be.apply_changes() #--------------------------------------------------------------------------- def config_discard_changes_clicked(self): self.clear_and_load() #--------------------------------------------------------------------------- def clear_and_load(self): self.not_multiline_options = set() if self.loading == False: self.load_defaults() self.load_options_from_cfg(self.section_ctrl.get_string_value()) #--------------------------------------------------------------------------- def load_options_from_cfg(self, given_section = None): self.loading = True try: self.cfg_be.open_configuration_file(self.server_profile.config_file_path or "") except Exception as exc: import traceback traceback.print_exc() mforms.Utilities.show_error("Error opening configuration file", "%s: %s" % (type(exc).__name__, exc), "OK", "", "") self.section_ctrl.clear() idx = 0 if given_section is None or given_section == "": given_section = self.server_profile.config_file_section # Fill section selector at the bottom of config file page section_ctrl_was_filled = False for i,sec in enumerate(self.cfg_be.get_sections()): self.section_ctrl.add_item(sec) section_ctrl_was_filled = True if sec == given_section: idx = i self.section = sec # If we have an empty file or file with no section, add user-specified default section to the selector if section_ctrl_was_filled == False and given_section is not None: self.section_ctrl.add_item(given_section) idx = 0 self.section_ctrl.set_selected(idx) # each opt is (name, value) for name, value in self.cfg_be.get_options(self.section): ctrl = self.opt2ctrl_map.get(name) if not ctrl: # We are here because of either 1) alias, or 2) unknown option, or # 3) initial load when map is not created # Try to resolve alias and update UI and maps names = self.cfg_be.option_alt_names(name) for alt_name in names: ctrl = self.opt2ctrl_map.get(alt_name) if ctrl is not None: # Fix UI and map self.opt2ctrl_map[name] = ctrl # Rename checkbox with name ctrl[1][0].set_text(name) del self.opt2ctrl_map[alt_name] break if ctrl is None: continue self.enabled_checkbox_click(name, True) self.set_string_value_to_control(ctrl, value) self.loading = False #--------------------------------------------------------------------------- def load_defaults(self): self.loading = True for name,ctrl_tuple in self.opt2ctrl_map.items(): if ctrl_tuple is not None: #tag = ctrl_tupple[0] ctrl = ctrl_tuple[1] ctrl_def = ctrl_tuple[2] if ctrl is not None and ctrl_def is not None: ctrl[0].set_active(False) self.enabled_checkbox_click(ctrl_def['name'], False) if 'default' in ctrl_def: default = ctrl_def['default'] if default is not None: self.set_string_value_to_control(ctrl_tuple, str(default)) else: self.set_string_value_to_control(ctrl_tuple, "") self.loading = False #------------------------------------------------------------------------------- def pack_to_top(self): self.file_name_ctrl = newLabel("") self.file_name_ctrl.set_tooltip("To change the path to the configuration file, edit the server profile in the Manage Server Instances dialog.") sys_config_path = self.server_profile.config_file_path if sys_config_path is None: sys_config_path = "" self.file_name_ctrl.set_text(sys_config_path) self.file_name_ctrl.set_size(200, -1) self.section_ctrl = newSelector() self.section_ctrl.set_size(100, -1) #------------------------------------------------------------------------------- def shutdown(self): if hasattr(self, "cfg_be") and hasattr(self.cfg_be, 'changeset') and self.cfg_be.changeset: res = Utilities.show_warning('Configuration file', '''Configuration file from instance '%s' may contain unsaved changes. Would you like to review them?''' % self.server_profile.name, 'Review', 'Cancel', 'Discard') if res == mforms.ResultOk: self.config_apply_changes_clicked() return res == mforms.ResultOther