from __future__ import annotations from typing import Literal from PySide6.QtWidgets import QApplication, QMessageBox from dns_benchmark_dialog import DNSBenchmarkDialog from main_window.constants import LOCATION_TARGET_ROLE from traffic_mode_dialog import TrafficModeDialog class RuntimeOpsMixin: def on_toggle_autoconnect(self) -> None: def work(): current = self.ctrl.vpn_autoconnect_enabled() enable = not current self.ctrl.vpn_set_autoconnect(enable) self.ctrl.log_gui(f"VPN autoconnect set to {enable}") self.refresh_vpn_tab() self._safe(work, title="Autoconnect error") def on_location_activated(self, _index: int) -> None: self._safe(self._apply_selected_location, title="Location error") def on_set_location(self) -> None: self._safe(self._apply_selected_location, title="Location error") def _apply_selected_location(self) -> None: idx = self.cmb_locations.currentIndex() if idx < 0: return iso = str(self.cmb_locations.currentData() or "").strip().upper() target = str(self.cmb_locations.currentData(LOCATION_TARGET_ROLE) or "").strip() label = str(self.cmb_locations.currentText() or "").strip() if not target: target = iso if not iso or not target: return desired = (self._vpn_desired_location or "").strip().lower() if desired and desired in (iso.lower(), target.lower()): return self.lbl_locations_meta.setText(f"Applying location {target}...") self.lbl_locations_meta.setStyleSheet("color: orange;") self._start_vpn_location_switching(target) self.refresh_login_banner() QApplication.processEvents() try: self.ctrl.vpn_set_location(target=target, iso=iso, label=label) except Exception: self._stop_vpn_location_switching() self.refresh_login_banner() raise self.ctrl.log_gui(f"VPN location set to {target} ({iso})") self._vpn_desired_location = target self.refresh_vpn_tab() self._trigger_vpn_egress_refresh(reason=f"location switch to {target}") # ---- Routes actions ------------------------------------------------ def on_routes_action( self, action: Literal["start", "stop", "restart"] ) -> None: def work(): res = self.ctrl.routes_service_action(action) self._set_text(self.txt_routes, res.pretty_text or str(res)) self.refresh_status_tab() self._safe(work, title="Routes error") def _append_routes_log(self, msg: str) -> None: line = (msg or "").strip() if not line: return self._append_text(self.txt_routes, line + "\n") self.ctrl.log_gui(line) def on_open_traffic_settings(self) -> None: def work(): def refresh_all_traffic() -> None: self.refresh_routes_tab() self.refresh_status_tab() dlg = TrafficModeDialog( self.ctrl, log_cb=self._append_routes_log, refresh_cb=refresh_all_traffic, parent=self, ) dlg.exec() refresh_all_traffic() self._safe(work, title="Traffic mode dialog error") def on_test_traffic_mode(self) -> None: def work(): view = self.ctrl.traffic_mode_test() msg = ( f"Traffic mode test: desired={view.desired_mode}, applied={view.applied_mode}, " f"iface={view.active_iface or '-'}, probe_ok={view.probe_ok}, " f"healthy={view.healthy}, auto_local_bypass={view.auto_local_bypass}, " f"bypass_routes={view.bypass_candidates}, overrides={view.overrides_applied}, " f"cgroup_uids={view.cgroup_resolved_uids}, cgroup_warning={view.cgroup_warning or '-'}, " f"message={view.message}, probe={view.probe_message}" ) self._append_routes_log(msg) self.refresh_routes_tab() self.refresh_status_tab() self._safe(work, title="Traffic mode test error") def on_routes_precheck_debug(self) -> None: def work(): res = self.ctrl.routes_precheck_debug(run_now=True) txt = (res.pretty_text or "").strip() if res.ok: QMessageBox.information(self, "Resolve precheck debug", txt or "OK") else: QMessageBox.critical(self, "Resolve precheck debug", txt or "ERROR") self.refresh_routes_tab() self.refresh_status_tab() self.refresh_trace_tab() self._safe(work, title="Resolve precheck debug error") def on_toggle_timer(self) -> None: def work(): enabled = self.chk_timer.isChecked() res = self.ctrl.routes_timer_set(enabled) self.ctrl.log_gui(f"Routes timer set to {enabled}") self._set_text(self.txt_routes, res.pretty_text or str(res)) self.refresh_routes_tab() self._safe(work, title="Timer error") def on_fix_policy_route(self) -> None: def work(): res = self.ctrl.routes_fix_policy_route() self._set_text(self.txt_routes, res.pretty_text or str(res)) self.refresh_status_tab() self._safe(work, title="Policy route error") # ---- DNS actions --------------------------------------------------- def _schedule_dns_autosave(self, _text: str = "") -> None: if self._dns_ui_refresh: return self.dns_save_timer.start() def _apply_dns_autosave(self) -> None: def work(): if self._dns_ui_refresh: return self.ctrl.dns_mode_set( self.chk_dns_via_smartdns.isChecked(), self.ent_smartdns_addr.text().strip(), ) self.ctrl.log_gui("DNS settings autosaved") self._safe(work, title="DNS save error") def on_open_dns_benchmark(self) -> None: def work(): dlg = DNSBenchmarkDialog( self.ctrl, settings=self._ui_settings, refresh_cb=self.refresh_dns_tab, parent=self, ) dlg.exec() self.refresh_dns_tab() self._safe(work, title="DNS benchmark error") def on_dns_mode_toggle(self) -> None: def work(): via = self.chk_dns_via_smartdns.isChecked() self.ctrl.dns_mode_set(via, self.ent_smartdns_addr.text().strip()) mode = "hybrid_wildcard" if via else "direct" self.ctrl.log_gui(f"DNS mode changed: mode={mode}") self.refresh_dns_tab() self._safe(work, title="DNS mode error") def on_smartdns_unit_toggle(self) -> None: def work(): enable = self.chk_dns_unit_relay.isChecked() action = "start" if enable else "stop" self.ctrl.smartdns_service_action(action) self.ctrl.log_smartdns(f"SmartDNS unit action from GUI: {action}") self.refresh_dns_tab() self.refresh_status_tab() self._safe(work, title="SmartDNS error") def on_smartdns_runtime_toggle(self) -> None: def work(): if self._dns_ui_refresh: return enable = self.chk_dns_runtime_nftset.isChecked() st = self.ctrl.smartdns_runtime_set(enabled=enable, restart=True) self.ctrl.log_smartdns( f"SmartDNS runtime accelerator set from GUI: enabled={enable} changed={st.changed} restarted={st.restarted} source={st.wildcard_source}" ) self.refresh_dns_tab() self.refresh_trace_tab() self._safe(work, title="SmartDNS runtime error") def on_smartdns_prewarm(self) -> None: def work(): aggressive = bool(self.chk_routes_prewarm_aggressive.isChecked()) result = self.ctrl.smartdns_prewarm(aggressive_subs=aggressive) mode_txt = "aggressive_subs=on" if aggressive else "aggressive_subs=off" self.ctrl.log_smartdns(f"SmartDNS prewarm requested from GUI: {mode_txt}") txt = (result.pretty_text or "").strip() if result.ok: QMessageBox.information(self, "SmartDNS prewarm", txt or "OK") else: QMessageBox.critical(self, "SmartDNS prewarm", txt or "ERROR") self.refresh_trace_tab() self._safe(work, title="SmartDNS prewarm error") def _update_prewarm_mode_label(self, _state: int = 0) -> None: aggressive = bool(self.chk_routes_prewarm_aggressive.isChecked()) if aggressive: self.lbl_routes_prewarm_mode.setText("Prewarm mode: aggressive (subs enabled)") self.lbl_routes_prewarm_mode.setStyleSheet("color: orange;") else: self.lbl_routes_prewarm_mode.setText("Prewarm mode: wildcard-only") self.lbl_routes_prewarm_mode.setStyleSheet("color: gray;") def _on_prewarm_aggressive_changed(self, _state: int = 0) -> None: self._update_prewarm_mode_label(_state) self._save_ui_preferences() # ---- Domains actions ----------------------------------------------- def on_domains_load(self) -> None: def work(): name = self._get_selected_domains_file() content, source, path = self._load_file_content(name) is_readonly = name in ("last-ips-map-direct", "last-ips-map-wildcard", "wildcard-observed-hosts") self.txt_domains.setReadOnly(is_readonly) self.btn_domains_save.setEnabled(not is_readonly) self._set_text(self.txt_domains, content) ro = "read-only" if is_readonly else "editable" self.lbl_domains_info.setText(f"{name} ({source}, {ro}) [{path}]") self._safe(work, title="Domains load error") def on_domains_save(self) -> None: def work(): name = self._get_selected_domains_file() content = self.txt_domains.toPlainText() self._save_file_content(name, content) self.ctrl.log_gui(f"Domains file saved: {name}") self._safe(work, title="Domains save error") # ---- close event --------------------------------------------------- def closeEvent(self, event) -> None: # pragma: no cover - GUI try: self._save_ui_preferences() self._login_flow_autopoll_stop() self.loc_typeahead_timer.stop() if self.locations_thread: self.locations_thread.quit() self.locations_thread.wait(1500) if self.events_thread: self.events_thread.stop() self.events_thread.wait(1500) finally: super().closeEvent(event)