http properties dialog prototype

This commit is contained in:
DYefremov
2018-11-04 00:36:07 +03:00
parent 2b3c357657
commit 3ac16cb7af
2 changed files with 599 additions and 212 deletions

View File

@@ -1,3 +1,13 @@
import socket
from enum import Enum
from ftplib import error_perm, FTP
from urllib.error import URLError, HTTPError
from urllib.parse import urlencode
from urllib.request import urlopen
from xml.dom.minidom import parse
from app.commons import run_task, run_idle
from app.connections import test_telnet
from app.properties import write_config, Profile, get_default_settings
from .uicommons import Gtk, UI_RESOURCES_PATH, TEXT_DOMAIN
from .main_helper import update_entry_data
@@ -7,6 +17,12 @@ def show_settings_dialog(transient, options):
return SettingsDialog(transient, options).show()
class Property(Enum):
FTP = "ftp"
HTTP = "http"
TELNET = "telnet"
class SettingsDialog:
def __init__(self, transient, options):
@@ -14,7 +30,9 @@ class SettingsDialog:
"on_picons_dir_field_icon_press": self.on_picons_dir_field_icon_press,
"on_profile_changed": self.on_profile_changed,
"on_reset": self.on_reset,
"apply_settings": self.apply_settings}
"apply_settings": self.apply_settings,
"on_connection_test": self.on_connection_test,
"on_info_bar_close": self.on_info_bar_close}
builder = Gtk.Builder()
builder.set_translation_domain(TEXT_DOMAIN)
@@ -27,6 +45,9 @@ class SettingsDialog:
self._port_field = builder.get_object("port_field")
self._login_field = builder.get_object("login_field")
self._password_field = builder.get_object("password_field")
self._http_login_field = builder.get_object("http_login_field")
self._http_password_field = builder.get_object("http_password_field")
self._http_port_field = builder.get_object("http_port_field")
self._telnet_login_field = builder.get_object("telnet_login_field")
self._telnet_password_field = builder.get_object("telnet_password_field")
self._telnet_port_field = builder.get_object("telnet_port_field")
@@ -40,13 +61,17 @@ class SettingsDialog:
self._enigma_radio_button = builder.get_object("enigma_radio_button")
self._neutrino_radio_button = builder.get_object("neutrino_radio_button")
self._support_ver5_check_button = builder.get_object("support_ver5_check_button")
self._settings_stack = builder.get_object("settings_stack")
self._info_bar = builder.get_object("info_bar")
self._message_label = builder.get_object("info_bar_message_label")
self._test_spinner = builder.get_object("test_spinner")
self._options = options
self._active_profile = options.get("profile")
self.set_settings()
profile = Profile(self._active_profile)
self._neutrino_radio_button.set_active(profile is Profile.NEUTRINO_MP)
self._support_ver5_check_button.set_sensitive(profile is not Profile.NEUTRINO_MP)
self._settings_stack.get_child_by_name(Property.HTTP.value).set_visible(profile is not Profile.NEUTRINO_MP)
def show(self):
response = self._dialog.run()
@@ -65,6 +90,7 @@ class SettingsDialog:
def on_profile_changed(self, item):
profile = Profile.ENIGMA_2 if self._enigma_radio_button.get_active() else Profile.NEUTRINO_MP
self._settings_stack.get_child_by_name(Property.HTTP.value).set_visible(profile is not Profile.NEUTRINO_MP)
self.set_profile(profile)
self._support_ver5_check_button.set_sensitive(profile is Profile.ENIGMA_2)
@@ -89,6 +115,9 @@ class SettingsDialog:
self._port_field.set_text(options.get("port", ""))
self._login_field.set_text(options.get("user", ""))
self._password_field.set_text(options.get("password", ""))
self._http_login_field.set_text(options.get("http_user", ""))
self._http_password_field.set_text(options.get("http_password", ""))
self._http_port_field.set_text(options.get("http_port", "80"))
self._telnet_login_field.set_text(options.get("telnet_user", ""))
self._telnet_password_field.set_text(options.get("telnet_password", ""))
self._telnet_port_field.set_text(options.get("telnet_port", ""))
@@ -111,6 +140,9 @@ class SettingsDialog:
options["port"] = self._port_field.get_text()
options["user"] = self._login_field.get_text()
options["password"] = self._password_field.get_text()
options["http_user"] = self._http_login_field.get_text()
options["http_password"] = self._http_password_field.get_text()
options["http_port"] = self._http_port_field.get_text()
options["telnet_user"] = self._telnet_login_field.get_text()
options["telnet_password"] = self._telnet_password_field.get_text()
options["telnet_port"] = self._telnet_port_field.get_text()
@@ -124,6 +156,75 @@ class SettingsDialog:
if profile is Profile.ENIGMA_2:
options["v5_support"] = self._support_ver5_check_button.get_active()
@run_task
def on_connection_test(self, item):
if self._test_spinner.get_state() is Gtk.StateType.ACTIVE:
return
self.show_spinner(True)
current_property = Property(self._settings_stack.get_visible_child_name())
if current_property is Property.HTTP:
self.test_http()
elif current_property is Property.TELNET:
self.test_telnet()
elif current_property is Property.FTP:
self.test_ftp()
def test_http(self):
user, password = self._http_login_field.get_text(), self._http_password_field.get_text()
try:
params = urlencode({"text": "Connection test", "type": 2, "timeout": 5})
with urlopen("http://{}/web/message?%s".format(self._host_field.get_text()) % params, timeout=2) as f:
dom = parse(f)
for elem in dom.getElementsByTagName("e2simplexmlresult"):
for ch in elem.childNodes:
if ch.nodeType == ch.ELEMENT_NODE:
msg = "".join(t.nodeValue for t in ch.childNodes if t.nodeType == t.TEXT_NODE)
self.show_info_message(msg, Gtk.MessageType.INFO)
except (URLError, HTTPError) as e:
self.show_info_message(str(e), Gtk.MessageType.ERROR)
finally:
self.show_spinner(False)
def test_telnet(self):
timeout = int(self._telnet_timeout_spin_button.get_value())
host, port = self._host_field.get_text(), self._telnet_port_field.get_text()
user, password = self._telnet_login_field.get_text(), self._telnet_password_field.get_text()
try:
gen = test_telnet(host, port, user, password, timeout)
res = next(gen)
print(res)
res = next(gen)
self.show_info_message(str(res), Gtk.MessageType.INFO)
self.show_spinner(False)
except (socket.timeout, OSError) as e:
self.show_info_message(str(e), Gtk.MessageType.ERROR)
self.show_spinner(False)
def test_ftp(self):
host, port = self._host_field.get_text(), self._port_field.get_text()
user, password = self._login_field.get_text(), self._password_field.get_text()
try:
with FTP(host=host, user=user, passwd=password, timeout=5) as ftp:
self.show_info_message("OK. {}".format(ftp.getwelcome()), Gtk.MessageType.INFO)
except (error_perm, ConnectionRefusedError) as e:
self.show_info_message(str(e), Gtk.MessageType.ERROR)
finally:
self.show_spinner(False)
@run_idle
def show_info_message(self, text, message_type):
self._info_bar.set_visible(True)
self._info_bar.set_message_type(message_type)
self._message_label.set_text(text)
@run_idle
def show_spinner(self, show):
self._test_spinner.start() if show else self._test_spinner.stop()
self._test_spinner.set_state(Gtk.StateType.ACTIVE if show else Gtk.StateType.NORMAL)
def on_info_bar_close(self, bar=None, resp=None):
self._info_bar.set_visible(False)
if __name__ == "__main__":
pass