edouard@3424: "Commons definitions for sikuli based beremiz IDE GUI tests"
edouard@3424: 
edouard@3424: import os
edouard@3424: import sys
edouard@3424: import subprocess
edouard@3447: import traceback
edouard@3668: import signal
edouard@3710: import re
edouard@3446: from threading import Thread, Event, Lock
edouard@3437: from time import time as timesec
edouard@3710: from xml.sax.saxutils import escape as escape_xml
edouard@3432: 
edouard@3446: import sikuli
edouard@3424: 
edouard@3424: beremiz_path = os.environ["BEREMIZPATH"]
edouard@3431: python_bin = os.environ.get("BEREMIZPYTHONPATH", "/usr/bin/python")
edouard@3424: opj = os.path.join
edouard@3424: 
edouard@3696: tessdata_path = os.environ["TESSDATAPATH"]
edouard@3424: 
edouard@3710: ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')
edouard@3710: def escape_ansi(line):
edouard@3710:     return ansi_escape.sub('', line)
edouard@3710: 
edouard@3710: def escape(txt):
edouard@3710:     return escape_xml(escape_ansi(txt))
edouard@3710: 
edouard@3424: class KBDShortcut:
edouard@3446:     """Send shortut to app by calling corresponding methods.
edouard@3424: 
edouard@3424:     example:
edouard@3446:         k = KBDShortcut()
edouard@3424:         k.Clean()
edouard@3424:     """
edouard@3424: 
edouard@3446:     fkeys = {"Stop":     sikuli.Key.F4,
edouard@3446:              "Run":      sikuli.Key.F5,
edouard@3446:              "Transfer": sikuli.Key.F6,
edouard@3446:              "Connect":  sikuli.Key.F7,
edouard@3446:              "Clean":    sikuli.Key.F9,
edouard@3446:              "Build":    sikuli.Key.F11,
edouard@3446:              "Save":     ("s",sikuli.Key.CTRL),
edouard@3446:              "New":      ("n",sikuli.Key.CTRL),
edouard@3446:              "Address":  ("l",sikuli.Key.CTRL)}  # to reach address bar in GTK's file selector
edouard@3424: 
edouard@3424:     def __init__(self, app):
edouard@3447:         self.app = app
edouard@3424:     
edouard@3424:     def __getattr__(self, name):
edouard@3424:         fkey = self.fkeys[name]
edouard@3446:         if type(fkey) != tuple:
edouard@3432:             fkey = (fkey,)
edouard@3424: 
edouard@3424:         def PressShortCut():
edouard@3447:             self.app.sikuliapp.focus()
edouard@3446:             sikuli.type(*fkey)
edouard@3447:             self.app.ReportText("Sending " + name + " shortcut")
edouard@3424: 
edouard@3424:         return PressShortCut
edouard@3424: 
edouard@3424: 
edouard@3424: class IDEIdleObserver:
edouard@3424:     "Detects when IDE is idle. This is particularly handy when staring an operation and witing for the en of it."
edouard@3424: 
edouard@3446:     def __init__(self):
edouard@3446:         """
edouard@3446:         Parameters: 
edouard@3446:             app (class BeremizApp)
edouard@3446:         """
edouard@3446:         self.r = sikuli.Region(self.sikuliapp.window())
edouard@3710:         self.targetOffset = self.r.getTopLeft()
edouard@3424: 
edouard@3424:         self.idechanged = False
edouard@3424:         
edouard@3424:         # 200 was selected because default 50 was still catching cursor blinking in console
edouard@3424:         # FIXME : remove blinking cursor in console
edouard@3424:         self.r.onChange(200,self._OnIDEWindowChange)
edouard@3424:         self.r.observeInBackground()
edouard@3424: 
edouard@3424:     def __del__(self):
edouard@3424:         self.r.stopObserver()
edouard@3424: 
edouard@3424:     def _OnIDEWindowChange(self, event):
edouard@3424:         self.idechanged = True
edouard@3424: 
edouard@3446:     def WaitIdleUI(self, period=1, timeout=15):
edouard@3424:         """
edouard@3424:         Wait for IDE to stop changing
edouard@3424:         Parameters: 
edouard@3424:             period (int): how many seconds with no change to consider idle
edouard@3424:             timeout (int): how long to wait for idle, in seconds
edouard@3424:         """
edouard@3437:         c = max(timeout/period,1)
edouard@3424:         while c > 0:
edouard@3424:             self.idechanged = False
edouard@3446:             sikuli.wait(period)
edouard@3424:             if not self.idechanged:
edouard@3424:                 break
edouard@3424:             c = c - 1
edouard@3424: 
edouard@3447:         self.ReportScreenShot("UI is idle" if c != 0 else "UI is not idle")
edouard@3447: 
edouard@3424:         if c == 0:
edouard@3424:             raise Exception("Window did not idle before timeout")
edouard@3424: 
edouard@3424:  
edouard@3424: class stdoutIdleObserver:
edouard@3424:     "Detects when IDE's stdout is idle. Can be more reliable than pixel based version (false changes ?)"
edouard@3424: 
edouard@3446:     def __init__(self):
edouard@3446:         """
edouard@3446:         Parameters: 
edouard@3446:             app (class BeremizApp)
edouard@3446:         """
edouard@3424:         self.stdoutchanged = False
edouard@3424: 
edouard@3437:         self.event = Event()
edouard@3437: 
edouard@3446:         self.pattern = None
edouard@3446:         self.success_event = Event()
edouard@3446: 
edouard@3670:         if self.proc is not None:
edouard@3670:             self.thread = Thread(target = self._waitStdoutProc).start()
edouard@3424: 
edouard@3528:     def __del__(self):
edouard@3528:         pass  # self.thread.join() ?
edouard@3528: 
edouard@3432:     def _waitStdoutProc(self):
edouard@3424:         while True:
edouard@3437:             a = self.proc.stdout.readline()
edouard@3424:             if len(a) == 0 or a is None: 
edouard@3424:                 break
edouard@3447:             self.ReportOutput(a)
edouard@3437:             self.event.set()
edouard@3446:             if self.pattern is not None and a.find(self.pattern) >= 0:
edouard@3446:                 sys.stdout.write("found pattern in '" + a +"'")
edouard@3446:                 self.success_event.set()
edouard@3446: 
edouard@3446:     def waitForChangeAndIdleStdout(self, period=2, timeout=15):
edouard@3437:         """
edouard@3437:         Wait for IDE'stdout to start changing
edouard@3437:         Parameters: 
edouard@3437:             timeout (int): how long to wait for change, in seconds
edouard@3437:         """
edouard@3437:         start_time = timesec()
edouard@3446: 
edouard@3447:         wait_result = self.event.wait(timeout)
edouard@3447: 
edouard@3447:         self.ReportScreenShot("stdout changed" if wait_result else "stdout didn't change")
edouard@3447: 
edouard@3447:         if wait_result:
edouard@3446:             self.event.clear()
edouard@3446:         else:
edouard@3446:             raise Exception("Stdout didn't become active before timeout")
edouard@3446: 
edouard@3446:         self.waitIdleStdout(period, timeout - (timesec() - start_time))
edouard@3446: 
edouard@3446:     def waitIdleStdout(self, period=2, timeout=15):
edouard@3424:         """
edouard@3424:         Wait for IDE'stdout to stop changing
edouard@3424:         Parameters: 
edouard@3424:             period (int): how many seconds with no change to consider idle
edouard@3424:             timeout (int): how long to wait for idle, in seconds
edouard@3424:         """
edouard@3446:         end_time = timesec() + timeout
edouard@3446:         self.event.clear()
edouard@3446:         while timesec() < end_time:
edouard@3446:             if self.event.wait(period):
edouard@3446:                 # no timeout -> got event -> not idle -> loop again
edouard@3446:                 self.event.clear()
edouard@3446:             else:
edouard@3446:                 # timeout -> no event -> idle -> exit
edouard@3447:                 self.ReportScreenShot("stdout is idle")
edouard@3446:                 return True
edouard@3446: 
edouard@3447:         self.ReportScreenShot("stdout did not idle")
edouard@3447: 
edouard@3446:         raise Exception("Stdout did not idle before timeout")
edouard@3446: 
edouard@3446:     def waitPatternInStdout(self, pattern, timeout, count=1):
edouard@3424:         found = 0
edouard@3446:         self.pattern = pattern
edouard@3446:         end_time = timesec() + timeout
edouard@3424:         while True:
edouard@3446:             remain = end_time - timesec()
edouard@3446:             if remain <= 0 :
edouard@3446:                 res = False
edouard@3446:                 break
edouard@3446: 
edouard@3446:             res = self.success_event.wait(remain)
edouard@3446:             if res:
edouard@3446:                 self.success_event.clear()
edouard@3424:                 found = found + 1
edouard@3424:                 if found >= count:
edouard@3424:                     break
edouard@3446:         self.pattern = None
edouard@3447:         self.ReportScreenShot("found pattern" if res else "pattern not found")
edouard@3446:         return res
edouard@3446: 
edouard@3446: class BeremizApp(IDEIdleObserver, stdoutIdleObserver):
edouard@3669:     def __init__(self, projectpath=None, exemple=None, testproject=None):
edouard@3446:         """
edouard@3446:         Starts Beremiz IDE, waits for main window to appear, maximize it.
edouard@3446: 
edouard@3446:             Parameters: 
edouard@3446:                 projectpath (str): path to project to open
edouard@3446:                 exemple (str): path relative to exemples directory
edouard@3446: 
edouard@3446:             Returns:
edouard@3446:                 Sikuli App class instance
edouard@3446:         """
edouard@3710:         self.ocropts = sikuli.OCR.globalOptions()
edouard@3710:         self.ocropts.dataPath(tessdata_path)
edouard@3842:         
edouard@3842:         # 0 use legacy Tesseract (not so good, but repeatable)
edouard@3842:         # 1 use RNN Tesseract (better but non-repeatable)
edouard@3842:         # 2 use both
edouard@3842:         self.ocropts.oem(0)
edouard@3853:         self.ocropts.psm(12)
edouard@3710:         self.ocropts.smallFont()
edouard@3710: 
edouard@3710:         self.imgnum = 0
edouard@3447:         self.starttime = timesec()
edouard@3447:         self.screen = sikuli.Screen()
edouard@3447: 
edouard@3710:         self.report = open("report.xhtml", "w")
edouard@3710:         self.report.write("""\
edouard@3710: <html xmlns="http://www.w3.org/1999/xhtml">
edouard@3447:   <head>
edouard@3710:     <meta charset="utf-8"/>
edouard@3710:     <meta name="color-scheme" content="light dark"/>
edouard@3447:     <title>Test report</title>
edouard@3447:   </head>
edouard@3447:   <body>
edouard@3447: """)
edouard@3447: 
edouard@3839:         command = ["setsid", python_bin, opj(beremiz_path,"Beremiz.py"), "--log=/dev/stdout"]
edouard@3446: 
edouard@3446:         if exemple is not None:
edouard@3446:             command.append(opj(beremiz_path,"exemples",exemple))
edouard@3446:         elif projectpath is not None:
edouard@3446:             command.append(projectpath)
edouard@3669:         elif testproject is not None:
edouard@3669:             command.append(opj(beremiz_path,"tests","projects",testproject))
edouard@3669: 
edouard@3446: 
edouard@3446:         # App class is broken in Sikuli 2.0.5: can't start process with arguments.
edouard@3446:         # 
edouard@3446:         # Workaround : - use subprocess module to spawn IDE process,
edouard@3446:         #              - use wmctrl to find IDE window details and maximize it
edouard@3446:         #              - pass exact window title to App class constructor
edouard@3446: 
edouard@3447:         self.ReportText("Launching " + repr(command))
edouard@3447: 
edouard@3446:         self.proc = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=0)
edouard@3446: 
edouard@3446:         # Window are macthed against process' PID
edouard@3446:         ppid = self.proc.pid
edouard@3446: 
edouard@3446:         # Timeout 5s
edouard@3446:         c = 50
edouard@3446:         while c > 0:
edouard@3446:             # equiv to "wmctrl -l -p | grep $pid"
edouard@3446:             try:
edouard@3446:                 wlist = filter(lambda l:(len(l)>2 and l[2]==str(ppid)), map(lambda s:s.split(None,4), subprocess.check_output(["wmctrl", "-l", "-p"]).splitlines()))
edouard@3446:             except subprocess.CalledProcessError:
edouard@3446:                 wlist = []
edouard@3446: 
edouard@3446:             # window with no title only has 4 fields do describe it
edouard@3446:             # beremiz splashcreen has no title
edouard@3446:             # wait until main window is visible
edouard@3446:             if len(wlist) == 1 and len(wlist[0]) == 5:
edouard@3446:                 windowID,_zero,wpid,_XID,wtitle = wlist[0] 
edouard@3446:                 break
edouard@3446: 
edouard@3446:             sikuli.wait(0.1)
edouard@3446:             c = c - 1
edouard@3446: 
edouard@3446:         if c == 0:
edouard@3446:             raise Exception("Couldn't find Beremiz window")
edouard@3446: 
edouard@3446:         # Maximize window x and y
edouard@3446:         subprocess.check_call(["wmctrl", "-i", "-r", windowID, "-b", "add,maximized_vert,maximized_horz"])
edouard@3446: 
edouard@3446:         # switchApp creates an App object by finding window by title, is not supposed to spawn a process
edouard@3446:         self.sikuliapp = sikuli.switchApp(wtitle)
edouard@3446:         self.k = KBDShortcut(self)
edouard@3446: 
edouard@3446:         IDEIdleObserver.__init__(self)
edouard@3446:         stdoutIdleObserver.__init__(self)
edouard@3446: 
edouard@3446:         # stubs for common sikuli calls to allow adding hooks later
edouard@3710:         for name, takes_matches in [
edouard@3710:             ("click", True),
edouard@3710:             ("doubleClick", True),
edouard@3710:             ("type", False),
edouard@3710:             ("rightClick", True),
edouard@3710:             ("wait", False)]:
edouard@3710:             def makeMyMeth(n,m):
edouard@3447:                 def myMeth(*args, **kwargs):
edouard@3534:                     self.ReportScreenShot("Begin: " + n + "(" + repr(args) + "," + repr(kwargs) + ")")
edouard@3710:                     if m:
edouard@3710:                         args = map(self.handle_PFRML_arg, args)
edouard@3710:                         kwargs = dict(map(lambda k,v:(k,self.handle_PFRML_arg(v)), kwargs.items()))
edouard@3544:                     try:
edouard@3544:                         getattr(sikuli, n)(*args, **kwargs)
edouard@3544:                     finally:
edouard@3544:                         self.ReportScreenShot("end: " + n + "(" + repr(args) + "," + repr(kwargs) + ")")
edouard@3447:                 return myMeth
edouard@3710:             setattr(self, name, makeMyMeth(name,takes_matches))
edouard@3710: 
edouard@3710:     def handle_PFRML_arg(self, arg):
edouard@3710:         if type(arg)==list:
edouard@3710:             return self.findBest(*arg)
edouard@3710:         if type(arg)==str and not arg.endswith(".png"):
edouard@3710:             return self.findBest(arg)
edouard@3710:         return arg
edouard@3710: 
edouard@3710:     def findBest(self, *args):
edouard@3710:         #match = self.r.findBest(*args)
edouard@3710:         match = None 
edouard@3710:         matches = sikuli.OCR.readWords(self.r) + sikuli.OCR.readLines(self.r)
edouard@3710:         for m in matches:
edouard@3710:             mText = m.getText().encode('ascii', 'ignore')
edouard@3710:             for arg in args:
edouard@3842:                 if arg in mText or arg in mText.translate(None, "\"`'|-. "):
edouard@3710:                     if match is None:
edouard@3710:                         match = m
edouard@3710:                     if mText == arg:
edouard@3710:                         match = m
edouard@3710:                         break
edouard@3710:         if match is None:
edouard@3710:             self.ReportText("Not found: " + repr(args) + " OCR content: ")
edouard@3710:             for m in matches:
edouard@3710:                 self.ReportText(repr(m) + ": " + m.getText().encode('ascii', 'ignore'))
edouard@3710:             raise Exception("Not Found: " + repr(args))
edouard@3710:         
edouard@3710:         # translate match to screen ref
edouard@3710:         #match.setTargetOffset(self.targetOffset)
edouard@3710:         match.setTopLeft(match.getTopLeft().offset(self.targetOffset))
edouard@3710: 
edouard@3710:         self.ReportTextImage("Found for " + repr(args) + ": " +
edouard@3710:             " ".join([repr(match), repr(match.getTarget()), repr(match.getTargetOffset())]),
edouard@3710:             self.screen.capture(match))
edouard@3710:         return match.getTarget()
edouard@3446: 
edouard@3671:     def dragNdrop(self, src, dst):
edouard@3710:         self.ReportScreenShot("Drag: (" + repr(src) + ")")
edouard@3710:         sikuli.drag(self.handle_PFRML_arg(src))
edouard@3671:         sikuli.mouseMove(5,0)
edouard@3710:         sikuli.dropAt(self.handle_PFRML_arg(dst))
edouard@3710:         self.ReportScreenShot("Drop: (" + repr(dst) + ")")
edouard@3671: 
edouard@3446:     def close(self):
edouard@3713: 
edouard@3713:         self.ReportScreenShot("Close app")
edouard@3839:         os.kill(self.proc.pid, signal.SIGKILL)
edouard@3713:         #self.sikuliapp.close()
edouard@3713:         #self.sikuliapp = None
edouard@3713: 
edouard@3447:         self.report.write("""
edouard@3447:   </body>
edouard@3447: </html>""")
edouard@3447:         self.report.close()
edouard@3446: 
edouard@3446:     def __del__(self):
edouard@3446:         if self.sikuliapp is not None:
edouard@3446:             self.sikuliapp.close()
edouard@3446:         IDEIdleObserver.__del__(self)
edouard@3446:         stdoutIdleObserver.__del__(self)
edouard@3446: 
edouard@3447:     def ReportScreenShot(self, msg):
edouard@3710:         cap = self.screen.capture(self.r)
edouard@3710:         self.ReportTextImage(msg, cap)
edouard@3710: 
edouard@3710:     def ReportTextImage(self, msg, img):
edouard@3447:         elapsed = "%.3fs: "%(timesec() - self.starttime)
edouard@3710:         fname = "capture"+str(self.imgnum)+".png"
edouard@3839:         sys.stdout.write(elapsed + " [" + fname + "] " + msg + "\n")
edouard@3710:         img.save(".", fname)
edouard@3710:         self.imgnum = self.imgnum + 1
edouard@3710:         self.report.write( "<p>" + escape(elapsed + msg) + "<br/><img src=\""+ fname + "\"/>" + "</p>")
edouard@3447: 
edouard@3447:     def ReportText(self, text):
edouard@3447:         elapsed = "%.3fs: "%(timesec() - self.starttime)
edouard@3710:         #res = u"<p><![CDATA[" + elapsed + text + "]]></p>"
edouard@3839:         sys.stdout.write(elapsed + text + "\n")
edouard@3710:         res = u"<p>" + escape(elapsed + text) + "</p>"
edouard@3710:         self.report.write(res)
edouard@3447: 
edouard@3447:     def ReportOutput(self, text):
edouard@3447:         elapsed = "%.3fs: "%(timesec() - self.starttime)
edouard@3675:         sys.stdout.write(elapsed + text)
edouard@3710:         self.report.write("<pre>" + escape(elapsed + text) + "</pre>")
edouard@3447: 
edouard@3447: 
edouard@3675: class AuxiliaryProcess(stdoutIdleObserver):
edouard@3668:     def __init__(self, beremiz_app, command):
edouard@3668:         self.app = beremiz_app
edouard@3668:         self.app.ReportText("Launching process " + repr(command))
edouard@3668:         self.proc = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=0)
edouard@3668:         self.app.ReportText("Launched process " + repr(command) + " PID: " + str(self.proc.pid))
edouard@3675:         stdoutIdleObserver.__init__(self)
edouard@3668: 
edouard@3668:     def close(self):
edouard@3668:         if self.proc is not None:
edouard@3668:             proc = self.proc
edouard@3668:             self.proc = None
edouard@3668:             # self.proc.stdout.close()
edouard@3668:             self.app.ReportText("Kill process PID: " + str(proc.pid))
edouard@3668:             try:
edouard@3668:                 os.kill(proc.pid, signal.SIGTERM)
edouard@3668:             except OSError:
edouard@3668:                 pass
edouard@3668:             proc.wait()
edouard@3668:             # self.thread.join()
edouard@3668: 
edouard@3675:     def ReportOutput(self, text):
edouard@3675:         self.app.ReportOutput("Aux: "+text)
edouard@3675: 
edouard@3675:     def ReportScreenShot(self, msg):
edouard@3675:         self.app.ReportOutput("Aux: "+msg)
edouard@3675: 
edouard@3668:     def __del__(self):
edouard@3668:         self.close()
edouard@3668: 
edouard@3447: def run_test(func, *args, **kwargs):
edouard@3447:     app = BeremizApp(*args, **kwargs)
edouard@3447:     try:
edouard@3447:         success = func(app)
edouard@3447:     except:
edouard@3447:         # sadly, sys.excepthook is broken in sikuli/jython 
edouard@3447:         # purpose of this run_test function is to work around it.
edouard@3447:         # and catch exception cleanly anyhow
edouard@3447:         e_type, e_value, e_traceback = sys.exc_info()
edouard@3447:         err_msg = "\n".join(traceback.format_exception(e_type, e_value, e_traceback))
edouard@3447:         app.ReportOutput(err_msg)
edouard@3447:         success = False
edouard@3447: 
edouard@3447:     app.close()
edouard@3447: 
edouard@3447:     if success:
edouard@3447:         sikuli.exit(0)
edouard@3447:     else:
edouard@3447:         sikuli.exit(1)
edouard@3447: 
edouard@3447: 
edouard@3447: