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

How to use recv_from_client method in locust

Best Python code snippet using locust

test_web_client.py

Source:test_web_client.py

...10    @patch('web_client.logger')   11    def test_recv_from_client_invalid(self, logger):12        """test if recv_from_client ignores non-json"""13        data = 'ie23\r\n'14        self.wc.recv_from_client(data)15        logger.warning.assert_called_with('[AnyName]: Not a JSON! Ignoring...')16        self.assertFalse(logger.info.called)17    18    @patch.object(web_client.WebClient, '_connect')    19    @patch('web_client.logger')20    def test_recv_from_client_split(self, logger, conn):21        """test if recv_from_client accept splitted json; cmd = connect"""22        self.wc.recv_from_client('{"cmd":"c')23        self.wc.recv_from_client('onnect","param":"abc"}\r\n')24        logger.info.assert_called_with('[AnyName]: cmd = connect')25        conn.assert_called_with(cmd='connect', param='abc')26    27    @patch('time.time')28    @patch('uuid.uuid4')29    @patch.object(web_client.WebClient, '_put_answer_in_queue')30    @patch('web_client.SSHChannel')31    def test_connect(self, ssh, put_ans, m_uuid, m_time):32        """test if _connect() tries to establish ssh session and returns correct result"""33        args = {"host":"abc", "port":666, 34                   "user":"devil", "secret":"hell"}35        m_time.return_value=123436        ssh_conn = ssh.return_value37        m_uuid.return_value='abc-def'38        #positive39        res = {"cmd":"any", "res":"ok", "conn_id":'abc-def'}40        self.wc._connect(cmd = 'any', **args)41        ssh.assert_called_with(**args)42        put_ans.assert_called_with(res)43        self.assertEqual(self.wc._sessions['abc-def'], [ssh_conn, 1234]) 44        45        #negative46        ssh.reset_mock()47        m_uuid.reset_mock()48        ssh.side_effect = Exception()49        res = {"cmd":"any", "res":"error"}50        self.wc._connect(cmd = 'any', **args)51        ssh.assert_called_with(**args)52        self.assertItemsEqual(m_uuid.call_args_list, [])53        put_ans.assert_called_with(res)        54    55    @patch('time.time')56    @patch('uuid.uuid4')57    @patch('web_client.PlugLess')58    def test_log_open(self, m_plug, m_uuid, m_time):59        """ create new log session on log_open call """60        conn = Mock()61        log = m_plug.return_value62        args = {'conn_id':'abc-def', 'cmd':'cmd','some_arg':'some_val'}63        self.wc._sessions = {'abc-def':[conn, 100]}64        m_uuid.return_value='aaa-bbb'65        m_time.return_value = 33366        self.wc._log_open(**args)67        m_plug.assert_called_with(ssh=conn, **args)68        self.assertEqual(self.wc._log_sessions['aaa-bbb'], 69                        [log, web_client.WebClient.PL_ACTIVE, 'cmd', 'abc-def'])70        self.assertEqual(self.wc._sessions['abc-def'], [conn, 333])                71        self.assertIn(log, self.wc._sock_read_fd)72    73    @patch.object(web_client.WebClient, '_log_open')74    @patch('web_client.logger')75    def test_log_open_called(self, logger, log_open):76        """log_open called only if correct conn_id provided"""77        self.wc._log_sessions['aaa-111']=[]78        self.wc.recv_from_client('{"cmd":"log_open","param":"aaa-111"}\r\n')79        self.assertItemsEqual(log_open.call_args_list, []) 80        self.wc.recv_from_client('{"cmd":"log_open","log_id":"aaa-111"}\r\n')81        self.assertItemsEqual(log_open.call_args_list, []) 82        self.wc.recv_from_client('{"cmd":"log_open","conn_id":"aaa-111"}\r\n')83        self.assertItemsEqual(log_open.call_args_list, []) 84        self.wc._sessions['aaa-112']=[]85        self.wc.recv_from_client('{"cmd":"log_open","param":"aaa-112"}\r\n')86        self.assertItemsEqual(log_open.call_args_list, []) 87        self.wc.recv_from_client('{"cmd":"log_open","log_id":"aaa-112"}\r\n')88        self.assertItemsEqual(log_open.call_args_list, []) 89        self.wc.recv_from_client('{"cmd":"log_open","conn_id":"aaa-112"}\r\n')90        log_open.assert_called_with(cmd='log_open', conn_id='aaa-112')91    def test_disconnect(self):92        """test if _disconnect() closes all log channels and ssh channel"""93        s1 = Mock()94        s2 = Mock()95        self.wc._sessions ={'aaa':[s1,1],96                            'bbb':[s2,2]}97        log = Mock()98        self.wc._log_sessions ={'000':[log,None,None, 'aaa'],99                                '001':[log,None,None, 'aaa'],100                                '002':[log,None,None, 'bbb'],101                                '003':[log,None,None, 'ccc'],}102        self.wc._disconnect('aaa')103        self.assertDictEqual(self.wc._sessions, {'bbb':[s2,2]})104        self.assertDictEqual(self.wc._log_sessions,105                   {'002':[log,None,None, 'bbb'],106                    '003':[log,None,None, 'ccc']})107        s1.close.assert_called_once_with()108        self.assertItemsEqual(s2.close.call_args_list, [])109        self.assertEqual(log.close.call_count,2)110    111    @patch.object(web_client.WebClient, '_disconnect')112    @patch('web_client.logger')113    def test_disconnect_called(self, logger, close):114        """_disconnect called only if correct conn_id provided"""115        self.wc._log_sessions['aaa-111']=[]116        self.wc.recv_from_client('{"cmd":"close","param":"aaa-111"}\r\n')117        self.assertItemsEqual(close.call_args_list, []) 118        self.wc.recv_from_client('{"cmd":"close","log_id":"aaa-111"}\r\n')119        self.assertItemsEqual(close.call_args_list, []) 120        self.wc.recv_from_client('{"cmd":"close","conn_id":"aaa-111"}\r\n')121        self.assertItemsEqual(close.call_args_list, [])122        self.wc._sessions['aaa-112']=[]123        self.wc.recv_from_client('{"cmd":"close","param":"aaa-112"}\r\n')124        self.assertItemsEqual(close.call_args_list, []) 125        self.wc.recv_from_client('{"cmd":"close","log_id":"aaa-112"}\r\n')126        self.assertItemsEqual(close.call_args_list, []) 127        self.wc.recv_from_client('{"cmd":"close","conn_id":"aaa-112"}\r\n')128        close.assert_called_with('aaa-112')129    @patch.object(web_client.WebClient, '_log_cmd')130    @patch('web_client.logger')131    def test_log_cmd_called(self, logger, log_cmd):132        """_log_cmd called only if correct log_id provided"""133        self.wc._sessions['aaa-111']=[]134        self.wc.recv_from_client('{"cmd":"log_page","param":"aaa-111"}\r\n')135        self.assertItemsEqual(log_cmd.call_args_list, [])136        self.wc.recv_from_client('{"cmd":"log_page","log_id":"aaa-111"}\r\n')137        self.assertItemsEqual(log_cmd.call_args_list, [])138        self.wc.recv_from_client('{"cmd":"log_page","conn_id":"aaa-111"}\r\n')139        self.assertItemsEqual(log_cmd.call_args_list, [])140        self.wc._log_sessions['aaa-112']=[]141        self.wc.recv_from_client('{"cmd":"log_page","param":"aaa-112"}\r\n')142        self.assertItemsEqual(log_cmd.call_args_list, [])143        self.wc.recv_from_client('{"cmd":"log_page","conn_id":"aaa-112"}\r\n')144        self.assertItemsEqual(log_cmd.call_args_list, [])145        self.wc.recv_from_client('{"cmd":"log_page","log_id":"aaa-112"}\r\n')146        log_cmd.assert_called_with(cmd='log_page',log_id='aaa-112')147    def test_put_answer_in_queue(self):148        self.assertEquals(self.wc._out_buff, [])149        self.assertEquals(self.wc._sock_write_fd,[])150        data = {'abc':123}151        self.wc._put_answer_in_queue(data)152        self.assertEquals(self.wc._sock_write_fd,[self.wc.sock])153        self.assertEquals(self.wc._out_buff, ['{"abc": 123}\r\n'])154    @patch.object(web_client, 'OUT_BUFF_SIZE')155    def test_put_answer_in_queue_long(self, buff):156        web_client.OUT_BUFF_SIZE = 4 # mock is used here to correctly restore constant after all157        self.assertEquals(self.wc._out_buff, [])158        self.assertEquals(self.wc._sock_write_fd,[])159        data = {'abc':123}...

server.py

Source:server.py

...9server_socket.bind((host_ip, port))10server_socket.listen(5)11CHUNK_SIZE = 102412SONG_LIST_FILE = "allsongs.json"13def recv_from_client(client_socket):14    # receive arguments from client15    res = client_socket.recv(CHUNK_SIZE)16    data = None17    while not data and not res:18        data = client_socket.recv(CHUNK_SIZE)19        res += data20    return res.decode()21def audio_stream(path, client_socket):22    # stream audio files to client23    wf = wave.open(path, 'rb')24    print("Start streaming")25    data = wf.readframes(CHUNK_SIZE)26    while data != b"":27        client_socket.send(data)28        data = wf.readframes(CHUNK_SIZE)29    print("End streaming")30    wf.close()31def main():32    while True:33        print("Waiting for client...")34        client_socket,addr = server_socket.accept()35        print('SERVER listening at',(host_ip, port))36        with open(SONG_LIST_FILE) as data_file:37            json_file_data = json.load(data_file)38        song_list = []39        song_list.extend(json_file_data)40        # receive operation type from client41        cmd = recv_from_client(client_socket)42        43        if cmd == '3':44            ydl_opts = {45			'outtmpl': 'songs/%(id)s.%(ext)s',46			'default_search': "ytsearch5",47			'format': 'bestaudio/best',48			'postprocessors': [{49				'key': 'FFmpegExtractAudio',50				'preferredcodec': 'wav',51				'preferredquality': '192',52				}],53			}54            55            name = recv_from_client(client_socket)56            57            with youtube_dl.YoutubeDL(ydl_opts) as ydl:58                info = ydl.extract_info(name, download=False)59                60            s = ""61            for i,k in enumerate(info['entries']):62                s += str(i+1) + "." + info['entries'][i]['title'] + "\n"63            client_socket.sendall(s.encode())64            # receive download information sent from client65            download = int(recv_from_client(client_socket))66			67            song_download_url = info['entries'][download-1]['webpage_url']68            song_download_id = info['entries'][download-1]['id']69            song_download_name = info['entries'][download-1]['title']70            song_download_obj = {"name": song_download_name, "id":song_download_id}71            song_list.append(song_download_obj)72            with youtube_dl.YoutubeDL(ydl_opts) as ydl:73                ydl.download([song_download_url])74                client_socket.send(b"DONE")75			76            with open(SONG_LIST_FILE, 'w') as outfile:77                json.dump( song_list, outfile)78        elif cmd == '2':79            songs_num = len(song_list)80            if songs_num == 0:81                client_socket.sendall('Song list is empty'.encode())82            else:83                msg = ""84                for i in range(songs_num):85                    msg += str(i+1) + '.' + song_list[i]["name"] + "\n"86                client_socket.sendall(msg.encode())87                play_choice = recv_from_client(client_socket)88                89                play_choice = int(play_choice)90                path = 'songs/' + song_list[play_choice - 1]["id"] + '.wav'91                client_socket.sendall(path.encode())92                audio_stream(path, client_socket)93        elif cmd == '1':94            songs_num = len(song_list)95            if songs_num == 0:96                client_socket.sendall('Song list is empty'.encode())97            else:98                songs = ""99                for i in range(songs_num):100                    songs += str(i+1) + "." + song_list[i]["name"] + "\n"101                client_socket.sendall(songs.encode())...

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.