Coverage for modules/org/openteacher/interfaces/qt/testMode/connection/connection : 45%

#! /usr/bin/env python3 

# -*- coding: utf-8 -*- 

#       Copyright 2011-2012, Milan Boers 

#       Copyright 2012-2013, Marten de Vries 

# 

#       This file is part of OpenTeacher. 

# 

#       OpenTeacher 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. 

# 

#       OpenTeacher 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 OpenTeacher.  If not, see <http://www.gnu.org/licenses/>. 

import os 

import urllib.parse 

import urllib.error 

import urllib.request 

import socket 

import json 

import base64 

import io 

MAXVERSION = 1.0 

def getConnectWidget(): 

        class ConnectWidget(QtWidgets.QWidget): 

                """We use HTTP so there's not actually a continuous connection, but 

                   this class is basically here to check whether a server exists. 

                """ 

                def __init__(self, connection, *args, **kwargs): 

                        super().__init__(*args, **kwargs) 

                        self.connection = connection 

                        self.connectLayout = QtWidgets.QFormLayout() 

                        self.setLayout(self.connectLayout) 

                        self.serverField = QtWidgets.QLineEdit() 

                        self.serverLabel = QtWidgets.QLabel() 

                        self.connectLayout.addRow(self.serverLabel, self.serverField) 

                        self.connectButton = QtWidgets.QPushButton() 

                        self.connectButton.clicked.connect(lambda: self.connection.connect(self.serverField.text())) 

                        self.connectLayout.addRow(self.connectButton) 

                        self.serverField.returnPressed.connect(self.connectButton.click) 

                        self.retranslate() 

                def retranslate(self): 

                        self.serverLabel.setText(_("Server IP or hostname:")) 

                        self.connectButton.setText(_("Connect")) 

        return ConnectWidget 

def getLoginWidget(): 

        class LoginWidget(QtWidgets.QWidget): 

                def __init__(self, connection, loginid, *args, **kwargs): 

                        super().__init__(*args, **kwargs) 

                        self.connection = connection 

                        self.loginLayout = QtWidgets.QFormLayout() 

                        self.setLayout(self.loginLayout) 

                        self.usernameLabel = QtWidgets.QLabel() 

                        self.usernameField = QtWidgets.QLineEdit() 

                        self.loginLayout.addRow(self.usernameLabel, self.usernameField) 

                        self.passwordLabel = QtWidgets.QLabel() 

                        self.passwordField = QtWidgets.QLineEdit() 

                        self.passwordField.setEchoMode(QtWidgets.QLineEdit.Password) 

                        self.loginLayout.addRow(self.passwordLabel, self.passwordField) 

                        self.checkButton = QtWidgets.QPushButton() 

                        self.checkButton.clicked.connect(lambda: connection.checkLogin( 

                                self.usernameField.text(), 

                                self.passwordField.text(), 

                                loginid 

                        )) 

                        self.loginLayout.addRow(self.checkButton) 

                        self.passwordField.returnPressed.connect(self.checkButton.click) 

                def retranslate(self): 

                        self.usernameLabel.setText(_("Username:")) 

                        self.passwordLabel.setText(_("Password:")) 

                        self.checkButton.setText(_("Login")) 

        return LoginWidget 

def getConnectLoginWidget(): 

        class ConnectLoginWidget(QtWidgets.QWidget): 

                def __init__(self, connection, loginid, *args, **kwargs): 

                        super().__init__(*args, **kwargs) 

                        self.layout = QtWidgets.QStackedLayout() 

                        self.setLayout(self.layout) 

                        self.connectWidget = ConnectWidget(connection) 

                        connection.connected.handle(self.afterConnect) 

                        self.layout.addWidget(self.connectWidget) 

                        self.loginWidget = LoginWidget(connection, loginid) 

                        self.layout.addWidget(self.loginWidget) 

                        self.retranslate() 

                def afterConnect(self): 

                        self.layout.setCurrentWidget(self.loginWidget) 

                def retranslate(self): 

                        self.connectWidget.retranslate() 

                        self.loginWidget.retranslate() 

        return ConnectLoginWidget 

class Connection: 

        def __init__(self, modules, *args, **kwargs): 

                super().__init__(*args, **kwargs) 

                self._modules = modules 

                # Connected event 

                self.connected = self._modules.default(type="event").createEvent() 

                # LoggedIn event 

                self.loggedIn = self._modules.default(type="event").createEvent() 

                # Setup opener 

                self.opener = urllib.request.build_opener() 

                self.server = None 

                self.serverName = None 

                self.auth = False 

        def retranslate(self): 

143                if hasattr(self, "loginTab"): 

                        self.connectLoginWidget.retranslate() 

                        self.loginTab.title = _("Login") 

        @property 

        def connectedLoggedIn(self): 

                """'Connected' to the server and logged in?""" 

                return self.server != None and self.auth == True 

        def connect(self, hostname): 

                """'Connect' to server""" 

                try: 

                        assert hostname 

                        #Replace hostname by IP (so DNS is not needed at every 

                        #request. Will speed things up.) 

                        hostname = socket.gethostbyname(hostname) 

                except (socket.gaierror, AssertionError): 

                        # Could not connect 

                        dialogShower = self._modules.default(type="dialogShower") 

                        dialogShower.showError.send(self.loginTab, _("Could not connect to the server. Possibly wrong hostname.")) 

                else: 

                        print(hostname) 

                        # Everything OK, Connected 

                        self.server = hostname 

                        # Try to fetch the index page 

                        index = self.get("https://%s:8080/" % (hostname)) 

                        self.serverName = index["name"] 

                        self.connected.send() 

        def _open(self, request): 

                try: 

                        body = io.TextIOWrapper(self.opener.open(request), encoding='UTF-8') 

                        return json.load(body) 

                except urllib.error.HTTPError as e: 

                        return e 

        def get(self, path): 

                """Get path""" 

                url = "https://%s:8080/%s" % (self.server, path) 

                return self._open(url) 

        def post(self, path, data): 

                data = urllib.parse.urlencode(data) 

                request = urllib.request.Request("https://%s:8080/%s" % (self.server, path), data) 

                return self._open(request) 

        def put(self, path, data): 

                data = urllib.parse.urlencode(data) 

                request = urllib.request.Request("https://%s:8080/%s" % (self.server, path), data) 

                request.get_method = lambda: 'PUT' 

                return self._open(request) 

        def login(self, loginid): 

                """Method to log in. You need to send a uuid along, so your 

                   loggedIn doesn't react to other requests. 

                """ 

                if self.connectedLoggedIn: 

                        self._afterLogin(loginid) 

                else: 

                        self.connectLoginWidget = ConnectLoginWidget(self, loginid) 

                        module = self._modules.default("active", type="ui") 

                        self.loginTab = module.addCustomTab(self.connectLoginWidget) 

                        #set tab title by retranslating 

                        self.retranslate() 

                        self.loginTab.closeRequested.handle(self.loginTab.close) 

        def checkLogin(self, username, passwd, loginid): 

                """Checks if login is right (and implicitly fetches the token) 

                   (to be used only inside this module) 

                """ 

                loginCreds = u"%s:%s" % (username, passwd) 

                loginCreds = loginCreds.encode("utf-8") 

                loginCreds = base64.b64encode(loginCreds) 

                self.opener.addheaders = [("Authorization", b"Basic " + loginCreds)] 

                me = self.get("users/me") 

                if type(me) == urllib.error.HTTPError: 

                        # User was not logged in 

                        dialogShower = self._modules.default("active", type="dialogShower") 

                        dialogShower.showError.send(self.loginTab, _("Could not login. Wrong username or password.")) 

                else: 

                        self.userId = int(os.path.basename(me)) 

                        self._afterLogin(loginid) 

        def _afterLogin(self, loginid): 

                """Logged in, set var to server""" 

                self.auth = True 

                self.loginTab.close() 

                self.loggedIn.send(loginid) 

class TestModeConnectionModule: 

        def __init__(self, moduleManager, *args, **kwargs): 

                super().__init__(*args, **kwargs) 

                self._mm = moduleManager 

                self.type = "testModeConnection" 

                self.priorities = { 

                        "default": 444, 

                } 

                self.uses = ( 

                        self._mm.mods(type="translator"), 

                ) 

                self.requires = ( 

                        self._mm.mods(type="event"), 

                        self._mm.mods(type="ui"), 

                ) 

                self.filesWithTranslations = ("connection.py",) 

        def _retranslate(self): 

                #Translations 

                global _ 

                global ngettext 

                try: 

                        translator = self._modules.default("active", type="translator") 

                except IndexError: 

exit                        _, ngettext = str, lambda a, b, n: a if n == 1 else b 

                else: 

                        _, ngettext = translator.gettextFunctions( 

                                self._mm.resourcePath("translations") 

                        ) 

                self.connection.retranslate() 

        def enable(self): 

                global QtWidgets 

                try: 

                        from PyQt5 import QtWidgets 

                except ImportError: 

                        return 

                global ConnectLoginWidget, ConnectWidget, LoginWidget 

                ConnectLoginWidget = getConnectLoginWidget() 

                ConnectWidget = getConnectWidget() 

                LoginWidget = getLoginWidget() 

                self._modules = set(self._mm.mods(type="modules")).pop() 

                self.connection = Connection(self._modules) 

                #setup translation 

                try: 

                        translator = self._modules.default("active", type="translator") 

                except IndexError: 

                        pass 

                else: 

                        translator.languageChanged.handle(self._retranslate) 

                self._retranslate() 

                self.active = True 

        def disable(self): 

                self.active = False 

                del self.connection 

                del self._modules 

        def getConnection(self): 

                return self.connection 

def init(moduleManager): 

        return TestModeConnectionModule(moduleManager)