KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _put_message method in localstack
Best Python code snippet using localstack_python
test_base_api.py
Source:test_base_api.py 
...23 if self.queue:24 self.queue.close()25 self.delete_queue(self.queue_name)26 super(TestGet, self).tearDown()27 def _put_message(self):28 md = pymqi.MD()29 self.queue.put(self.message, md)30 return md31 ###########################################################################32 #33 # Real Tests start here34 #35 ###########################################################################36 def test_get_nontruncated(self):37 """Test nontruncated without buffer."""38 self._put_message()39 md_get = pymqi.MD()40 message = self.queue.get(None, md_get)41 self.assertEqual(self.message, message)42 def test_get_nontruncated_0(self):43 """Test nontruncated with zerro buffer length."""44 self._put_message()45 md_get = pymqi.MD()46 try:47 self.queue.get(0, md_get)48 except pymqi.MQMIError as ex:49 self.assertEqual(ex.reason, pymqi.CMQC.MQRC_TRUNCATED_MSG_FAILED)50 self.assertEqual(ex.original_length, # pylint: disable=no-member51 len(self.message))52 def test_get_nontruncated_short(self):53 """Test nontruncated with short buffer."""54 md_put = self._put_message()55 md_get = pymqi.MD()56 try:57 self.queue.get(self.buffer_length, md_get)58 except pymqi.MQMIError as ex:59 self.assertEqual(ex.reason, pymqi.CMQC.MQRC_TRUNCATED_MSG_FAILED)60 self.assertEqual(ex.original_length, # pylint: disable=no-member61 len(self.message))62 def test_get_nontruncated_enough(self):63 """Test nontruncated with big enough buffer."""64 md_put = self._put_message()65 md_get = pymqi.MD()66 message = self.queue.get(len(self.message), md_get)67 self.assertEqual(self.message, message)68 def test_get_truncated(self):69 """Test truncated without buffer."""70 self._put_message()71 gmo = pymqi.GMO()72 gmo.Options = pymqi.CMQC.MQGMO_ACCEPT_TRUNCATED_MSG73 md_get = pymqi.MD()74 try:75 self.queue.get(0, md_get, gmo)76 except pymqi.MQMIError as ex:77 self.assertEqual(ex.reason, pymqi.CMQC.MQRC_TRUNCATED_MSG_ACCEPTED)78 self.assertEqual(ex.message, b'') # pylint: disable=no-member79 self.assertEqual(ex.original_length, # pylint: disable=no-member80 len(self.message))81 def test_get_truncated_0(self):82 """Test truncated with zero buffer length."""83 self._put_message()84 gmo = pymqi.GMO()85 gmo.Options = pymqi.CMQC.MQGMO_ACCEPT_TRUNCATED_MSG86 md_get = pymqi.MD()87 try:88 self.queue.get(0, md_get, gmo)89 except pymqi.MQMIError as ex:90 self.assertEqual(ex.reason, pymqi.CMQC.MQRC_TRUNCATED_MSG_ACCEPTED)91 self.assertEqual(ex.message, b'') # pylint: disable=no-member92 self.assertEqual(ex.original_length, # pylint: disable=no-member93 len(self.message))94 def test_get_truncated_short(self):95 """Test truncated with short buffer."""96 self._put_message()97 gmo = pymqi.GMO()98 gmo.Options = pymqi.CMQC.MQGMO_ACCEPT_TRUNCATED_MSG99 md_get = pymqi.MD()100 try:101 self.queue.get(self.buffer_length, md_get, gmo)102 except pymqi.MQMIError as ex:103 self.assertEqual(ex.reason, pymqi.CMQC.MQRC_TRUNCATED_MSG_ACCEPTED)104 self.assertEqual(ex.message, # pylint: disable=no-member105 self.message[:self.buffer_length])106 self.assertEqual(ex.original_length, # pylint: disable=no-member107 len(self.message))108 def test_get_truncated_enough(self):109 """Test truncated with big buffer."""110 self._put_message()111 gmo = pymqi.GMO()112 gmo.Options = pymqi.CMQC.MQGMO_ACCEPT_TRUNCATED_MSG113 md_get = pymqi.MD()114 message = self.queue.get(len(self.message), md_get, gmo)115 self.assertEqual(self.message, message)116 def test_get_nontruncated_big_msg(self):117 """Test get nontruncated big message"""118 md_put = pymqi.MD()119 if version_info.major >= 3:120 self.queue.put(bytes(4097), md_put)121 else:122 self.queue.put(bytes(b'\0'*4097), md_put)123 md_get = pymqi.MD()124 message = self.queue.get(None, md_get)...
window.py
Source:window.py
...39 self.start_engine()40 logging.info("Started the Engine")41 def start_engine(self):42 """Starts the engine in the listening `EngineProcess`"""43 self._put_message(StartEngine())44 def close(self, force_close: bool = False):45 """46 Closes the OpenGLWindow47 Sends a message to the Engine process to close, and waits for it to close48 Args:49 force_close (bool): Whether to exit the main_loop are wait for the user to close the window manually50 """51 if not self._initialized:52 return53 self._put_message(CloseProcess(force_close=force_close))54 self._engine_process.join()55 self._engine_process = None56 self._mp_context = None57 self._queue.close()58 self._queue = None59 self._initialized = False60 def delete_model(self, model_id: int):61 """Deletes the model identified by `model_id`"""62 self._put_message(DeleteModel(model_id=model_id))63 def set_lines(self,64 model_id: int,65 lines_data: np.ndarray,66 default_color: np.ndarray = np.zeros((1, 3), dtype=np.float32),67 line_width: float = 1.0,68 line_color: Optional[np.ndarray] = None):69 """70 Sets/Updates line models to be rendered in the OpenGL Window71 Args:72 model_id (int): The identifying integer of the model for the engine (will override previous models)73 lines_data (np.ndarray): The array of lines to render `(N, 2, 3)`74 default_color (np.ndarray): The default color of the lines (black by default)75 line_width (float): The width of the lines rendered76 line_color (np.ndarray): Optionally the color for each line `(N, 3)`77 """78 check_sizes(lines_data, [-1, 2, 3])79 check_sizes(default_color, [1, 3])80 if line_color is not None:81 check_sizes(line_color, [lines_data.shape[0], 3])82 model_data = LinesModelData(default_color=default_color,83 line_data=lines_data,84 line_width=line_width,85 line_color=line_color)86 self._put_message(UpdateModel(model_id=model_id, model=model_data))87 def set_voxels(self,88 model_id: int,89 points: np.ndarray,90 voxel_size: float = 1.0,91 default_color: np.ndarray = np.zeros((1, 3), dtype=np.float32),92 line_width: float = 1.0):93 """94 Sets/Updates voxel models to be rendered in the OpenGL Window95 Args:96 model_id (int): The identifying integer of the model for the engine (will override previous models)97 points (np.ndarray): The array of points to render in a voxel grid `(N, 3)`98 voxel_size (float): The size of the voxels to render99 default_color (np.ndarray): The default color of the lines (black by default)100 line_width (float): The width of the lines rendered101 """102 check_sizes(points, [-1, 3])103 check_sizes(default_color, [1, 3])104 model_data = VoxelsModelData(default_color=default_color,105 voxel_points=points,106 voxel_size=voxel_size,107 line_width=line_width)108 self._put_message(UpdateModel(model_id, model=model_data))109 def set_ellipses(self,110 model_id: int,111 centers: Optional[np.ndarray] = None,112 covariances: Optional[np.ndarray] = None,113 colors: Optional[np.ndarray] = None,114 default_color: np.ndarray = np.array([[1.0, 0.0, 0.0]], dtype=np.float32)):115 """116 Sets/Updates low-polygons ellipses models to be rendered in the OpenGL Window117 Args:118 model_id (int): The identifying integer of the model for the engine (will override previous models)119 centers (np.ndarray): The centers of the ellipses to render (if None, a unique sphere will be rendered)120 `(N, 3)`121 covariances (np.ndarray): The symmetric matrices defining the shape of the ellipse122 (typically a covariance matrix) `(N, 3, 3)`123 colors (np.ndarray): The colors of the ellipses124 default_color (np.ndarray): The default color of the lines (red by default)125 """126 check_sizes(centers, [-1, 3])127 check_sizes(covariances, [-1, 3, 3])128 check_sizes(default_color, [1, 3])129 model_data = EllipsesModelData(default_color=default_color, colors=colors,130 means=centers, covariances=covariances)131 self._put_message(UpdateModel(model_id, model=model_data))132 def set_cameras(self,133 model_id: int,134 positions: np.ndarray,135 default_color: np.ndarray = np.zeros((1, 3), dtype=np.float32),136 line_width: float = 1.0,137 scale: float = 1.0):138 """139 Sets/Updates camera models to be rendered in the OpenGL Window140 A camera model is a simple oriented pyramid shape indicating the position and orientation of a camera141 Args:142 model_id (int): The identifying id of the model for the engine (will override previous models)143 positions (np.ndarray): The positions of the cameras to indicate `(N, 4, 4)`144 default_color (np.ndarray): The default color of the cameras (black by default)145 line_width (float): The width of the lines of the model146 scale (float): The scale of the camera model147 """148 check_sizes(positions, [-1, 4, 4])149 check_sizes(default_color, [-1, 3])150 model_data = CamerasModelData(instance_model_to_world=positions.astype(np.float32),151 default_color=default_color,152 width=line_width,153 camera_size=scale)154 self._put_message(UpdateModel(model_id=model_id, model=model_data))155 def set_poses(self,156 model_id: int,157 positions: np.ndarray,158 line_width: float = 1.0,159 scale: float = 1.0):160 """161 Sets/Updates the poses model to be rendered in the OpenGL Window162 """163 check_sizes(positions, [-1, 4, 4])164 model_data = PosesModelData(instance_model_to_world=positions.astype(np.float32),165 width=line_width,166 scale=scale)167 self._put_message(UpdateModel(model_id=model_id, model=model_data))168 def set_pointcloud(self,169 model_id: int,170 pointcloud: np.ndarray,171 color: Optional[np.ndarray] = None,172 default_color: Optional[np.ndarray] = None,173 point_size: int = 2):174 """175 Sets/Updates a point cloud to be rendered in the OpenGLWindow176 Args:177 model_id (int): The identifying id of the model for the engine (will override previous models)178 pointcloud (np.ndarray): The Point Cloud geometric data `(N, 3)`179 color (np.ndarray): The (Optional) color array describing the color of each point `(N,3)`180 default_color (np.ndarray): The (Optional) default color for the point cloud `(1, 3)`181 point_size (int): The pixel size of the points displayed by OpenGL182 """183 check_sizes(pointcloud, [-1, 3])184 if color is not None:185 check_sizes(color, [pointcloud.shape[0], 3])186 if default_color is not None:187 check_sizes(default_color, [1, 3])188 model_data = PointCloudModelData(xyz=pointcloud,189 color=color,190 default_color=default_color,191 point_size=point_size)192 self._put_message(UpdateModel(model_id=model_id, model=model_data))193 def update_camera(self, camera_pose: np.ndarray):194 """Updates the camera position for the view rendered in the screen"""195 check_sizes(camera_pose, [4, 4])196 self._put_message(UpdateCameraPosition(camera_position=camera_pose))197 # ------------------------------------------------------------------------------------------------------------------198 def _put_message(self, message: Message):199 """Puts the message Message in the Queue"""200 assert_debug(isinstance(message, Message))201 assert_debug(self._initialized, "Cannot Send messages, the EngineProcess is not started.")...
cache.py
Source:cache.py
...13 body = TextField(null=True)14 created_time = DateTimeField(default=datetime.datetime.now)15 class Meta:16 database = database17def _put_message(d):18 with database.execution_context() as ctx: # flake8: noqa19 msg = models.Message(d['id'], d['type'], d.get('body'))20 Message.create(21 id=msg.id,22 type=msg.type,23 body=json_encode(msg.body) if msg.body else None)24 return msg25def put_message(msg, async=True):26 if async:27 return executor.submit(_put_message, msg)28 else:29 return _put_message(msg)30def _get_messages_before(datetime):31 with database.execution_context() as ctx: # flake8: noqa32 messages = [33 models.Message(msg.id, msg.type,34 json_decode(msg.body) if msg.body else None)35 for msg in Message.select().where(Message.created_time < datetime)36 ]37 return messages38def get_messages_before(datetime, async=True):39 if async:40 return executor.submit(_get_messages_before, datetime)41 else:42 return _get_messages_before(datetime)43def _delete_messages_before(datetime):...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
