How to use _save_message_to_file method in avocado

Best Python code snippet using avocado_python

messages.py

Source:messages.py Github

copy

Full Screen

...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...

Full Screen

Full Screen

message_sender.py

Source:message_sender.py Github

copy

Full Screen

...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...

Full Screen

Full Screen

sample_pipeline.py

Source:sample_pipeline.py Github

copy

Full Screen

...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."""...

Full Screen

Full Screen

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:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run avocado 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful