refactored load_every_help into two separate support classes: PdTest and TestLog

svn path=/trunk/; revision=15565

2 files changed, 268 insertions, 0 deletions

diff --git a/scripts/tests/load_every_help.py b/scripts/tests/load_every_help.py
new file mode 100755
index 00000000..81508871
--- /dev/null
+++ b/scripts/tests/load_every_help.py
@@ -0,0 +1,49 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+import sys
+import time
+import socket
+import os
+import re
+from pdtest import PdTest, TestLog
+
+try:
+    pdrootdir = sys.argv[1]
+except IndexError:
+    print 'only one arg: root dir of pd'
+    sys.exit(2)
+
+test = PdTest(pdrootdir)
+
+now = time.localtime(time.time())
+date = time.strftime('20%y-%m-%d', now)
+datestamp = time.strftime('20%y-%m-%d_%H.%M.%S', now)
+
+outputfile = '/tmp/load_every_help_' + socket.gethostname() + '_' + datestamp + '.log'
+fd = open(outputfile, 'w')
+fd.write('load_every_help\n')
+fd.write('========================================================================\n')
+fd.flush()
+
+extradir = os.path.join(pdrootdir, 'extra', 'earplug~')
+for root, dirs, files in os.walk(extradir):
+    for name in files:
+        m = re.search(".*-help\.pd$", name)
+        if m:
+            test.runtest(fd, root, name)
+
+# docdir = os.path.join(pdrootdir, 'doc')
+# for root, dirs, files in os.walk(docdir):
+#     for name in files:
+#         m = re.search(".*\.pd$", name)
+#         if m:
+#             test.runtest(fd, root, name)
+
+fd.close()
+
+
+# upload the load and send the email report
+log = TestLog(outputfile)
+log.upload()
+log.email(subject='load_every_help ' + datestamp)
diff --git a/scripts/tests/pdtest.py b/scripts/tests/pdtest.py
new file mode 100755
index 00000000..28b3a612
--- /dev/null
+++ b/scripts/tests/pdtest.py
@@ -0,0 +1,219 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+import os
+import re
+import random
+import signal
+import smtplib
+import socket
+import subprocess
+import sys
+import time
+
+class PdTest():
+
+    def __init__(self, pdrootdir):
+        self.pdrootdir = pdrootdir
+        self.rand = random.SystemRandom()
+
+    def find_pdexe(self):
+        # start with the Windows/Mac OS X location
+        exe = self.pdrootdir + '/bin/pd'
+        if not os.path.isfile(exe):
+            # try the GNU/Linux location
+            exe = self.pdrootdir+'../../bin/pd'
+            if not os.path.isfile(exe):
+                print "ERROR: can't find pd executable"
+                exit
+        return os.path.realpath(exe)
+
+    def make_netreceive_patch(self, port):
+        filename = '/tmp/.____pd_netreceive_'+str(self.port)+'____.pd'
+        fd = open(filename, 'w')
+        fd.write('#N canvas 222 130 454 304 10;')
+        fd.write('#X obj 201 13 import vanilla;')
+        fd.write('#X obj 111 83 netreceive ' + str(self.port) + ' 0 old;')
+        fd.write('#X obj 111 103 loadbang;')
+        fd.write('#X obj 111 123 print netreceive_patch;')
+    # it would be nice to have this patch tell us when it is closed...
+    #    fd.write('#X obj 211 160 tof/destroysend pd;')
+    #    fd.write('#X obj 211 160 closebang;')
+    #    fd.write('#X obj 211 180 print CLOSE;')
+        fd.write('#X connect 2 0 3 0;')
+    #    fd.write('#X connect 3 0 4 0;')
+        fd.close()
+        self.netreceive_patch = filename
+
+    def send_to_socket(self, message):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        s.settimeout(5)
+        try:
+            s.connect(('localhost', self.port))
+            s.send(message)
+            s.close()
+        except socket.timeout:
+            print "socket timed out while sending"
+
+    def send_to_pd(self, message):
+        self.send_to_socket('; pd ' + message + ';\n')
+
+    def open_patch(self, filename):
+        dir, file = os.path.split(filename)
+        self.send_to_pd('open ' + file + ' ' + dir)
+
+    def close_patch(self, filename):
+        dir, file = os.path.split(filename)
+        self.send_to_pd('; pd-' + file + ' menuclose')
+
+    def launch_pd(self):
+        pdexe = self.find_pdexe()
+        p = subprocess.Popen([pdexe, '-nogui', '-stderr', '-open', self.netreceive_patch],
+                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT, 
+                             close_fds=True);
+        line = p.stdout.readline()
+        while line != 'netreceive_patch: bang\n':
+            line = p.stdout.readline()
+        return p
+
+    def quit_pd(self, process):
+        self.send_to_pd('quit')
+        time.sleep(1)
+        try:
+            os.kill(process.pid, signal.SIGTERM)
+        except OSError:
+            print 'OSError on SIGTERM'
+        time.sleep(1)
+        try:
+            os.kill(process.pid, signal.SIGKILL)
+        except OSError:
+            print "OSError on SIGKILL"
+
+
+    #---------- list of lines to ignore ----------#
+    def remove_ignorelines(self, list):
+        ignorelines = [
+            'expr, expr~, fexpr~ version 0.4 under GNU General Public License \n',
+            'fiddle version 1.1 TEST4\n',
+            'sigmund version 0.07\n',
+            'bonk version 1.5\n'
+            'pique 0.1 for PD version 23\n',
+            'this is pddplink 0.1, 3rd alpha build...\n',
+            'beware! this is tot 0.1, 19th alpha build...\n',
+            'foo: you have opened the [loadbang] help document\n',
+            'print: bang\n',
+            'print: 207\n',
+            'print: 2 1\n',
+            'obj3\n',
+            'obj4 34\n',
+            'initial_bang: bang\n',
+            '\n'
+            ]
+        ignorepatterns = [
+            'ydegoyon@free.fr',
+            'olaf.matthes@gmx.de',
+            'Olaf.*Matthes',
+            '[a-z]+ v0\.[0-9]',
+            'IOhannes m zm',
+            'part of zexy-',
+            'Pd: 0.43.1-extended',
+            'based on sync from jMax'
+            ]
+        for ignore in ignorelines:
+            try:
+                list.remove(ignore)
+            except ValueError:
+                pass
+        for line in list:
+            for pattern in ignorepatterns:
+                m = re.search('.*' + pattern + '.*', line)
+                while m:
+                    try:
+                        list.remove(m.string)
+                        m = re.search('.*' + pattern + '.*', line)
+                    except ValueError:
+                        break
+        return list
+
+
+    def runtest(self, log, root, filename):
+        patchoutput = []
+        patch = os.path.join(root, filename)
+        self.port = int(self.rand.random() * 10000) + int(self.rand.random() * 10000) + 40000
+        self.make_netreceive_patch(self.port)
+        p = self.launch_pd()
+        try:
+            self.open_patch(patch)
+            time.sleep(5)
+            self.close_patch(patch)
+            self.quit_pd(p)
+        except socket.error:
+            patchoutput.append('socket.error')                 
+        while True:
+            line = p.stdout.readline()
+            m = re.search('EOF on socket', line)
+            if not m and line:
+                patchoutput.append(line)
+            else:
+                break
+        patchoutput = self.remove_ignorelines(patchoutput)
+        if len(patchoutput) > 0:
+            log.write('\n\n__________________________________________________\n')
+            log.write('loading: ' + patch + '\n')
+            for line in patchoutput:
+                log.write(line)
+            log.flush()
+
+class TestLog():
+
+    def __init__(self, logfile):
+        self.logfile = logfile
+        self.rsynclog = []
+        self.separator = '______________________________________________________________________\n\n'
+
+    def upload(self):
+        # upload the log file to the autobuild website
+        date = time.strftime('20%y-%m-%d', time.localtime(time.time()))
+        logfilename = os.path.basename(self.logfile)
+        rsyncfile = str('rsync://128.238.56.50/upload/'
+                        + date + '/logs/' + logfilename)
+        cmd = ['rsync', '-axv', self.logfile, rsyncfile]
+        self.rsynclog.append(' '.join(cmd) + '\n')
+        p = subprocess.Popen(cmd,
+                             shell=False,
+                             stdout=subprocess.PIPE,
+                             stderr=subprocess.PIPE)
+        p.wait()
+        self.rsynclog.append(''.join(p.stdout.readlines()))
+        self.rsynclog.append(''.join(p.stderr.readlines()))
+        self.rsynclog.append(self.separator)
+        self.rsynclog.append('Complete log:\n')
+        self.rsynclog.append('http://autobuild.puredata.info/auto-build/' + date + '/logs/'
+                             + logfilename + '\n')
+        self.rsynclog.append(self.separator)
+
+    def email(self,
+              subject='PdTest results',
+              fromaddr='pd@pdlab.idmi.poly.edu',
+              toaddr='hans@at.or.at',
+              smtpserver='in1.smtp.messagingengine.com'):
+        server = smtplib.SMTP(smtpserver)
+        header = []
+        header.append('From: ' + fromaddr + '\n')
+        header.append('To: ' + toaddr + '\n')
+        header.append('Subject: ' + subject + '\n')
+        logoutput = []
+        fd = open(self.logfile, 'r')
+        for line in fd:
+            logoutput.append(line)
+        server.sendmail(fromaddr, toaddr, ''.join(header + self.rsynclog + logoutput))
+        server.quit()
+
+
+#------------------------------------------------------------------------------#
+# for testing from the command line:
+def main(argv):
+    print('nothing here yet')
+
+if __name__ == "__main__":
+    main(sys.argv[1:])