KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use resume_session method in localstack
Best Python code snippet using localstack_python
ServeurTcp.py
Source:ServeurTcp.py 
1from socket import socket, AF_INET, SOCK_STREAM, timeout2from threading import Thread3import json4from notifier import Notifier5import schedule6import time7import utility8import datetime910DEBUG = False1112class SyntheseRobot:13 OP = "Robot_OP"14 HS = "Robot_HS"15 ANTI_THEFT = "Robot_ANTI_THEFT"1617class ErrorLevels:18 OK = "OK"19 ERROR = "ERROR"2021class ErrorMessages:22 CLOSED = "Connection_closed"23 LOST = "Connection_lost"24 ABORTED = "Connection_aborted" #Pas utiliser pour l'instant à voir l'utilité2526class Server(Thread):2728 def __init__(self):29 Thread.__init__(self)30 self.socket = socket(AF_INET, SOCK_STREAM)31 self.socket.bind(("", 888))32 self.socket.settimeout(0.5)33 self.running = False34 self.client_pool = []35 self.getConfig()36 self.notifier = Notifier(self.tokens["telegram"])3738 def getConfig(self):39 with open('./config.json') as json_file:40 config = json.load(json_file)41 self.tokens = config["Tokens"]42 self.translate = config["Translate"]43 self.robots = config["Ip_Robot"]44 self.robots_language = config["Robot_Language"]45 self.clients = config["Robot_PhoneNumber"]4647 def stop(self):48 print("Server shutdown...")49 self.notifier.sendTelegramMsg(self.tokens["chat_id"],"Serveur de notification arrêté !",list(),False)50 if self.client_pool :51 for client in self.client_pool:52 client.close_connection()53 self.running = False5455 def sendNotification(self, robotAdresse, msg):56 self.getConfig()57 sn = self.robots[robotAdresse]58 language = self.robots_language[sn]59 if language not in self.translate["Supported Language"]:60 language = "en"61 clients = []62 if sn in self.clients:63 clients = self.clients[sn]64 65 self.notifier.sendNotifications(msg, clients, self.tokens, sn, self.translate, language)6667 return clients6869 def client_handling_stopped(self, client, error_level, error_msg, workTime=None, extraction=None, urlMap=""):70 msg = ""7172 if error_msg in [SyntheseRobot.HS,ErrorMessages.CLOSED,ErrorMessages.LOST]:73 clients = self.sendNotification(client.address[0],error_msg)74 msg += f"{self.robots[client.address[0]]} : {self.translate['Messages'][error_msg]['fr']}"75 76 if workTime is not None:77 workTimeS = str(workTime).split(".")[0]78 msg += f"\n{self.robots[client.address[0]]} : Temps de travail : {workTimeS}"79 if extraction is not None:80 msg += f"\n{self.robots[client.address[0]]} : Nombre d'extraction {extraction}"81 if msg != "":82 self.notifier.sendTelegramMsg(self.tokens["chat_id"],msg,clients,True,buttons={"Générer le pdf":f"pdf_{urlMap[1:]}"}, sn=self.robots[client.address[0]], url_map=urlMap)8384 self.client_pool = [client for client in self.client_pool if client.alive]8586 def send_anti_theft(self, client, leave_area : bool):87 if leave_area:88 msg = "Leave_area"89 else:90 msg = "Enter_area"91 clients = self.sendNotification(client.address[0],msg)92 self.notifier.sendTelegramMsg(self.tokens["chat_id"],f"{self.robots[client.address[0]]} : {self.translate['Messages'][msg]['fr']}",clients,True, sn=self.robots[client.address[0]])9394 def run(self):95 print("Server start...")96 self.notifier.sendTelegramMsg(self.tokens["chat_id"],"Serveur de notification lancé !",list(),False)97 self.running = True98 self.socket.listen(5)99 while self.running:100 try:101 client, address = self.socket.accept()102 except timeout:103 continue104 if DEBUG:105 print(f"[{address[0]}] connected")106 107 if address[0] not in self.robots:108 self.notifier.sendTelegramMsg(self.tokens["chat_id"],f"Robot inconnu ! [ip:{address[0]}] ",list(),False)109 else:110 msg = f"{self.robots[address[0]]} : " + self.translate["Messages"]["Robot_ON"]["fr"]111 client_handling = ClientHandling(client, address, self.client_handling_stopped,self.robots[address[0]],msg,self.notifier,self.tokens,self.send_anti_theft)112 client_handling.start()113 self.client_pool.append(client_handling)114115116class ClientHandling(Thread):117118 def __init__(self, client, address, exit_callback, sn, msg, notifier: Notifier, tokens, send_anti_theft):119 Thread.__init__(self)120 self.client = client121 self.address = address122 self.sn = sn123 self.exit_callback = exit_callback124 self.alive = True125 self.path_gps_with_extract = None126 self.resume_session = None127 self.start_time = None128 self.end_time = None129 self.current_ext = dict()130 self.client.settimeout(10)131 self.msg = msg132 self.notifier = notifier133 self.tokens = tokens134 self.urlMap = ""135 self.send_anti_theft = send_anti_theft136 self.anti_theft = False137138 def _stop(self, error_level: ErrorLevels, error_msg: ErrorMessages):139 self.alive = False140 self.close_connection()141 if self.path_gps_with_extract is not None:142 self.path_gps_with_extract.close()143 if self.resume_session is not None:144 self.resume_session.close()145 extraction = None146 workTime = None147 if self.start_time is not None:148 dateD = datetime.datetime.strptime(self.start_time, "%d-%m-%Y %H-%M-%S %f")149 dateF = datetime.datetime.strptime(self.end_time, "%d-%m-%Y %H-%M-%S %f")150 workTime = str(dateF - dateD)151 if self.current_ext:152 extraction = str(self.current_ext)153 self.exit_callback(self, error_level, error_msg, workTime, extraction, self.urlMap)154155 def close_connection(self):156 self.alive = False157 self.client.close()158 if DEBUG:159 print(f"[{self.address[0]}] disconnected")160161 def onMessage(self, message):162 if SyntheseRobot.HS in message:163 self._stop(ErrorLevels.OK, SyntheseRobot.HS)164 elif ";" in message:165166 infos = message.split(";")167 if infos[1] == SyntheseRobot.OP:168169 if self.anti_theft:170 self.anti_theft = False171 self.send_anti_theft(self, self.anti_theft)172173 if self.path_gps_with_extract is None:174 self.path_gps_with_extract = utility.Logger(f"{self.sn}/{infos[0]}/path_gps_with_extract.txt", add_time=False)175 176 if len(infos) == 4:177 self.path_gps_with_extract.write_and_flush(f"{infos[2]} : {infos[3]}\n")178 for key, value in eval(infos[3]).items():179 if key not in self.current_ext:180 self.current_ext[key] = value181 else:182 self.current_ext[key] += value183 else:184 self.path_gps_with_extract.write_and_flush(f"{infos[2]}\n")185186 if self.resume_session is not None:187 if len(infos) == 4:188 self.resume_session.remove_end_line()189 self.resume_session.remove_end_line()190 self.resume_session.write_and_flush(f"Extraction number : {self.current_ext}\n")191 else:192 self.resume_session.remove_end_line()193 self.end_time = utility.get_current_time()194 self.resume_session.write_and_flush(f"End time : {self.end_time}")195196 elif infos[0] == "START":197 utility.create_directories(self.sn)198 utility.create_directories(f"{self.sn}/{infos[1]}")199 sessionNumber = str(infos[1]).replace(" ","%20")200 self.urlMap = f"\nhttp://172.16.0.9/map/{self.sn}/{sessionNumber}"201 self.notifier.sendTelegramMsg(self.tokens["chat_id"],self.msg,list(),False, sn=self.sn, url_map=self.urlMap) 202 with utility.Logger(f"{self.sn}/{infos[1]}/field.txt", add_time=False) as field_file:203 for coord in eval(infos[4]):204 field_file.write_and_flush(f"{coord}\n") 205 if len(infos) >= 6:206 with utility.Logger(f"{self.sn}/{infos[1]}/field_name.txt", add_time=False) as field_name:207 field_name.write_and_flush(f"{infos[5]}\n") 208 self.resume_session = utility.Logger(f"{self.sn}/{infos[1]}/session_resume.txt", add_time=False)209 self.resume_session.write_and_flush(f"Start time : {infos[1]}\n")210 self.start_time = f"{infos[1]}"211 self.resume_session.write_and_flush(f"Voltage at start : {infos[2]}\n")212 self.resume_session.write_and_flush(f"Treated plant : {infos[3]}\n")213 self.resume_session.write_and_flush("Extraction number : {}\n")214 self.end_time = utility.get_current_time()215 self.resume_session.write_and_flush(f"End time : {self.end_time}")216 elif infos[0] == "STOP":217 if self.resume_session is not None:218 if len(infos) == 3:219 self.resume_session.remove_end_line()220 self.resume_session.remove_end_line()221 self.resume_session.write_and_flush(f"Extraction number : {infos[2]}\n")222 else:223 self.resume_session.remove_end_line()224 self.end_time = utility.get_current_time()225 self.resume_session.write_and_flush(f"End time : {self.end_time}")226227 elif infos[0] == "CLOSE":228 self.close_connection()229 self._stop(ErrorLevels.OK, ErrorMessages.CLOSED)230231 elif infos[1] == SyntheseRobot.ANTI_THEFT:232 if not self.anti_theft:233 self.anti_theft = True234 self.send_anti_theft(self, self.anti_theft)235 236 def run(self):237 try:238 response = self.client.recv(1024)239 while response and self.alive:240 self.onMessage(response.decode("utf-8"))241 if self.alive:242 response = self.client.recv(1024)243 except (ConnectionAbortedError,timeout) as event:244 if str(event) == "timed out":245 self._stop(ErrorLevels.OK, ErrorMessages.LOST)246 if self.alive: 247 self._stop(ErrorLevels.ERROR, ErrorMessages.ABORTED)248 else: 249 return 250 if self.alive:251 self._stop(ErrorLevels.OK, ErrorMessages.CLOSED)252 253254if __name__ == "__main__":255 try:256 server = Server()257 server.start()258 server.join()259 except KeyboardInterrupt:260 print("Ctrl+c are catch...")261 finally:
...
sessions.py
Source:sessions.py
1import os, pickle, re, traceback, readline, shlex2from collections import OrderedDict3import iglesia4from radiopadre_client import config5from iglesia.utils import message, warning, error, bye, ff, INPUT, make_radiopadre_dir6from iglesia import logger7_recent_sessions = None8_last_input = None9RECENTS_FILE = os.path.join(iglesia.RADIOPADRE_DIR, "radiopadre-client.sessions.recent")10HISTORY_FILE = os.path.join(iglesia.RADIOPADRE_DIR, "radiopadre-client.sessions.history")11def _load_recent_sessions(must_exist=True):12 """13 Load recent sessions from RECENTS_FILE.14 :param must_exist: if True and session file does not exist, exit with error15 :return: dict of latest sessions16 """17 global _recent_sessions18 if _recent_sessions is not None:19 return _recent_sessions20 if os.path.exists(RECENTS_FILE):21 _recent_sessions = OrderedDict()22 try:23 for line in open(RECENTS_FILE, "rt"):24 key, args = line.strip().split(":::", 1)25 _recent_sessions[key] = args26 except Exception as exc:27 message(ff("Error reading {RECENTS_FILE}: {exc}"))28 _recent_sessions = None29 if _recent_sessions is None and must_exist:30 bye("no recent radiopadre sessions and no arguments given. Run with -h for help.")31 return _recent_sessions32def check_recent_sessions(options, argv, parser=None):33 """34 Loads a recent session if requested35 :param options: Options object from ArgumentParser36 :param argv: Argument list (from sys.argv[1:] initially)37 :param parser: ArgumentParser object used to (re)parse the options38 :return: options, argv39 Where options and argv may have been loaded from the recent40 options file41 """42 make_radiopadre_dir()43 # load history44 try:45 readline.read_history_file(HISTORY_FILE)46 except IOError:47 pass48 resume_session = None49 # a single-digit argument resumes session #N50 if len(options.arguments) == 1 and re.match("^\d$", options.arguments[0]):51 resume_session = int(options.arguments[0])52 # no arguments is resume session #053 elif not options.arguments:54 resume_session = 055 if resume_session is not None:56 last = _load_recent_sessions()57 num_recent = len(last)58 if resume_session >= num_recent:59 bye(ff("no recent session #{resume_session}"))60 message("Your most recent radiopadre sessions are:")61 message("")62 for i, (_, opts) in enumerate(list(last.items())[::-1]):63 message(" [#{0}] {1}".format(i, opts), color="GREEN")64 message("")65 print("\nInteractive startup mode. Edit arguments and press Enter to run, or Ctrl+C to bail out. ")66 print(" (Ctrl+U + <NUM> + Enter will paste other recent session arguments from the list above)\n")67 inp = None68 cmdline = ''69 readline.set_startup_hook(lambda: readline.insert_text(cmdline))70 while inp is None:71 # form up list of fake args to be re-parsed for the last session72 cmdline = list(last.items())[-(resume_session + 1)][1]73 # non-persisting options raised in command line shall be appended to the fake args74 for opt in config.NON_PERSISTING_OPTIONS:75 if opt.startswith("--") and getattr(options, opt[2:].replace("-", "_"), None):76 cmdline += " " + opt77 cmdline += " "78 ## colors confuse Ctrl+U and such79 # prompt = ff("{logger.Colors.GREEN}[#{resume_session}]:{logger.Colors.ENDC} ")80 prompt = ff("[#{resume_session}] ")81 inp = INPUT(prompt)82 inp = inp.strip()83 if not inp:84 resume_session = 085 inp = None86 elif re.match("^\d+$", inp):87 res = int(inp)88 if res >= num_recent:89 warning(ff("no recent session #{res}"))90 else:91 resume_session = res92 readline.remove_history_item(1)93 inp = None94 readline.set_startup_hook(None)95 global _last_input96 _last_input = inp97 argv = shlex.split(inp, posix=False)98 options = parser.parse_args(argv)99 return options, argv100def save_recent_session(session_key, argv):101 """102 Saves session arguments into recents file.103 :param session_key: key to save under (only one session per key is saved)104 :param argv: argument list to save105 :return: None106 """107 # add current line to history, if not already there108 cmdline = " ".join([x if x and not ' ' in x else "'{}'".format(x) for x in argv])109 if not _last_input:110 if cmdline != readline.get_history_item(readline.get_current_history_length()):111 readline.add_history(cmdline)112 make_radiopadre_dir()113 try:114 readline.write_history_file(HISTORY_FILE)115 except IOError:116 traceback.print_exc()117 warning("Error writing history file (see above). Proceeding anyway.")118 readline.clear_history()119 # reform command-line without persisting options120 cmdline = " ".join([x if x and not ' ' in x else "'{}'".format(x) for x in argv121 if x not in config.NON_PERSISTING_OPTIONS])122 recents = _load_recent_sessions(False) or OrderedDict()123 session_key = ":".join(map(str, session_key))124 if session_key in recents:125 del recents[session_key]126 if len(recents) >= 5:127 del recents[list(recents.keys())[0]]128 recents[session_key] = cmdline129 make_radiopadre_dir()130 with open(RECENTS_FILE, 'wt') as rf:131 for key, cmdline in recents.items():132 rf.write("{}:::{}\n".format(key, cmdline))133 global _recent_sessions...
on_completion.py
Source:on_completion.py
1# coding: utf-82#3# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.4#5# Licensed under the Apache License, Version 2.0 (the "License"). You may not use this file6# except in compliance with the License. A copy of the License is located at7#8# http://aws.amazon.com/apache2.0/9#10# or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for12# the specific language governing permissions and limitations under the License.13#14import pprint15import re # noqa: F40116import six17import typing18from enum import Enum19if typing.TYPE_CHECKING:20 from typing import Dict, List, Optional, Union, Any21 from datetime import datetime22class OnCompletion(Enum):23 """24 This defines the callback mechanism when the task is completed, i.e., whether the requester wants to be resumed after the task is fulfilled or just be notified about errors without being resumed.25 Allowed enum values: [RESUME_SESSION, SEND_ERRORS_ONLY]26 """27 RESUME_SESSION = "RESUME_SESSION"28 SEND_ERRORS_ONLY = "SEND_ERRORS_ONLY"29 def to_dict(self):30 # type: () -> Dict[str, Any]31 """Returns the model properties as a dict"""32 result = {self.name: self.value}33 return result34 def to_str(self):35 # type: () -> str36 """Returns the string representation of the model"""37 return pprint.pformat(self.value)38 def __repr__(self):39 # type: () -> str40 """For `print` and `pprint`"""41 return self.to_str()42 def __eq__(self, other):43 # type: (Any) -> bool44 """Returns true if both objects are equal"""45 if not isinstance(other, OnCompletion):46 return False47 return self.__dict__ == other.__dict__48 def __ne__(self, other):49 # type: (Any) -> bool50 """Returns true if both objects are not equal"""...
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?
