KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use send_text_response method in pytractor

Best Python code snippet using pytractor_python

shell_app.py

Source:shell_app.py

...12class ThreadingHTTPServer(SocketServer.ThreadingMixIn,13        BaseHTTPServer.HTTPServer):14    pass15class Request(BaseHTTPServer.BaseHTTPRequestHandler):16    def send_text_response(self, text, status_code=200):17        self.send_response(status_code)18        self.send_header('Content-type', 'text/plain')19        self.end_headers()20        self.wfile.write(text)21        self.wfile.close()22    def do_PUT(self):23        try:24            if self.headers.get('Auth-Token') != auth_token or \25                    self.headers.get('User-Agent') != 'pritunl' or \26                    self.headers.get('Origin') or \27                    self.headers.get('Referer'):28                self.send_response(401)29                return30            self.args = self.path.split('/')31            data_len = int(self.headers.get('Content-Length', 0))32            data = self.rfile.read(data_len)33            if data:34                self.data = json.loads(data)35            else:36                self.data = {}37            if self.args[1] == 'remove':38                self.do_remove()39            elif self.args[1] == 'start':40                self.do_start()41            elif self.args[1] == 'stop':42                self.do_stop()43            elif self.args[1] == 'enable':44                self.do_enable()45            elif self.args[1] == 'disable':46                self.do_disable()47            else:48                self.send_response(404)49        except Exception as exception:50            self.send_text_response(str(exception), 500)51    def do_POST(self):52        try:53            if self.headers.get('Auth-Token') != auth_token or \54                    self.headers.get('User-Agent') != 'pritunl' or \55                    self.headers.get('Origin') or \56                    self.headers.get('Referer'):57                self.send_response(401)58                return59            self.args = self.path.split('/')60            data_len = int(self.headers.get('Content-Length', 0))61            data = self.rfile.read(data_len)62            if data:63                self.data = json.loads(data)64            else:65                self.data = {}66            if self.args[1] == 'import':67                self.do_import()68            else:69                self.send_response(404)70        except Exception as exception:71            self.send_text_response(str(exception), 500)72    def do_DELETE(self):73        try:74            if self.headers.get('Auth-Token') != auth_token or \75                    self.headers.get('User-Agent') != 'pritunl' or \76                    self.headers.get('Origin') or \77                    self.headers.get('Referer'):78                self.send_response(401)79                return80            self.args = self.path.split('/')81            data_len = int(self.headers.get('Content-Length', 0))82            data = self.rfile.read(data_len)83            if data:84                self.data = json.loads(data)85            else:86                self.data = {}87            if self.args[1] == 'remove':88                self.do_remove()89            else:90                self.send_response(404)91        except Exception as exception:92            self.send_text_response(str(exception), 500)93    def do_GET(self):94        try:95            self.args = self.path.split('/')96            if self.args[1] == 'token':97                self.do_token()98            elif self.args[1] == 'list':99                if self.headers.get('Auth-Token') != auth_token or \100                        self.headers.get('User-Agent') != 'pritunl' or \101                        self.headers.get('Origin') or \102                        self.headers.get('Referer'):103                    self.send_response(401)104                    return105                self.do_list()106            else:107                self.send_response(404)108        except Exception as exception:109            self.send_text_response(str(exception), 500)110    def do_token(self):111        self.send_text_response(auth_token, 200)112    def do_import(self):113        profile_path = self.data.get('profile_path')114        profile_uri = self.data.get('profile_uri')115        if profile_path:116            profile.import_file(profile_path)117        elif profile_uri:118            try:119                profile.import_uri(profile_uri)120            except httplib.HTTPException:121                self.send_text_response('Unable to retrieve profile or ' +122                    'profile path does not exist', 500)123        else:124            raise ValueError('Must have profile_path or profile_uri')125        self.send_response(200)126    def do_remove(self):127        prfl = profile.Profile.get_profile(self.args[2])128        if prfl:129            prfl.delete()130        self.send_response(200)131    def do_start(self):132        passwd = self.data.get('passwd')133        evt = threading.Event()134        def status_callback():135            pass136        def connect_callback():137            evt.set()138            pass139        prfl = profile.Profile.get_profile(self.args[2])140        if not prfl:141            self.send_text_response('Profile not found', 404)142            return143        prfl.sync_conf()144        if prfl.auth_type and not passwd:145            self.send_text_response('Password required', 400)146            return147        if not prfl.start(status_callback, connect_callback, passwd=passwd):148            self.send_text_response('Profile has already been started', 500)149            return150        evt.wait(CONNECT_TIMEOUT + 5)151        if prfl.status in ERROR_STATES:152            error_msg = {153                ERROR: 'An error occurred while connecting to server',154                AUTH_ERROR: 'Failed to authenticate with server',155                TIMEOUT_ERROR: 'Server connection timed out',156            }[prfl.status]157            self.send_text_response(158                error_msg, 500)159            return160        self.send_response(200)161    def do_stop(self):162        prfl = profile.Profile.get_profile(self.args[2])163        if not prfl:164            self.send_text_response('Profile not found', 404)165            return166        prfl.stop()167        self.send_response(200)168    def do_enable(self):169        prfl = profile.Profile.get_profile(self.args[2])170        if not prfl:171            self.send_text_response('Profile not found', 404)172            return173        prfl.set_autostart(True)174        self.send_response(200)175    def do_disable(self):176        prfl = profile.Profile.get_profile(self.args[2])177        if not prfl:178            self.send_text_response('Profile not found', 404)179            return180        prfl.set_autostart(False)181        self.send_response(200)182    def do_list(self):183        output = '%s %s %s %s\n' % (184            'PROFILE_ID'.ljust(33, ' '),185            'ENABLED'.ljust(8, ' '),186            'STATE'.ljust(13, ' '),187            'NAME',188        )189        for prfl in profile.Profile.iter_profiles():190            output += '%s %s %s %s\n' % (191                prfl.id.ljust(33, ' '),192                str(prfl.autostart).lower().ljust(8, ' '),193                prfl.status.ljust(13, ' '),194                prfl.name,195            )196        self.send_text_response(output.rstrip('\n'))197class ShellApp(object):198    def __init__(self):199        self.start_server()200    def autostart(self):201        def status_callback():202            pass203        for prfl in profile.Profile.iter_profiles():204            if not prfl.autostart:205                continue206            logger.info('Auto starting profile...', 'shell',207                profile_id=prfl.id,208                profile_name=prfl.name,209            )210            def connect_callback():...

webserver.py

Source:webserver.py

...18        except ValueError:19            path = self.path20        if path == "/":21            with open("webui/index.html", "r") as index:22                self.send_text_response(index.read())23            return24        if path == "/devices":25            result = []26            for device in devices:27                result.append({28                    "name": device.name,29                    "type": device.__class__.__name__,30                    "status": device.get_status()31                })32            self.send_text_response(result)33        if path == "/context":34            self.send_text_response(request_context["context"].serialize())35        if path == "/event":36            if "page" in parameters:37                page = int(parameters["page"][0])38            else:39                page = 040            self.send_text_response(get_logger().get_events(page=page,41                                                            count=60))42    def do_POST(self):43        content_len = int(self.headers.get('content-length', 0))44        payload = self.rfile.read(content_len)45        command = json.loads(payload.decode("UTF-8"))46       47        try:48           request_context["core"].execute(**command)49           self.send_text_response("ok")50        except Exception as e:51           self.send_text_response("error:%s" % str(e))52    def send_text_response(self, response):53        self.send_response(200)54        self.end_headers()55        if isinstance(response, dict) or isinstance(response, list):56            response = json.dumps(response, indent=3)57        self.wfile.write(bytes(response, 'UTF-8'))58    def log_message(self, format, *args):59        pass60class MyTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):61    def server_bind(self):62        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)63        self.socket.bind(self.server_address)64class Webserver(threading.Thread):65    def __init__(self, core, context):66        request_context.update({...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.