Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.