merge

docs/checklogfile.py

@@ -0,0 +1,112 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-  
+import time, sys, getopt, os, random
+
+ip = ""
+login = "admin"
+password = "12345"
+
+def usage():
+    print "Usage:\r\n\
+        checklogfile.py <filename> - to check local log file\r\n\
+        checklogfile.py -d <ip-address> - to download and check log file"
+
+
+def parse(path):
+    f = open(path)
+    lines = [line.rstrip('\n') for line in f]
+ 
+    #Remove UTF8 BOM 
+    lines[0] = lines[0].replace('\xEF\xBB\xBF','')
+
+    #Remove empty lines
+    for i in range(len(lines)-1, len(lines)-10, -1):
+        #print "str -", i, lines[-i] 
+        if lines[i] == " "*118+"\x0D":
+            del lines[i]
+
+    print "len: ", len(lines)
+    return lines
+
+
+def do_login():
+        os.system("curl -s -b /tmp/cookies.txt -c /tmp/cookies.txt -d \"login=%s&password=%s\" http://%s/login.cgi > /dev/null" % (login, password, ip))
+
+
+def log_download(path):
+        print "Downloading log file..."
+	if 0 == os.system("curl -b /tmp/cookies.txt -c /tmp/cookies.txt \"http://%s/history.cgi?page=all&_=%s\" --output %s --progress" % (ip, random.randint(0, 0xFFFF), path)):
+            return True
+        else:
+            return False
+
+
+def check(path):
+    lines = parse(path)
+ 
+    lastptr = 0
+    for i in range(len(lines)):
+        #i = lines.index(s)
+        s = lines[i]
+        print "\rProcessing: %d%%" %( (i+1)*100/len(lines)),
+
+        #Check if string in valid format
+        if len(s) > 119 or s[0] != '"' or s.count('"') != 2 or s.count('\x00') != 0:
+            print "\nError. Log file is corrupted:"
+            print "Wrong string (at index %d) : %s" %(i+1, s)
+            return False
+
+        strdate = s.replace('"','').split(";")[3].rstrip()
+        #print strdate
+        newdate = time.strptime(strdate, "%d.%m.%y %H:%M:%S")
+        if i != 0:
+            if newdate < lastdate:
+                if lastptr == 0:
+                    lastptr = i+1
+                else:
+                    print "\nError. Wrong message order!"
+                    print "Look at index:", i+1 
+ 
+        lastdate = newdate        
+   
+    print "\nLast message index:", lastptr
+    print "Log file is OK"
+    return True
+
+        
+def main():
+    global ip
+
+    try:
+        opts, args = getopt.getopt(sys.argv[1:], "hd:", \
+                ["help", "download="])
+    except getopt.GetoptError:
+        usage()
+        sys.exit(2)
+    
+    for opt, arg in opts:
+        if opt in ("-h", "--help"):
+                usage()
+                sys.exit()
+        if opt in ("-d", "--download"):
+                ip = arg
+
+    if ip == "" and len(sys.argv) < 2:
+        usage()
+        sys.exit(2)
+
+    if ip == "": 
+        path = sys.argv[-1]
+    else:
+        path = "/tmp/checklog.csv"
+        do_login()
+        if not log_download(path):
+            print "Download error"
+            sys.exit(2)
+    
+    check(path)
+
+
+if __name__ == "__main__":
+    main()
+

docs/test_log.py

@@ -2,7 +2,7 @@
 # -*- coding: utf-8 -*-  
 import os, random, sys, time, json, getopt, threading
 
-sys.path.append('../epu_tester')
+sys.path.append('../../stm32_prs/stm32_prs/branches/epu_tester')
 import extdevices
 import checklogfile
 
@@ -170,58 +170,7 @@ def on_off_test():
 		print "ON"
 		time.sleep(4)
 
-                #Waiting LOG restore
-                delay = 0
-		res = 1
-		while res != 0:
-			res = os.system("tail ./trace.txt | grep -a \"Write page\" > /dev/null")
-			if res != 0:
-				print "Waiting restore: %d" % delay
-				delay+=5
-				time.sleep(5)
-			else:
-				time.sleep(12)
-            
-                do_login()
-
-                #Get start timestamp
-                if timestamp == '':
-                    strdate = log_get_string(1, 0).replace('"','').split(";")[3].rstrip()
-                    print "Started at:", strdate
-                    timestamp = time.strptime(strdate, "%d.%m.%y %H:%M:%S")
-                
-                #Check if Log index was overfloved since test start. If so, finish the test.
-                lastpage = log_get_pagesnum()
-                if lastpage > 3000:
-                    lastpages = 3000
-                strdate = log_get_string(lastpage, 0).replace('"','').split(";")[3].rstrip()
-                if timestamp < time.strptime(strdate, "%d.%m.%y %H:%M:%S"):
-                    print "Test finished. Checking LOG.."
-                    #Download and check log file
-                    PATH = "/tmp/testlog.csv"
-                    if log_download(PATH):
-                        res = checklogfile.check(PATH)
-
-                    break                   
-
-                #Validate LOG pages
-                if msg:
-                    print "Finding message:", msg
-                    for i in range(1, int(log_get_pagesnum())):
-                            parsed_page = json.loads(log_get_page(i))
-                            if len(parsed_page) != 10:
-                                    print "Error. Corrupted page:"
-                                    stop_spam_task()
-                                    for i in parsed_page:
-                                        print i 
-                                    return False
-                            if msg in parsed_page:
-                                    print "Found:", i 
-                                    break
-                            else:
-                                    print "Page: %d" % i
-            
-                msg = log_get_string(2, 0)
+           
                 time.sleep(random.randint(1, 4))		
 	    except KeyboardInterrupt:
 		sys.exit(0)