1

"Commons definitions for sikuli based beremiz IDE GUI tests"

     2

     3

import os

     4

import sys

     5

import subprocess

     6

import traceback

     7

import signal

     8

import re

     9

from threading import Thread, Event, Lock

    10

from time import time as timesec

    11

from xml.sax.saxutils import escape as escape_xml

    12

    13

import sikuli

    14

    15

beremiz_path = os.environ["BEREMIZPATH"]

    16

python_bin = os.environ.get("BEREMIZPYTHONPATH", "/usr/bin/python")

    17

opj = os.path.join

    18

    19

tessdata_path = os.environ["TESSDATAPATH"]

    20

    21

ansi_escape = re.compile(r'(?:\x1B[@-_]|[\x80-\x9F])[0-?]*[ -/]*[@-~]')

    22

def escape_ansi(line):

    23

    return ansi_escape.sub('', line)

    24

    25

def escape(txt):

    26

    return escape_xml(escape_ansi(txt))

    27

    28

class KBDShortcut:

    29

    """Send shortut to app by calling corresponding methods.

    30

    31

    example:

    32

        k = KBDShortcut()

    33

        k.Clean()

    34

    """

    35

    36

    fkeys = {"Stop":     sikuli.Key.F4,

    37

             "Run":      sikuli.Key.F5,

    38

             "Transfer": sikuli.Key.F6,

    39

             "Connect":  sikuli.Key.F7,

    40

             "Clean":    sikuli.Key.F9,

    41

             "Build":    sikuli.Key.F11,

    42

             "Save":     ("s",sikuli.Key.CTRL),

    43

             "New":      ("n",sikuli.Key.CTRL),

    44

             "Address":  ("l",sikuli.Key.CTRL)}  # to reach address bar in GTK's file selector

    45

    46

    def __init__(self, app):

    47

        self.app = app

    48

    49

    def __getattr__(self, name):

    50

        fkey = self.fkeys[name]

    51

        if type(fkey) != tuple:

    52

            fkey = (fkey,)

    53

    54

        def PressShortCut():

    55

            self.app.sikuliapp.focus()

    56

            sikuli.type(*fkey)

    57

            self.app.ReportText("Sending " + name + " shortcut")

    58

    59

        return PressShortCut

    60

    61

    62

class IDEIdleObserver:

    63

    "Detects when IDE is idle. This is particularly handy when staring an operation and witing for the en of it."

    64

    65

    def __init__(self):

    66

        """

    67

        Parameters: 

    68

            app (class BeremizApp)

    69

        """

    70

        self.r = sikuli.Region(self.sikuliapp.window())

    71

        self.targetOffset = self.r.getTopLeft()

    72

    73

        self.idechanged = False

    74

    75

        # 200 was selected because default 50 was still catching cursor blinking in console

    76

        # FIXME : remove blinking cursor in console

    77

        self.r.onChange(200,self._OnIDEWindowChange)

    78

        self.r.observeInBackground()

    79

    80

    def __del__(self):

    81

        self.r.stopObserver()

    82

    83

    def _OnIDEWindowChange(self, event):

    84

        self.idechanged = True

    85

    86

    def WaitIdleUI(self, period=1, timeout=15):

    87

        """

    88

        Wait for IDE to stop changing

    89

        Parameters: 

    90

            period (int): how many seconds with no change to consider idle

    91

            timeout (int): how long to wait for idle, in seconds

    92

        """

    93

        c = max(timeout/period,1)

    94

        while c > 0:

    95

            self.idechanged = False

    96

            sikuli.wait(period)

    97

            if not self.idechanged:

    98

                break

    99

            c = c - 1

   100

   101

        self.ReportScreenShot("UI is idle" if c != 0 else "UI is not idle")

   102

   103

        if c == 0:

   104

            raise Exception("Window did not idle before timeout")

   105

   106

   107

class stdoutIdleObserver:

   108

    "Detects when IDE's stdout is idle. Can be more reliable than pixel based version (false changes ?)"

   109

   110

    def __init__(self):

   111

        """

   112

        Parameters: 

   113

            app (class BeremizApp)

   114

        """

   115

        self.stdoutchanged = False

   116

   117

        self.event = Event()

   118

   119

        self.pattern = None

   120

        self.success_event = Event()

   121

   122

        if self.proc is not None:

   123

            self.thread = Thread(target = self._waitStdoutProc).start()

   124

   125

    def __del__(self):

   126

        pass  # self.thread.join() ?

   127

   128

    def _waitStdoutProc(self):

   129

        while True:

   130

            a = self.proc.stdout.readline()

   131

            if len(a) == 0 or a is None: 

   132

                break

   133

            self.ReportOutput(a)

   134

            self.event.set()

   135

            if self.pattern is not None and a.find(self.pattern) >= 0:

   136

                sys.stdout.write("found pattern in '" + a +"'")

   137

                self.success_event.set()

   138

   139

    def waitForChangeAndIdleStdout(self, period=2, timeout=15):

   140

        """

   141

        Wait for IDE'stdout to start changing

   142

        Parameters: 

   143

            timeout (int): how long to wait for change, in seconds

   144

        """

   145

        start_time = timesec()

   146

   147

        wait_result = self.event.wait(timeout)

   148

   149

        self.ReportScreenShot("stdout changed" if wait_result else "stdout didn't change")

   150

   151

        if wait_result:

   152

            self.event.clear()

   153

        else:

   154

            raise Exception("Stdout didn't become active before timeout")

   155

   156

        self.waitIdleStdout(period, timeout - (timesec() - start_time))

   157

   158

    def waitIdleStdout(self, period=2, timeout=15):

   159

        """

   160

        Wait for IDE'stdout to stop changing

   161

        Parameters: 

   162

            period (int): how many seconds with no change to consider idle

   163

            timeout (int): how long to wait for idle, in seconds

   164

        """

   165

        end_time = timesec() + timeout

   166

        self.event.clear()

   167

        while timesec() < end_time:

   168

            if self.event.wait(period):

   169

                # no timeout -> got event -> not idle -> loop again

   170

                self.event.clear()

   171

            else:

   172

                # timeout -> no event -> idle -> exit

   173

                self.ReportScreenShot("stdout is idle")

   174

                return True

   175

   176

        self.ReportScreenShot("stdout did not idle")

   177

   178

        raise Exception("Stdout did not idle before timeout")

   179

   180

    def waitPatternInStdout(self, pattern, timeout, count=1):

   181

        found = 0

   182

        self.pattern = pattern

   183

        end_time = timesec() + timeout

   184

        while True:

   185

            remain = end_time - timesec()

   186

            if remain <= 0 :

   187

                res = False

   188

                break

   189

   190

            res = self.success_event.wait(remain)

   191

            if res:

   192

                self.success_event.clear()

   193

                found = found + 1

   194

                if found >= count:

   195

                    break

   196

        self.pattern = None

   197

        self.ReportScreenShot("found pattern" if res else "pattern not found")

   198

        return res

   199

   200

class BeremizApp(IDEIdleObserver, stdoutIdleObserver):

   201

    def __init__(self, projectpath=None, exemple=None, testproject=None):

   202

        """

   203

        Starts Beremiz IDE, waits for main window to appear, maximize it.

   204

   205

            Parameters: 

   206

                projectpath (str): path to project to open

   207

                exemple (str): path relative to exemples directory

   208

   209

            Returns:

   210

                Sikuli App class instance

   211

        """

   212

        self.ocropts = sikuli.OCR.globalOptions()

   213

        self.ocropts.dataPath(tessdata_path)

   214

   215

        # 0 use legacy Tesseract (not so good, but repeatable)

   216

        # 1 use RNN Tesseract (better but non-repeatable)

   217

        # 2 use both

   218

        self.ocropts.oem(0)

   219

        self.ocropts.psm(12)

   220

        self.ocropts.smallFont()

   221

   222

        self.imgnum = 0

   223

        self.starttime = timesec()

   224

        self.screen = sikuli.Screen()

   225

   226

        self.report = open("report.xhtml", "w")

   227

        self.report.write("""\

   228

<html xmlns="http://www.w3.org/1999/xhtml">

   229

  <head>

   230

    <meta charset="utf-8"/>

   231

    <meta name="color-scheme" content="light dark"/>

   232

    <title>Test report</title>

   233

  </head>

   234

  <body>

   235

""")

   236

   237

        command = ["setsid", python_bin, opj(beremiz_path,"Beremiz.py"), "--log=/dev/stdout"]

   238

   239

        if exemple is not None:

   240

            command.append(opj(beremiz_path,"exemples",exemple))

   241

        elif projectpath is not None:

   242

            command.append(projectpath)

   243

        elif testproject is not None:

   244

            command.append(opj(beremiz_path,"tests","projects",testproject))

   245

   246

   247

        # App class is broken in Sikuli 2.0.5: can't start process with arguments.

   248

        # 

   249

        # Workaround : - use subprocess module to spawn IDE process,

   250

        #              - use wmctrl to find IDE window details and maximize it

   251

        #              - pass exact window title to App class constructor

   252

   253

        self.ReportText("Launching " + repr(command))

   254

   255

        self.proc = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=0)

   256

   257

        # Window are macthed against process' PID

   258

        ppid = self.proc.pid

   259

   260

        # Timeout 5s

   261

        c = 50

   262

        while c > 0:

   263

            # equiv to "wmctrl -l -p | grep $pid"

   264

            try:

   265

                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()))

   266

            except subprocess.CalledProcessError:

   267

                wlist = []

   268

   269

            # window with no title only has 4 fields do describe it

   270

            # beremiz splashcreen has no title

   271

            # wait until main window is visible

   272

            if len(wlist) == 1 and len(wlist[0]) == 5:

   273

                windowID,_zero,wpid,_XID,wtitle = wlist[0] 

   274

                break

   275

   276

            sikuli.wait(0.1)

   277

            c = c - 1

   278

   279

        if c == 0:

   280

            raise Exception("Couldn't find Beremiz window")

   281

   282

        # Maximize window x and y

   283

        subprocess.check_call(["wmctrl", "-i", "-r", windowID, "-b", "add,maximized_vert,maximized_horz"])

   284

   285

        # switchApp creates an App object by finding window by title, is not supposed to spawn a process

   286

        self.sikuliapp = sikuli.switchApp(wtitle)

   287

        self.k = KBDShortcut(self)

   288

   289

        IDEIdleObserver.__init__(self)

   290

        stdoutIdleObserver.__init__(self)

   291

   292

        # stubs for common sikuli calls to allow adding hooks later

   293

        for name, takes_matches in [

   294

            ("click", True),

   295

            ("doubleClick", True),

   296

            ("type", False),

   297

            ("rightClick", True),

   298

            ("wait", False)]:

   299

            def makeMyMeth(n,m):

   300

                def myMeth(*args, **kwargs):

   301

                    self.ReportScreenShot("Begin: " + n + "(" + repr(args) + "," + repr(kwargs) + ")")

   302

                    if m:

   303

                        args = map(self.handle_PFRML_arg, args)

   304

                        kwargs = dict(map(lambda k,v:(k,self.handle_PFRML_arg(v)), kwargs.items()))

   305

                    try:

   306

                        getattr(sikuli, n)(*args, **kwargs)

   307

                    finally:

   308

                        self.ReportScreenShot("end: " + n + "(" + repr(args) + "," + repr(kwargs) + ")")

   309

                return myMeth

   310

            setattr(self, name, makeMyMeth(name,takes_matches))

   311

   312

    def handle_PFRML_arg(self, arg):

   313

        if type(arg)==list:

   314

            return self.findBest(*arg)

   315

        if type(arg)==str and not arg.endswith(".png"):

   316

            return self.findBest(arg)

   317

        return arg

   318

   319

    def findBest(self, *args):

   320

        #match = self.r.findBest(*args)

   321

        match = None 

   322

        matches = sikuli.OCR.readWords(self.r) + sikuli.OCR.readLines(self.r)

   323

        for m in matches:

   324

            mText = m.getText().encode('ascii', 'ignore')

   325

            for arg in args:

   326

                if arg in mText or arg in mText.translate(None, "\"`'|-. "):

   327

                    if match is None:

   328

                        match = m

   329

                    if mText == arg:

   330

                        match = m

   331

                        break

   332

        if match is None:

   333

            self.ReportText("Not found: " + repr(args) + " OCR content: ")

   334

            for m in matches:

   335

                self.ReportText(repr(m) + ": " + m.getText().encode('ascii', 'ignore'))

   336

            raise Exception("Not Found: " + repr(args))

   337

   338

        # translate match to screen ref

   339

        #match.setTargetOffset(self.targetOffset)

   340

        match.setTopLeft(match.getTopLeft().offset(self.targetOffset))

   341

   342

        self.ReportTextImage("Found for " + repr(args) + ": " +

   343

            " ".join([repr(match), repr(match.getTarget()), repr(match.getTargetOffset())]),

   344

            self.screen.capture(match))

   345

        return match.getTarget()

   346

   347

    def dragNdrop(self, src, dst):

   348

        self.ReportScreenShot("Drag: (" + repr(src) + ")")

   349

        sikuli.drag(self.handle_PFRML_arg(src))

   350

        sikuli.mouseMove(5,0)

   351

        sikuli.dropAt(self.handle_PFRML_arg(dst))

   352

        self.ReportScreenShot("Drop: (" + repr(dst) + ")")

   353

   354

    def close(self):

   355

   356

        self.ReportScreenShot("Close app")

   357

        os.kill(self.proc.pid, signal.SIGKILL)

   358

        #self.sikuliapp.close()

   359

        #self.sikuliapp = None

   360

   361

        self.report.write("""

   362

  </body>

   363

</html>""")

   364

        self.report.close()

   365

   366

    def __del__(self):

   367

        if self.sikuliapp is not None:

   368

            self.sikuliapp.close()

   369

        IDEIdleObserver.__del__(self)

   370

        stdoutIdleObserver.__del__(self)

   371

   372

    def ReportScreenShot(self, msg):

   373

        cap = self.screen.capture(self.r)

   374

        self.ReportTextImage(msg, cap)

   375

   376

    def ReportTextImage(self, msg, img):

   377

        elapsed = "%.3fs: "%(timesec() - self.starttime)

   378

        fname = "capture"+str(self.imgnum)+".png"

   379

        sys.stdout.write(elapsed + " [" + fname + "] " + msg + "\n")

   380

        img.save(".", fname)

   381

        self.imgnum = self.imgnum + 1

   382

        self.report.write( "<p>" + escape(elapsed + msg) + "<br/><img src=\""+ fname + "\"/>" + "</p>")

   383

   384

    def ReportText(self, text):

   385

        elapsed = "%.3fs: "%(timesec() - self.starttime)

   386

        #res = u"<p><![CDATA[" + elapsed + text + "]]></p>"

   387

        sys.stdout.write(elapsed + text + "\n")

   388

        res = u"<p>" + escape(elapsed + text) + "</p>"

   389

        self.report.write(res)

   390

   391

    def ReportOutput(self, text):

   392

        elapsed = "%.3fs: "%(timesec() - self.starttime)

   393

        sys.stdout.write(elapsed + text)

   394

        self.report.write("<pre>" + escape(elapsed + text) + "</pre>")

   395

   396

   397

class AuxiliaryProcess(stdoutIdleObserver):

   398

    def __init__(self, beremiz_app, command):

   399

        self.app = beremiz_app

   400

        self.app.ReportText("Launching process " + repr(command))

   401

        self.proc = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=0)

   402

        self.app.ReportText("Launched process " + repr(command) + " PID: " + str(self.proc.pid))

   403

        stdoutIdleObserver.__init__(self)

   404

   405

    def close(self):

   406

        if self.proc is not None:

   407

            proc = self.proc

   408

            self.proc = None

   409

            # self.proc.stdout.close()

   410

            self.app.ReportText("Kill process PID: " + str(proc.pid))

   411

            try:

   412

                os.kill(proc.pid, signal.SIGTERM)

   413

            except OSError:

   414

                pass

   415

            proc.wait()

   416

            # self.thread.join()

   417

   418

    def ReportOutput(self, text):

   419

        self.app.ReportOutput("Aux: "+text)

   420

   421

    def ReportScreenShot(self, msg):

   422

        self.app.ReportOutput("Aux: "+msg)

   423

   424

    def __del__(self):

   425

        self.close()

   426

   427

def run_test(func, *args, **kwargs):

   428

    app = BeremizApp(*args, **kwargs)

   429

    try:

   430

        success = func(app)

   431

    except:

   432

        # sadly, sys.excepthook is broken in sikuli/jython 

   433

        # purpose of this run_test function is to work around it.

   434

        # and catch exception cleanly anyhow

   435

        e_type, e_value, e_traceback = sys.exc_info()

   436

        err_msg = "\n".join(traceback.format_exception(e_type, e_value, e_traceback))

   437

        app.ReportOutput(err_msg)

   438

        success = False

   439

   440

    app.close()

   441

   442

    if success:

   443

        sikuli.exit(0)

   444

    else:

   445

        sikuli.exit(1)

   446

   447

   448