KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use getlastlog method in pyatom
Best Python code snippet using pyatom_python
JSMaker.py
Source:JSMaker.py 
1"""2MetPX Copyright (C) 2004-2006 Environment Canada3MetPX comes with ABSOLUTELY NO WARRANTY; For details type see the file4named COPYING in the root of the source directory tree.5"""6"""7#############################################################################################8# Name: JSMaker.py9#10# Author: Daniel Lemay11#12# Date: 2004-10-1213#14#############################################################################################15"""16import os, sys, commands, re, logging, time17sys.path.append(sys.path[0] + "/../../lib");18sys.path.append("../../lib")19from PDSPath import *20from ColumboPaths import *21from CompositePDSClient import CompositePDSClient22from CompositePDSInputDir import CompositePDSInputDir23import readMaxFile24class JSMaker:25 """26 #############################################################################################27 # This object is used to create javascript28 #############################################################################################29 """30 def __init__(self):31 self.firstTag = '<script type="text/javascript">'32 self.lastTag = '</script>' 33 34 def createMaxers(self, clientDict, inputDirDict):35 theClients = clientDict.keys()36 theDirs = inputDirDict.keys()37 clientsRegex, defaultClient, inputDirsRegex, defaultInputDir, graphsRegex, defaultGraph = readMaxFile.readQueueMax(ETC + "/maxSettings.conf", 'PDS')38 self.clientsMax = readMaxFile.setValueMax(theClients, clientsRegex, defaultClient)39 self.dirsMax = readMaxFile.setValueMax(theDirs, inputDirsRegex, defaultInputDir)40 self.graphsMax = readMaxFile.setValueMax(theClients, graphsRegex, defaultGraph)41 42 def setNCSMax(self, circuitDict):43 theCircuits = circuitDict.keys()44 circuitRegex, defaultCircuit, timerRegex, defaultTimer, graphRegex, defaultGraph = readMaxFile.readQueueMax(ETC + "/maxSettings.conf", 'PX')45 self.circuitMax = readMaxFile.setValueMax(theCircuits, circuitRegex, defaultCircuit)46 self.timerMax = readMaxFile.setValueMax(theCircuits, timerRegex, defaultTimer)47 self.graphMax = readMaxFile.setValueMax(theCircuits, graphRegex, defaultGraph)48 def createJSArrayClients(self, clientDict):49 keys = clientDict.keys()50 print "var clients = new Array();"51 i = 0 52 for key in keys:53 stopped = self.isStopped(clientDict, key) # 1 if stopped54 inError = self.isInError(clientDict, key) # 1 if in error55 print """clients[%d] = ["%s", %d, "%s", %d, %d, %d, %d];""" % (i, key, clientDict[key].getCompositeQueue(), clientDict[key].getBestLog().rstrip("\n").replace('"', "'"), stopped, inError, int(self.clientsMax[key]), int(self.graphsMax[key]))56 i += 157 def createJSArrayNCSCircuits(self, circuitDict):58 """59 # Creates the statics JavaScript array containing the information.60 # For a clearer view of the array please refer to the program's documentation.61 """62 keys = circuitDict.keys()63 print "var circuits = new Array();"64 i = 065 for key in keys:66 circuit = circuitDict[key]67 stopped = circuit.getGlobalStatus() # 1 if stopped partially, 2 if completely stopped, 0 if OK68 inError = self.isInError(circuitDict, key) # 1 if in error69 if circuit.getGlobalType().find('pxReceiver') != -1:70 rcv = self.tooLong(circuit.getGlobalLastRcv(), int(self.timerMax[key]))71 trans = 072 elif circuit.getGlobalType().find('pxSender') != -1:73 trans = self.tooLong(circuit.getGlobalLastTrans(), int(self.timerMax[key]))74 rcv = 075 print """circuits[%d] = ["%s", "%s", "%s", "%s", %d, "%s", "%s", %d, %d, %d, %d, %d, %d, %d];""" % (i, key, circuit.getGlobalType(), circuit.getGlobalLastRcv(), circuit.getGlobalLastTrans(), circuit.getCompositeQueue(), circuit.getSocketState(), circuit.getBestLog().rstrip("\n").replace('"', "'"), stopped, inError, int(self.circuitMax[key]), circuit.getSocketFlag(), rcv, trans, int(self.graphMax[key]))76 i += 177 78 def createJSArrayNCSInfos(self, circuitDict, name):79 myCircuit = circuitDict[name]80 hosts = myCircuit.getHosts()81 print 'var myCircuit = "%s" % (name)'82 print "var circuitInfos = new Array();"83 i = 084 for host in hosts:85 print """circuitInfos[%d] = ["%s", %d, "%s", "%s", "%s", "%s", %d];""" % (i, host, myCircuit.getQueue(host), myCircuit.getSocket(host), myCircuit.getLastLog(host)[0].rstrip("\n").replace('"', "'"), myCircuit.getStatus(host), myCircuit.getType(host), int(self.circuitMax[name]))86 i += 187 88 def createJSArrayInputDirs(self, inputDirDict):89 keys = inputDirDict.keys()90 print "var inputDirs = new Array();"91 i = 0 92 for key in keys:93 print """inputDirs[%d] = ["%s", %d, %d ];""" % (i, key, inputDirDict[key].getCompositeQueue(), int(self.dirsMax[key]))94 i += 195 96 def createJSArrayRepartition(self, clientDict, name):97 myClient = clientDict[name]98 hosts = myClient.getHosts()99 print 'var myclient = "%s"' % (name)100 print "var clientInfos = new Array();"101 i = 0102 for host in hosts:103 print """clientInfos[%d] = ["%s", %d, "%s", "%s", "%s"];""" % (i, host, myClient.getQueue(host), myClient.getLastLog(host)[0].rstrip("\n").replace('"', "'"), myClient.getDate(host), myClient.getStatus(host))104 i += 1105 def createJSArrayRepartitionInputDir(self, inputDirDict, inputDir):106 myInputDir = inputDirDict[inputDir]107 hosts = myInputDir.getHosts()108 print "var inputDirInfos = new Array();"109 i = 0110 for host in hosts:111 print """inputDirInfos[%d] = ["%s", %d];""" % (i, host, myInputDir.getQueue(host))112 i += 1113 def createJSArrayListing(self, listingDict):114 keys = listingDict.keys()115 print "var listing = new Array();"116 i = 0 117 for key in keys:118 print """listing[%d] = ["%s", "%s"];""" % (i, listingDict[key][1], key)119 i += 1120 def staticHtmlForTable(self, tableId, bodyId, headers ):121 print "<table id='%s' " % (tableId),122 print "width='98%'>"123 print """124 <thead>125 <tr>126 """ 127 for header in headers:128 print """129 <th class='header'>%s</th> 130 """ % (header)131 print """132 </tr>133 </thead>134 <tbody id="%s"></tbody>135 </table>136 """ %(bodyId)137 def makeHtmlListingTable(self, tableId, bodyId):138 print "<table id='%s' " % (tableId),139 print "cellpadding='0' cellspacing='0' width='100%' bgcolor='#00ff00'>"140 print """141 <tbody id="%s"></tbody>142 </table>143 """ %(bodyId)144 def makeJS (self):145 pass146 147 def isStopped (self, clientDict, name):148 myClient = clientDict[name]149 machines = clientDict[name].getHosts()150 for machine in machines:151 if (myClient.getStatus(machine) == "STOPPED"):152 return 1153 return 0154 def isInError (self, clientDict, name):155 myClient = clientDict[name]156 regex1 = re.compile(r'ERROR|has been queued too long|Timeout|\[ERROR\]')157 regex2 = re.compile(r"Interrupted system call|425 Can't build data connection")158 """159 machines = clientDict[name].getHosts()160 for machine in machines:161 match = regex.search(myClient.getLastLog(machine)[0])162 if (match):163 return 1164 return 0165 """166 if regex1.search(myClient.getBestLog()):167 if regex2.search(myClient.getBestLog()):168 return 0169 else:170 return 1171 return 0172 def tooLong(self, lastCom, limit):173 if lastCom == "NOT FOUND":174 return 1175 nbSec = limit * 60176 now = time.localtime()177 last = time.strptime(lastCom, "%Y-%m-%d %H:%M:%S")178 if time.mktime(now) - nbSec > time.mktime(last):179 return 1180 else:...
alerts.py
Source:alerts.py
1# functions.py2import globals3import http.client4#import urllib5import functions6import requests7#import pushover8#from pushover import init, Client9from pushover import Pushover10# From Github (https://github.com/dimaba/sendmail)11from sendmail import MailSender12PUSHOVER_TOKEN = { 'token' : globals.pushOverAPPToken}13PUSHOVER_USER = { 'user' : globals.pushOverUserKey}14#Pushover.__init__globals.pushOverAPPToken)15#Client(globals.pushOverUserKey).send_message("PiSS has Started", title="PiSS is Running")16def sendIFTTT():17 print(globals.IFTTTAction[0])18 receive = requests.get(globals.IFTTTAction[0])19def sendNotification(Messagetype, zonenumber, title, subject, mailFrom, messageDetail, mailTo, customTitle, customMessage):20 GETLASTLog=functions.getLastSensorLog(globals.ZoneinAlarm)21 zoneinfo = functions.PISSZoneStatus(globals.ZoneinAlarm)22 if (globals.ENABLE_PUSH==1):23 po = Pushover(globals.pushOverAPPToken[0])24 po.user(globals.pushOverUserKey[0]) 25 msg=po.msg("PiSS")26 27 msg.set("html",1)28 29 if Messagetype == 0: 30 msg.set("title", globals.pushsmtpTitle[0] % globals.SYSTEM_NAME)31 msg.set("message", globals.pushsmtpMessages[0])32 if Messagetype == 1:33 msg.set("title", globals.pushsmtpTitle[1] % (globals.SYSTEM_NAME,zoneinfo[2]))34 msg.set("message", globals.pushsmtpMessages[1] % zoneinfo[2])35 msg.set("priority","1")36 msg.set("retry","60")37 msg.set("sound","siren")38 msg.set("expire","3600")39 if Messagetype == 2:40 msg.set("title", globals.pushsmtpTitle[2] % globals.SYSTEM_NAME)41 msg.set("message", globals.pushsmtpMessages[2] % globals.SYSTEM_NAME)42 if Messagetype == 3:43 msg.set("title", globals.pushsmtpTitle[3] % (globals.SYSTEM_NAME))44 msg.set("message", globals.pushsmtpMessages[3])45 msg.set("priority","1")46 msg.set("retry","60")47 msg.set("sound","siren")48 msg.set("expire","3600")49 if Messagetype == 4:50 msg.set("title", globals.pushsmtpTitle[4] % (globals.SYSTEM_NAME))51 msg.set("message", globals.pushsmtpMessages[4])52 msg.set("priority","1")53 msg.set("retry","60")54 msg.set("sound","siren")55 msg.set("expire","3600")56 if Messagetype == 5:57 msg.set("title", globals.pushsmtpTitle[5])58 msg.set("message", globals.pushsmtpMessages[5])59 if Messagetype == 6:60 msg.set("title", globals.pushsmtpTitle[6])61 msg.set("message", globals.pushsmtpMessages[6])62 if Messagetype == 7:63 msg.set("title", globals.pushsmtpTitle[7])64 msg.set("message", globals.pushsmtpMessages[7])65 if Messagetype == 8: # Custom message... 66 msg.set("title", customTitle)67 msg.set("message", customMessage)68 if Messagetype == 9: # Custom message... 69 msg.set("title", customTitle)70 msg.set("message", customMessage)71 msg.set("priority","1")72 msg.set("retry","60")73 msg.set("sound","siren")74 msg.set("expire","3600")75 if Messagetype == 10:76 msg.set("title", globals.pushsmtpTitle[8] % (globals.SYSTEM_NAME,zoneinfo[2]))77 msg.set("message", globals.pushsmtpMessages[8] % zoneinfo[2])78 msg.set("priority","1")79 msg.set("retry","60")80 msg.set("sound","siren")81 msg.set("expire","3600")82 msg.set("url",globals.pushURL)83 po.send(msg)84 85 if (globals.ENABLE_EMAIL==1):86 ourmailsender = MailSender(globals.smtpUser, globals.smtpPassword, (globals.smtpServer, globals.smtpPort))87 88 if Messagetype == 0: 89 #Do NOthing90 return91 92 if Messagetype == 1:93 subject = globals.pushsmtpTitle[1] % (globals.SYSTEM_NAME,zoneinfo[2])94 messageDetail = globals.pushsmtpMessages[1] % zoneinfo[2]95 plaintext = "Alarm Notification - Zone: " + zoneinfo[2] + " , \n" \96 "Time of Event - \n" + str(GETLASTLog) + " \n" \97 "" + messageDetail + "\n" + \98 " Status=" + str(GETLASTLog[1]) + \99 "\n" + subject + "" 100 ourmailsender.set_message(plaintext, subject, globals.SYSTEM_NAME)101 ourmailsender.set_recipients(globals.mailRecip) # Don't need to specify address in array as it's already in that format (a turple)102 ourmailsender.connect()...
pday.py
Source:pday.py
1#!/usr/bin/python32import os3import datetime4#import optparse5from sys import argv6log_file = 'start_end.log'7lock_file = '/var/lock/pycuslockfile_for_onestart'8io = 09def getLog() -> io : # IO<10 if not os.path.exists(log_file):11 with open(log_file, 'wt') as fout:12 fout.write('')13 with open(log_file, 'rt') as fin:14 return fin.read()15def getLastLog(n):16 return getLog().split('\n')[-n]17def getLastLogDate(n):18 x = getLastLog(n)19 return x.split(':', 1)[1].split('T')[0].strip()20def getLastLogTime(n):21 x = getLastLog(n)22 return x.split(':', 1)[1].split('T')[1].strip().split('.')[0]23def getLastDateTime(n):24 x = getLastLog(n)25 x = x.split(':', 1)[1].strip()26 d = datetime.datetime.strptime(x, "%Y-%m-%dT%H:%M:%S.%f")27 return d28def log(string) -> io : # IO>29 with open(log_file, 'at') as fout:30 fout.write(string)31 fout.write('\n')32def getDate() -> io : # IO<33 return datetime.datetime.now().isoformat()34def logStart() -> io : # IO<>35 log('start at: ' + getDate())36def logEnd() -> io : # IO<>37 log('end at: ' + getDate())38def dayStart(a):39 if lock():40 logStart()41 sleepLog()42def dayEnd(a):43 if freeLock():44 logEnd()45def isLock():46 return os.path.exists(lock_file)47def lock():48 if not isLock():49 with open(lock_file, 'w') as f:50 f.write('')51 return True52 else:53 return False54def freeLock():55 if isLock():56 os.remove(lock_file)57 return True58 else:59 return False60def viewLog(a):61 print(getLog())62def pday():63 if len(argv) < 2:64 print('Need more arguments like: start, end...')65 exit(0)66 cmd = argv[1]67 args = ''.join(argv[2:])68 commands = {69 'start': dayStart,70 'end': dayEnd,71 'log': viewLog,72 }73 cmd_fun = commands.get(cmd, lambda x: print('no such command'))74 cmd_fun(args)75 76def getSleepTime():77 cur_date = datetime.date.today().isoformat()78 now = datetime.datetime.now()79 last_log_date = getLastLogDate(2)80 if cur_date != last_log_date:81 start = getLastDateTime(3)82 end = getLastDateTime(2)83 no_sleeps = end - start84 sleeps = now - end85 sleep_h = sleeps.seconds // (60*60)86 sleep_m = (sleeps.seconds % (sleep_h * (60*60))) // 6087 end_time = '{:02}:{:02}'.format(end.hour, end.minute)88 start_time = '{:02}:{:02}'.format(now.hour, now.minute)89 sleep_time = '{:02}:{:02}'.format(sleep_h, sleep_m)90 return end_time, sleep_time, start_time91def sleepLog():92 end, sleep, start = getSleepTime()93 sleep_log_file = 'sleep.log'94 with open(sleep_log_file, 'at') as fout:95 fout.write(end + ',' + sleep + '\n')96 fout.write(start + ',')97 #save date wake woke sleep98if __name__=='__main__':...
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 pyatom 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?
