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

How to use _set_connection method in yandex-tank

Best Python code snippet using yandex-tank

http_socket_client.py

Source:http_socket_client.py

...5class SocketClient:6    def __init__(self, host, port):7        self.target_host = host8        self.target_port = port9    def _set_connection(self, timeout):10        start = int(time.time())11        end = start + timeout12        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)13        self.client.settimeout(4)14        status = 115        while int(time.time()) < end:16            try:17                status = self.client.connect_ex((self.target_host, self.target_port))18                if status == 0:19                    return status20            except Exception:21                pass22        if status != 0:23            raise ConnectionRefusedError24    def get(self, params):25        self._set_connection(4)26        request = 'GET {params} HTTP/1.1\r\n'.format(params=params) + \27                  'Host:{host}\r\n\r\n'.format(host=self.target_host + ':' + str(self.target_port))28        self.client.send(request.encode())29        total_data = []30        while True:31            data = self.client.recv(4096)32            if data:33                total_data.append(data.decode())34            else:35                break36        data = ''.join(total_data).splitlines()37        return json.loads(data[-1])38    def post(self, params, data):39        self._set_connection(4)40        body_bytes = data.encode()41        req = f'POST {params} HTTP/1.1\r\n' + \42              'Host: {host}\r\n'.format(host=self.target_host + ':' + str(self.target_port)) + \43              'Content-Type: application/json\r\n' + \44              'Content-Length: {length}\r\n\r\n'.format(length=len(body_bytes)) + \45              '{body}'.format(body=data)46        self.client.send(req.encode())47        total_data = []48        while True:49            data = self.client.recv(4096)50            if data:51                total_data.append(data.decode())52            else:53                break54        self.client.close()55        data = ''.join(total_data).splitlines()56        return json.loads(data[-1])57    def put(self, params, data):58        self._set_connection(4)59        body_bytes = data.encode()60        req = f'PUT {params} HTTP/1.1\r\n' + \61              'Host: {host}\r\n'.format(host=self.target_host + ':' + str(self.target_port)) + \62              'Content-Type: application/json\r\n' + \63              'Content-Length: {length}\r\n\r\n'.format(length=len(body_bytes)) + \64              '{body}'.format(body=data)65        self.client.send(req.encode())66        total_data = []67        while True:68            data = self.client.recv(4096)69            if data:70                total_data.append(data.decode())71            else:72                break...

db.py

Source:db.py

...20            self.config = copy.deepcopy(config)21        22        print('PostgreSQL Connection Info : ', self.config)23        self.psql_pool = None24        self._set_connection()25        26    def _set_connection(self):27        if self.psql_pool:28            return True29        30        try:31            # Create Connection Pools32            self.psql_pool = psycopg2.pool.SimpleConnectionPool(1, self.config['maxpool'],33                host=self.config['host'],34                port=self.config['port'],35                dbname=self.config['dbname'],36                user=self.config['user'],37                password=self.config['password'])38        except BaseException as err:39            print(f"Unexpected {err=}, {type(err)=}")40        41        if self.psql_pool:42            print('Successfully Connected')43        else:44            print('Failed to Connect')45        46        return self.psql_pool is not None47        48    @contextmanager49    def _get_conn(self):50        assert self._set_connection(), "PostgreSQL is not connected"51        52        conn = self.psql_pool.getconn()53        try:54            yield conn55        finally:56            self.psql_pool.putconn(conn)57    @contextmanager58    def _get_cursor(self, conn):59        cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)60        try:61            yield cur62        finally:63            cur.close()    64    65    def get_metadata(self):66        return {'is_alive': self._set_connection(), 'config': self.config}67    68    # Execute Query69    def query(self, sql, params = ()):70        with self._get_conn() as conn, self._get_cursor(conn) as cur:71            cur.execute(sql, params)72            return cur.fetchall()73    def execute(self, sql, params=()):74        with self._get_conn() as conn, self._get_cursor(conn) as cur:75            cur.execute(sql, params)...

transaction.py

Source:transaction.py

...33    def _get_connection(self):34        if not self.started:35            raise RuntimeError('Transaction not started! Use begin()')36        return self._connection37    def _set_connection(self, connection):38        self._connection = connection39    connection = property(_get_connection, _set_connection)40    def disable_autocommit(self):41        self.db.set_autocommit(False)42    def restore_autocommit(self):43        self.db.set_autocommit(self._autocommit)44    def commit(self, close_transaction=False):45        return self.db.commit(close_transaction=close_transaction)46    def rollback(self, close_transaction=False):...

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.