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

How to use _save_message_to_file method in avocado

Best Python code snippet using avocado_python

messages.py

Source:messages.py

...197        else:198            data = self._split_complete_lines(message.get("log", b""))199        if data:200            data = self._tag + self._tag.join(data)201            self._save_message_to_file(202                DEFAULT_LOG_FILE, data, task, message.get("encoding")203            )204    @staticmethod205    def _message_to_line(message, encoding):206        """207        Converts the message to string.208        When the message doesn't end with a new line, the new line is added.209        :param message: message for decoding210        :type message: bytes211        :param encoding: encoding of the message212        :type encoding: str213        :return: encoded message with new line character214        :rtype: str215        """216        message = message.decode(encoding)217        if not message.endswith("\n"):218            message = f"{message}\n"219        return message220    @staticmethod221    def _save_message_to_file(filename, buff, task, encoding=None):222        """223        Method for saving messages into the file224        It can decode and save messages.The message will be decoded when225        encoding is not None. When the decoded message doesn't end with a new226        line the new line will be added. Every message is saved in the append227        mode.228        :param filename: name of the file229        :type filename: str230        :param buff: message to be saved231        :type buff: bytes232        :param task: message related task.233        :type task: :class:`avocado.core.nrunner.Task`234        :param encoding: encoding of buff, default is None235        :type encoding: str236        """237        def _save_to_file(file_name, mode):238            with open(file_name, mode) as fp:  # pylint: disable=W1514239                fp.write(buff)240        file = os.path.join(task.metadata["task_path"], filename)241        if encoding:242            buff = BaseRunningMessageHandler._message_to_line(buff, encoding)243            _save_to_file(file, "a")244        else:245            _save_to_file(file, "ab")246class LogMessageHandler(BaseRunningMessageHandler):247    """248    Handler for log message.249    It will save the log to the debug.log file in the task directory.250    :param status: 'running'251    :param type: 'log'252    :param log: log message253    :type log: string254    :param time: Time stamp of the message255    :type time: float256    example: {'status': 'running', 'type': 'log', 'log': 'log message',257             'time': 18405.55351474}258    """259    _tag = b"[stdlog] "260    def handle(self, message, task, job):261        """Logs a textual message to a file.262        This assumes that the log message will not contain a newline, and thus263        one is explicitly added here.264        """265        if task.metadata.get("logfile") is None:266            task.metadata["logfile"] = os.path.join(267                task.metadata["task_path"], "debug.log"268            )269        self._save_to_default_file(message, task)270class StdoutMessageHandler(BaseRunningMessageHandler):271    """272    Handler for stdout message.273    It will save the stdout to the stdout and debug file in the task directory.274    :param status: 'running'275    :param type: 'stdout'276    :param log: stdout message277    :type log: bytes278    :param encoding: optional value for decoding messages279    :type encoding: str280    :param time: Time stamp of the message281    :type time: float282    example: {'status': 'running', 'type': 'stdout', 'log': 'stdout message',283             'time': 18405.55351474}284    """285    _tag = b"[stdout] "286    def handle(self, message, task, job):287        self._save_to_default_file(message, task)288        self._save_message_to_file(289            "stdout", message["log"], task, message.get("encoding", None)290        )291class StderrMessageHandler(BaseRunningMessageHandler):292    """293    Handler for stderr message.294    It will save the stderr to the stderr and debug file in the task directory.295    :param status: 'running'296    :param type: 'stderr'297    :param log: stderr message298    :type log: bytes299    :param encoding: optional value for decoding messages300    :type encoding: str301    :param time: Time stamp of the message302    :type time: float303    example: {'status': 'running', 'type': 'stderr', 'log': 'stderr message',304             'time': 18405.55351474}305    """306    _tag = b"[stderr] "307    def handle(self, message, task, job):308        self._save_to_default_file(message, task)309        self._save_message_to_file(310            "stderr", message["log"], task, message.get("encoding", None)311        )312class WhiteboardMessageHandler(BaseRunningMessageHandler):313    """314    Handler for whiteboard message.315    It will save the stderr to the whiteboard file in the task directory.316    :param status: 'running'317    :param type: 'whiteboard'318    :param log: whiteboard message319    :type log: bytes320    :param encoding: optional value for decoding messages321    :type encoding: str322    :param time: Time stamp of the message323    :type time: float324    example: {'status': 'running', 'type': 'whiteboard',325             'log': 'whiteboard message', 'time': 18405.55351474}326    """327    def handle(self, message, task, job):328        encoding = message.get("encoding", "utf-8")329        whiteboard = task.metadata.get("whiteboard", "")330        whiteboard += message["log"].decode(encoding)331        task.metadata["whiteboard"] = whiteboard332        self._save_message_to_file("whiteboard", message["log"], task, encoding)333class FileMessageHandler(BaseRunningMessageHandler):334    """335    Handler for file message.336    In task directory will save log into the runner specific file. When the337    file doesn't exist, the file will be created. If the file exist,338    the message data will be appended at the end.339    :param status: 'running'340    :param type: 'file'341    :param path: relative path to the file. The file will be created under342                the Task directory and the absolute path will be created343                as `absolute_task_directory_path/relative_file_path`.344    :type path: string345    :param log: data to be saved inside file346    :type log: bytes347    :param time: Time stamp of the message348    :type time: float349    example: {'status': 'running', 'type': 'file', 'path':'foo/runner.log',350             'log': 'this will be saved inside file',351             'time': 18405.55351474}352    """353    def handle(self, message, task, job):354        filename = os.path.relpath(os.path.join("/", message["path"]), "/")355        file = os.path.join(task.metadata["task_path"], filename)356        if not os.path.exists(file):357            os.makedirs(os.path.dirname(file), exist_ok=True)358        self._save_message_to_file(359            filename, message["log"], task, message.get("encoding", None)360        )361class OutputMessageHandler(BaseRunningMessageHandler):362    """363    Handler for displaying messages in UI.364    It will show the message content in avocado UI.365    :param status: 'running'366    :param type: 'output'367    :param log: output message368    :type log: bytes369    :param encoding: optional value for decoding messages370    :type encoding: str371    :param time: Time stamp of the message372    :type time: float...

message_sender.py

Source:message_sender.py

...33            print('Keys are not generated')34            return35        if host:36            path = 'files/message.txt'37            self._save_message_to_file(path)38            message_file = File(path)39            message_file.encrypt(self._key.key, self._iv.key,40                                 mode=mode, progress_func=self._progress_func)41            time.sleep(.1)42            send_thread = SendThread(43                message_file, mode=mode, host=host, show_progress_func=self._progress_func)44            send_thread.start()45        else:46            print('You have to specify receiver IP address')47    def _save_message_to_file(self, path):48        """49        Save message data to file given by a path50        """51        message = self._text_input.get('1.0', tk.END)52        with open(path, 'w') as file:53            file.write(message)54    def set_keys(self, key, iv):55        self._key = key...

sample_pipeline.py

Source:sample_pipeline.py

...14"""Sample Kubeflow pipeline."""15import kfp16from kfp.v2 import dsl17@dsl.component(base_image="python:3.10", packages_to_install=["cloudpathlib==0.10.0"])18def _save_message_to_file(message: str, gcs_filepath: str) -> None:19    """Saves a given message to a given file in GCS."""20    import cloudpathlib as cpl21    with cpl.CloudPath(gcs_filepath).open("w") as fp:22        fp.write(message)23@kfp.dsl.pipeline(name="sample-pipeline")24def pipeline(message: str, gcs_filepath: str) -> None:25    """Sample Kubeflow pipeline definition."""...

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.