How to use _clean_sessions method in autotest

Best Python code snippet using autotest_python

base_syncdata.py

Source:base_syncdata.py

...94        if self.tmpdir.name:95            logging.error("SyncListenServer on port %d was not closed." %96                          self.port)97            self.close()98    def _clean_sessions(self):99        """100        Close and delete timed-out connection.101        """102        to_del = []103        for session_id, session in self.sessions.items():104            if session.data_lock.acquire(False):105                if ((not session.is_finished() and not session.timeout()) or106                        session.is_finished()):107                    if not session.is_finished():108                        logging.warn("Sync session %s timed out and will"109                                     " be closed and deleted." %110                                     (session.hosts))111                    session.close()112                    to_del.append(session_id)113                session.data_lock.release()114        for td in to_del:115            del(self.sessions[td])116    def _recv_data(self, connection, session):117        session.data_lock.acquire()118        client, addr = connection119        session.connection[addr[0]] = connection120        try:121            logging.debug("Try recv from client")122            session.sync_data[addr[0]] = net_recv_object(client,123                                                         _DEFAULT_TIMEOUT)124            session.data_recv += 1125        except socket.timeout:126            logging.warn("Fail to communicate with client"127                         " %s. Synchronization of data "128                         "is not possible." % (addr))129        except error.NetCommunicationError:130            pass131        if not session.is_finished():132            if (session.data_recv == len(session.hosts) and133                    session.timeout()):134                for client, _ in session.connection.values():135                    try:136                        net_send_object(client, session.sync_data)137                        net_recv_object(client, _DEFAULT_TIMEOUT)138                    except (socket.timeout, error.NetCommunicationError):139                        self._clean_sessions()140                session.set_finish()141        session.data_lock.release()142    def __call__(self, signum, frame):143        self.exit_event.set()144    def _start_server(self, address, port):145        signal.signal(signal.SIGTERM, self)146        self.server_thread = utils.InterruptedThread(self._server,147                                                     (address, port))148        self.server_thread.start()149        while not self.exit_event.isSet():150            signal.pause()151        self.server_thread.join(2 * _DEFAULT_TIMEOUT)152        logging.debug("Server thread finished.")153        for session in self.sessions.itervalues():154            session.close()155        self.listen_server.close()156        logging.debug("ListenServer closed finished.")157    def _server(self, address, port):158        self.listen_server = barrier.listen_server(address, port)159        logging.debug("Wait for clients")160        self.listen_server.socket.settimeout(_DEFAULT_TIMEOUT)161        while not self.exit_event.isSet():162            try:163                connection = self.listen_server.socket.accept()164                logging.debug("Client %s connected.", connection[1][0])165                session_id, hosts, timeout = net_recv_object(connection[0],166                                                             _DEFAULT_TIMEOUT)167                self._clean_sessions()168                if session_id not in self.sessions:169                    logging.debug("Add new session")170                    self.sessions[session_id] = SessionData(hosts, timeout)171                logging.debug("Start recv thread.")172                utils.InterruptedThread(self._recv_data, (connection,173                                                          self.sessions[session_id])).start()174            except (socket.timeout, error.NetCommunicationError):175                self._clean_sessions()176        logging.debug("SyncListenServer on closed.")177    def close(self):178        """179        Close SyncListenServer thread.180        Close all open connection with clients and listen server.181        """182        utils.signal_pid(self.server_pid, signal.SIGTERM)183        if utils.pid_is_alive(self.server_pid):184            parallel.fork_waitfor_timed(self.tmpdir.name, self.server_pid,185                                        2 * _DEFAULT_TIMEOUT)186        self.tmpdir.clean()187        logging.debug("SyncListenServer was killed.")188class SyncData(object):189    """...

flask.py

Source:flask.py

...108		self._oldest = None109		self._mutex = threading.RLock()110		self._logger = logging.getLogger(__name__)111	def create(self):112		self._clean_sessions()113		key = ''.join('%02X' % ord(z) for z in uuid.uuid4().bytes)114		created = time.time()115		valid_until = created + self.__class__.VALIDITY_UNVERIFIED116		with self._mutex:117			self._sessions[key] = (created, False, valid_until)118		return key, valid_until119	def remove(self, key):120		with self._mutex:121			if not key in self._sessions:122				return123			del self._sessions[key]124	def verify(self, key):125		self._clean_sessions()126		if not key in self._sessions:127			return False128		with self._mutex:129			created, verified, _ = self._sessions[key]130			if verified:131				return False132			valid_until = created + self.__class__.VALIDITY_VERIFIED133			self._sessions[key] = created, True, created + self.__class__.VALIDITY_VERIFIED134		return key, valid_until135	def validate(self, key):136		self._clean_sessions()137		return key in self._sessions and self._sessions[key][1]138	def _clean_sessions(self):139		if self._oldest is not None and self._oldest > time.time():140			return141		with self._mutex:142			self._oldest = None143			for key, value in self._sessions.items():144				created, verified, valid_until = value145				if not verified:146					valid_until = created + self.__class__.VALIDITY_UNVERIFIED147				if valid_until < time.time():148					del self._sessions[key]149				elif self._oldest is None or valid_until < self._oldest:150					self._oldest = valid_until...

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.