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

How to use do_shutdown method in localstack

Best Python code snippet using localstack_python

tcp_server.py

Source:tcp_server.py

1# -*- coding: utf-8 -*-2"""3    proxy.py4    ~~~~~~~~5    â¡â¡â¡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on6    Network monitoring, controls & Application development, testing, debugging.7    :copyright: (c) 2013-present by Abhinav Singh and contributors.8    :license: BSD, see LICENSE for more details.9"""10from abc import abstractmethod11import socket12import selectors13from typing import Dict, Any, Optional14from proxy.core.acceptor import Work15from proxy.common.types import Readables, Writables16class BaseTcpServerHandler(Work):17    """BaseTcpServerHandler implements Work interface.18    An instance of BaseTcpServerHandler is created for each client19    connection.  BaseServerHandler lifecycle is controlled by20    Threadless core using asyncio.21    BaseServerHandler ensures that pending buffers are flushed22    before client connection is closed.23    Implementations must provide:24    a) handle_data(data: memoryview)25    c) (optionally) intialize, is_inactive and shutdown methods26    """27    def __init__(self, *args: Any, **kwargs: Any) -> None:28        super().__init__(*args, **kwargs)29        self.must_flush_before_shutdown = False30        print('Connection accepted from {0}'.format(self.client.addr))31    @abstractmethod32    def handle_data(self, data: memoryview) -> Optional[bool]:33        """Optionally return True to close client connection."""34        pass    # pragma: no cover35    def get_events(self) -> Dict[socket.socket, int]:36        events = {}37        # We always want to read from client38        # Register for EVENT_READ events39        if self.must_flush_before_shutdown is False:40            events[self.client.connection] = selectors.EVENT_READ41        # If there is pending buffer for client42        # also register for EVENT_WRITE events43        if self.client.has_buffer():44            if self.client.connection in events:45                events[self.client.connection] |= selectors.EVENT_WRITE46            else:47                events[self.client.connection] = selectors.EVENT_WRITE48        return events49    def handle_events(50            self,51            readables: Readables,52            writables: Writables) -> bool:53        """Return True to shutdown work."""54        do_shutdown = False55        if self.client.connection in readables:56            try:57                data = self.client.recv()58                if data is None:59                    # Client closed connection, signal shutdown60                    print(61                        'Connection closed by client {0}'.format(62                            self.client.addr))63                    do_shutdown = True64                else:65                    r = self.handle_data(data)66                    if isinstance(r, bool) and r is True:67                        print(68                            'Implementation signaled shutdown for client {0}'.format(69                                self.client.addr))70                        if self.client.has_buffer():71                            print(72                                'Client {0} has pending buffer, will be flushed before shutting down'.format(73                                    self.client.addr))74                            self.must_flush_before_shutdown = True75                        else:76                            do_shutdown = True77            except ConnectionResetError:78                print(79                    'Connection reset by client {0}'.format(80                        self.client.addr))81                do_shutdown = True82        if self.client.connection in writables:83            print('Flushing buffer to client {0}'.format(self.client.addr))84            self.client.flush()85            if self.must_flush_before_shutdown is True:86                do_shutdown = True87            self.must_flush_before_shutdown = False88        if do_shutdown:89            print(90                'Shutting down client {0} connection'.format(91                    self.client.addr))...

connect.py

Source:connect.py

1import socket2import ssl3import sys4MAX_READ_BUFFER_LEN = 40965HOST = '127.0.0.1'6PORT = 91507def listen_for_connection(context):8    if 'listener_socket' in context:9        context['listener_socket'].listen()10        conn, addr = context['listener_socket'].accept()11        print('Connected by', addr)12        return conn13def listener_open():14    listener_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)15    listener_socket.bind((HOST, PORT))16    return {'listener_socket': listener_socket}17def get_connect_context(context):18    if context == None:19        print("Error: Check if OR is online.")20        sys.exit(1)21    if 'link' in context:22        context = context['link']23    assert 'tcp_socket' in context24    return context25def tcp_request(ip, port, request_bytes, do_shutdown, max_response_len=MAX_READ_BUFFER_LEN):26    context = tcp_open(ip, port)27    tcp_write(context, request_bytes)28    response_bytes = tcp_read(context, max_response_len)29    tcp_close(context, do_shutdown)30    return response_bytes31def tcp_open(ip, port):32    tcp_sock = socket.create_connection((ip, port))33    return {'tcp_socket': tcp_sock}34def tcp_write(context, request_bytes):35    context = get_connect_context(context)36    context['tcp_socket'].sendall(request_bytes)37def tcp_read(context, max_response_len=MAX_READ_BUFFER_LEN):38    context = get_connect_context(context)39    return bytearray(context['tcp_socket'].recv(max_response_len))40def tcp_close(context, do_shutdown):41    context = get_connect_context(context)42    if do_shutdown:43        try:44            context['tcp_socket'].shutdown(socket.SHUT_RDWR)45        except socket.error as e:46            print("Socket error '{}' during shutdown".format(e))47    context['tcp_socket'].close()48def tls_open(ip, port):49    context = tcp_open(ip, port)50    ssl_context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS)51    ssl_sock = ssl_context.wrap_socket(context['tcp_socket'], server_hostname=ip)52    ssl_sock.do_handshake()53    context.update({'ssl_socket': ssl_sock})54    return context55def tls_write(context, request_bytes):56    context = get_connect_context(context)57    context['ssl_socket'].sendall(request_bytes)58def tls_read(context, max_response_len=MAX_READ_BUFFER_LEN):59    context = get_connect_context(context)60    context['ssl_socket'].settimeout(3)61    return bytearray(context['ssl_socket'].recv(max_response_len))62def tls_close(context, do_shutdown=True):63    context = get_connect_context(context)64    if do_shutdown:65        try:66            context['ssl_socket'].shutdown(socket.SHUT_RDWR)67        except socket.error as e:68            # A "Socket is not connected" error here is harmless69            print("Socket error '{}' during shutdown".format(e))70    context['ssl_socket'].close()71def tls_request(ip, port, request_bytes, do_shutdown, max_response_len=MAX_READ_BUFFER_LEN):72    context = tls_open(ip, port)73    tls_write(context, request_bytes)74    response = tls_read(context, max_response_len)...

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.