diff --git a/src/main-window.ui b/src/main-window.ui index 1ababe8..45b854d 100644 --- a/src/main-window.ui +++ b/src/main-window.ui @@ -48,6 +48,16 @@ + + + folder-download-symbolic + Import from WSDL + + + + diff --git a/src/meson.build b/src/meson.build index e6b6d16..451b701 100644 --- a/src/meson.build +++ b/src/meson.build @@ -44,6 +44,8 @@ roster_sources = [ 'preferences_dialog.py', 'request_tab_widget.py', 'script_executor.py', + 'wsdl_importer.py', + 'wsdl_import_dialog.py', ] install_data(roster_sources, install_dir: moduledir) diff --git a/src/window.py b/src/window.py index 6da3077..715ffcf 100644 --- a/src/window.py +++ b/src/window.py @@ -36,6 +36,7 @@ from .project_manager import ProjectManager from .tab_manager import TabManager from .icon_picker_dialog import IconPickerDialog from .environments_dialog import EnvironmentsDialog +from .wsdl_import_dialog import WsdlImportDialog from .request_tab_widget import RequestTabWidget from .widgets.history_item import HistoryItem from .widgets.project_item import ProjectItem @@ -67,6 +68,8 @@ class RosterWindow(Adw.ApplicationWindow): projects_listbox = Gtk.Template.Child() add_project_button = Gtk.Template.Child() + import_wsdl_button = Gtk.Template.Child() + # History (hidden but kept for compatibility) history_listbox = Gtk.Template.Child() @@ -1410,6 +1413,26 @@ class RosterWindow(Adw.ApplicationWindow): widget.original_request = None widget.modified = True + @Gtk.Template.Callback() + def on_import_wsdl_clicked(self, button): + """Open the WSDL import dialog.""" + projects = self.project_manager.load_projects() + dialog = WsdlImportDialog(self.project_manager, projects) + dialog.connect('import-completed', self._on_wsdl_import_completed) + dialog.present(self) + + def _on_wsdl_import_completed(self, dialog, project_id: str): + """Refresh sidebar after a successful WSDL import.""" + self._load_projects() + projects = self.project_manager.load_projects() + count = 0 + for p in projects: + if p.id == project_id: + count = len(p.requests) + name = p.name + break + self._show_toast(f"Imported {count} operation(s) into '{name}'") + def _on_delete_request(self, widget, saved_request, project): """Delete saved request with confirmation.""" dialog = Adw.AlertDialog() diff --git a/src/wsdl_import_dialog.py b/src/wsdl_import_dialog.py new file mode 100644 index 0000000..5abf0da --- /dev/null +++ b/src/wsdl_import_dialog.py @@ -0,0 +1,354 @@ +# wsdl_import_dialog.py +# +# Copyright 2025 Pavel Baksy +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# 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 for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: GPL-3.0-or-later + +import gi +gi.require_version('Soup', '3.0') +from gi.repository import Adw, Gtk, GObject, GLib, Soup +from typing import List + +from .wsdl_importer import parse_wsdl, build_http_request, WsdlParseResult, WsdlOperation + + +class WsdlImportDialog(Adw.Dialog): + """Two-step dialog: fetch WSDL URL → select operations → import.""" + + __gtype_name__ = 'WsdlImportDialog' + + __gsignals__ = { + # Emitted after a successful import; carries the target project_id + 'import-completed': (GObject.SIGNAL_RUN_FIRST, None, (str,)), + } + + def __init__(self, project_manager, existing_projects): + super().__init__() + self.set_title("Import from WSDL") + self.set_content_width(560) + self.set_content_height(540) + + self._pm = project_manager + self._existing_projects = list(existing_projects) + self._wsdl_result: WsdlParseResult | None = None + self._op_checkboxes: List[tuple[WsdlOperation, Gtk.CheckButton]] = [] + self._soup = Soup.Session.new() + + self._build_ui() + + # ------------------------------------------------------------------ build + + def _build_ui(self): + tv = Adw.ToolbarView() + tv.add_top_bar(Adw.HeaderBar()) + + self._stack = Gtk.Stack() + self._stack.set_transition_type(Gtk.StackTransitionType.SLIDE_LEFT_RIGHT) + self._stack.set_transition_duration(200) + self._stack.add_named(self._make_url_page(), "url") + self._stack.add_named(self._make_ops_page(), "ops") + tv.set_content(self._stack) + + ab = Gtk.ActionBar() + + self._back_btn = Gtk.Button(label="Back") + self._back_btn.connect("clicked", lambda _: self._show_url_page()) + ab.pack_start(self._back_btn) + + self._fetch_btn = Gtk.Button(label="Fetch WSDL") + self._fetch_btn.add_css_class("suggested-action") + self._fetch_btn.connect("clicked", self._on_fetch) + ab.pack_end(self._fetch_btn) + + self._import_btn = Gtk.Button(label="Import") + self._import_btn.add_css_class("suggested-action") + self._import_btn.connect("clicked", self._on_import) + ab.pack_end(self._import_btn) + + tv.add_bottom_bar(ab) + self.set_child(tv) + self._show_url_page() + + def _make_url_page(self) -> Gtk.Widget: + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=16) + box.set_valign(Gtk.Align.CENTER) + box.set_vexpand(True) + box.set_margin_top(32) + box.set_margin_bottom(24) + box.set_margin_start(24) + box.set_margin_end(24) + + icon = Gtk.Image.new_from_icon_name("network-server-symbolic") + icon.set_pixel_size(64) + icon.add_css_class("dim-label") + box.append(icon) + + title = Gtk.Label(label="Import from WSDL") + title.add_css_class("title-2") + box.append(title) + + subtitle = Gtk.Label(label="Enter the URL of a WSDL document to discover SOAP operations") + subtitle.add_css_class("dim-label") + subtitle.set_wrap(True) + subtitle.set_justify(Gtk.Justification.CENTER) + box.append(subtitle) + + grp = Adw.PreferencesGroup() + self._url_row = Adw.EntryRow() + self._url_row.set_title("WSDL URL") + self._url_row.set_input_purpose(Gtk.InputPurpose.URL) + self._url_row.connect("entry-activated", lambda _: self._on_fetch(None)) + grp.add(self._url_row) + box.append(grp) + + self._err_label = Gtk.Label() + self._err_label.add_css_class("error") + self._err_label.set_wrap(True) + self._err_label.set_visible(False) + box.append(self._err_label) + + self._spinner = Gtk.Spinner() + self._spinner.set_size_request(32, 32) + self._spinner.set_halign(Gtk.Align.CENTER) + self._spinner.set_visible(False) + box.append(self._spinner) + + return box + + def _make_ops_page(self) -> Gtk.Widget: + outer = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=12) + outer.set_margin_top(16) + outer.set_margin_bottom(16) + outer.set_margin_start(16) + outer.set_margin_end(16) + + # Service info summary + self._svc_group = Adw.PreferencesGroup() + self._svc_group.set_title("Service") + self._svc_name_row = Adw.ActionRow() + self._svc_name_row.set_title("Name") + self._svc_group.add(self._svc_name_row) + self._svc_url_row = Adw.ActionRow() + self._svc_url_row.set_title("Endpoint") + self._svc_group.add(self._svc_url_row) + outer.append(self._svc_group) + + # Operations header row + ops_hdr = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=8) + ops_hdr.set_margin_top(4) + ops_lbl = Gtk.Label(label="Operations") + ops_lbl.add_css_class("title-4") + ops_lbl.set_hexpand(True) + ops_lbl.set_xalign(0) + ops_hdr.append(ops_lbl) + self._sel_all_btn = Gtk.Button(label="Select All") + self._sel_all_btn.add_css_class("flat") + self._sel_all_btn.connect("clicked", self._on_select_all) + ops_hdr.append(self._sel_all_btn) + outer.append(ops_hdr) + + # Scrollable operations list + scroll = Gtk.ScrolledWindow() + scroll.set_vexpand(True) + scroll.set_min_content_height(100) + self._ops_list = Gtk.ListBox() + self._ops_list.add_css_class("boxed-list") + self._ops_list.set_selection_mode(Gtk.SelectionMode.NONE) + scroll.set_child(self._ops_list) + outer.append(scroll) + + # Import target + tgt_grp = Adw.PreferencesGroup() + tgt_grp.set_title("Import to") + tgt_grp.set_margin_top(4) + + new_row = Adw.ActionRow() + new_row.set_title("Create new project") + self._new_radio = Gtk.CheckButton() + self._new_radio.set_active(True) + new_row.add_prefix(self._new_radio) + new_row.set_activatable_widget(self._new_radio) + tgt_grp.add(new_row) + + self._new_name_row = Adw.EntryRow() + self._new_name_row.set_title("Project name") + tgt_grp.add(self._new_name_row) + + if self._existing_projects: + exist_row = Adw.ActionRow() + exist_row.set_title("Add to existing project") + self._exist_radio = Gtk.CheckButton() + self._exist_radio.set_group(self._new_radio) + exist_row.add_prefix(self._exist_radio) + exist_row.set_activatable_widget(self._exist_radio) + tgt_grp.add(exist_row) + + self._proj_combo = Adw.ComboRow() + self._proj_combo.set_title("Project") + self._proj_combo.set_model(Gtk.StringList.new([p.name for p in self._existing_projects])) + self._proj_combo.set_sensitive(False) + tgt_grp.add(self._proj_combo) + + self._new_radio.connect("toggled", self._on_target_toggled) + else: + self._exist_radio = None + self._proj_combo = None + + outer.append(tgt_grp) + return outer + + # ----------------------------------------------------------- page control + + def _show_url_page(self): + self._stack.set_visible_child_name("url") + self._back_btn.set_visible(False) + self._fetch_btn.set_visible(True) + self._import_btn.set_visible(False) + + def _show_ops_page(self): + self._stack.set_visible_child_name("ops") + self._back_btn.set_visible(True) + self._fetch_btn.set_visible(False) + self._import_btn.set_visible(True) + + # --------------------------------------------------------- event handlers + + def _on_target_toggled(self, _radio): + is_new = self._new_radio.get_active() + self._new_name_row.set_sensitive(is_new) + if self._proj_combo: + self._proj_combo.set_sensitive(not is_new) + + def _on_select_all(self, _btn): + all_on = all(cb.get_active() for _, cb in self._op_checkboxes) + for _, cb in self._op_checkboxes: + cb.set_active(not all_on) + self._sel_all_btn.set_label("Deselect All" if not all_on else "Select All") + + def _on_fetch(self, _btn): + url = self._url_row.get_text().strip() + if not url: + self._set_err("Please enter a WSDL URL") + return + if not url.startswith(('http://', 'https://')): + self._set_err("URL must start with http:// or https://") + return + + self._err_label.set_visible(False) + self._spinner.set_visible(True) + self._spinner.start() + self._fetch_btn.set_sensitive(False) + + try: + msg = Soup.Message.new("GET", url) + except Exception as e: + self._fetch_failed(f"Invalid URL: {e}") + return + + msg.get_request_headers().append("Accept", "text/xml, application/xml, */*") + self._soup.send_and_read_async(msg, GLib.PRIORITY_DEFAULT, None, + self._on_downloaded, msg) + + def _on_downloaded(self, session, async_result, msg): + self._spinner.stop() + self._spinner.set_visible(False) + self._fetch_btn.set_sensitive(True) + + try: + bytes_data = session.send_and_read_finish(async_result) + except GLib.Error as e: + self._set_err(f"Download failed: {e.message}") + return + + status = msg.get_status() + if not (200 <= status < 300): + self._set_err(f"Server returned HTTP {status}") + return + + xml = bytes_data.get_data().decode('utf-8', errors='replace') + parsed = parse_wsdl(xml) + + if parsed.error: + self._set_err(parsed.error) + return + + self._wsdl_result = parsed + self._populate_ops_page(parsed) + self._show_ops_page() + + def _on_import(self, _btn): + if not self._wsdl_result: + return + + selected = [op for op, cb in self._op_checkboxes if cb.get_active()] + if not selected: + return + + if self._new_radio.get_active(): + name = self._new_name_row.get_text().strip() + if not name: + return + project = self._pm.add_project(name) + project_id = project.id + elif self._exist_radio and self._exist_radio.get_active() and self._proj_combo: + idx = self._proj_combo.get_selected() + if idx == Gtk.INVALID_LIST_POSITION or idx >= len(self._existing_projects): + return + project_id = self._existing_projects[idx].id + else: + return + + for op in selected: + http_req = build_http_request(op) + self._pm.add_request(project_id, op.name, http_req) + + self.emit('import-completed', project_id) + self.close() + + # ----------------------------------------------------------------- helpers + + def _set_err(self, msg: str): + self._err_label.set_text(msg) + self._err_label.set_visible(True) + + def _fetch_failed(self, msg: str): + self._spinner.stop() + self._spinner.set_visible(False) + self._fetch_btn.set_sensitive(True) + self._set_err(msg) + + def _populate_ops_page(self, result: WsdlParseResult): + self._svc_name_row.set_subtitle(result.service_name) + self._svc_url_row.set_subtitle(result.endpoint_url or "Not specified") + + while child := self._ops_list.get_first_child(): + self._ops_list.remove(child) + self._op_checkboxes.clear() + + for op in result.operations: + row = Adw.ActionRow() + row.set_title(op.name) + if op.soap_action: + row.set_subtitle(f"SOAPAction: {op.soap_action}") + cb = Gtk.CheckButton() + cb.set_active(True) + row.add_prefix(cb) + row.set_activatable_widget(cb) + self._ops_list.append(row) + self._op_checkboxes.append((op, cb)) + + self._new_name_row.set_text(result.service_name) + self._sel_all_btn.set_label("Select All") diff --git a/src/wsdl_importer.py b/src/wsdl_importer.py new file mode 100644 index 0000000..77cb1e3 --- /dev/null +++ b/src/wsdl_importer.py @@ -0,0 +1,202 @@ +# wsdl_importer.py +# +# Copyright 2025 Pavel Baksy +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# 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 for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: GPL-3.0-or-later + +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple + +_WSDL = 'http://schemas.xmlsoap.org/wsdl/' +_SOAP11 = 'http://schemas.xmlsoap.org/wsdl/soap/' +_SOAP12 = 'http://schemas.xmlsoap.org/wsdl/soap12/' +_ENV11 = 'http://schemas.xmlsoap.org/soap/envelope/' +_ENV12 = 'http://www.w3.org/2003/05/soap-envelope' + + +def _q(ns: str, tag: str) -> str: + return f'{{{ns}}}{tag}' + + +def _local(tag: str) -> str: + return tag.split('}')[-1] if '}' in tag else tag + + +@dataclass +class WsdlOperation: + name: str + soap_action: str + endpoint_url: str + soap_version: str # '1.1' or '1.2' + target_namespace: str + body_template: str + + +@dataclass +class WsdlParseResult: + service_name: str + endpoint_url: str + operations: List[WsdlOperation] = field(default_factory=list) + error: Optional[str] = None + + +def parse_wsdl(xml_content: str) -> WsdlParseResult: + """Parse WSDL 1.1 XML and return service info + list of operations.""" + try: + root = ET.fromstring(xml_content) + except ET.ParseError as e: + return WsdlParseResult(service_name='', endpoint_url='', error=f'Invalid XML: {e}') + + if _local(root.tag) != 'definitions': + return WsdlParseResult( + service_name='', endpoint_url='', + error='Document is not a WSDL 1.1 definitions element' + ) + + target_ns = root.get('targetNamespace', '') + service_name = root.get('name', 'WSDL Service') + + # --- Service element: name + endpoint URL --- + service_el = root.find(_q(_WSDL, 'service')) + if service_el is None: + service_el = root.find('service') + + if service_el is not None and service_el.get('name'): + service_name = service_el.get('name') + + endpoint_url = '' + default_soap_ver = '1.1' + + if service_el is not None: + for port in service_el: + addr11 = port.find(_q(_SOAP11, 'address')) + if addr11 is not None: + endpoint_url = addr11.get('location', '') + default_soap_ver = '1.1' + break + addr12 = port.find(_q(_SOAP12, 'address')) + if addr12 is not None: + endpoint_url = addr12.get('location', '') + default_soap_ver = '1.2' + break + + # --- Bindings: collect (soap_action, soap_version) per operation name --- + # op_name -> (soap_action, soap_version) + op_info: Dict[str, Tuple[str, str]] = {} + + for binding in root.iter(): + if _local(binding.tag) != 'binding': + continue + + is11 = binding.find(_q(_SOAP11, 'binding')) is not None + is12 = binding.find(_q(_SOAP12, 'binding')) is not None + if not is11 and not is12: + continue + bv = '1.2' if is12 else '1.1' + + for op in binding: + if _local(op.tag) != 'operation': + continue + raw_name = op.get('name', '') + op_name = _local(raw_name) + if not op_name: + continue + + soap_op = op.find(_q(_SOAP11, 'operation')) + if soap_op is None: + soap_op = op.find(_q(_SOAP12, 'operation')) + action = soap_op.get('soapAction', '') if soap_op is not None else '' + + if op_name not in op_info: + op_info[op_name] = (action, bv) + + # Fallback: portType operations when no SOAP binding was found + if not op_info: + for pt in root.iter(): + if _local(pt.tag) != 'portType': + continue + for op in pt: + if _local(op.tag) == 'operation': + op_name = op.get('name', '') + if op_name and op_name not in op_info: + op_info[op_name] = ('', default_soap_ver) + + if not op_info: + return WsdlParseResult( + service_name=service_name, + endpoint_url=endpoint_url, + error='No SOAP operations found in this WSDL document' + ) + + operations = [ + WsdlOperation( + name=name, + soap_action=action, + endpoint_url=endpoint_url, + soap_version=ver, + target_namespace=target_ns, + body_template=_build_envelope(name, target_ns, ver), + ) + for name, (action, ver) in op_info.items() + ] + + return WsdlParseResult( + service_name=service_name or 'WSDL Service', + endpoint_url=endpoint_url, + operations=operations, + ) + + +def _build_envelope(op_name: str, target_ns: str, soap_version: str) -> str: + env_ns = _ENV12 if soap_version == '1.2' else _ENV11 + ns_decl = f'\n xmlns:tns="{target_ns}"' if target_ns else '' + op_el = f'tns:{op_name}' if target_ns else op_name + return ( + f'\n' + f'\n' + f' \n' + f' \n' + f' <{op_el}>\n' + f' \n' + f' \n' + f' \n' + f'' + ) + + +def build_http_request(operation: WsdlOperation): + """Convert a WsdlOperation into an HttpRequest ready to send.""" + from .models import HttpRequest + + headers: Dict[str, str] = {} + if operation.soap_version == '1.2': + ct = 'application/soap+xml; charset=utf-8' + if operation.soap_action: + ct += f'; action="{operation.soap_action}"' + headers['Content-Type'] = ct + else: + headers['Content-Type'] = 'text/xml; charset=utf-8' + if operation.soap_action: + headers['SOAPAction'] = f'"{operation.soap_action}"' + + return HttpRequest( + method='POST', + url=operation.endpoint_url, + headers=headers, + body=operation.body_template, + syntax='XML', + )