bin/convwatch.py

   1 # -*- tab-width: 4; indent-tabs-mode: nil; py-indent-offset: 4 -*-
   2 #
   3 # This file is part of the LibreOffice project.
   4 #
   5 # This Source Code Form is subject to the terms of the Mozilla Public
   6 # License, v. 2.0. If a copy of the MPL was not distributed with this
   7 # file, You can obtain one at http://mozilla.org/MPL/2.0/.
   8 #
   9
  10 import getopt
  11 import os
  12 import subprocess
  13 import sys
  14 import time
  15 import uuid
  16 try:
  17     from urllib.parse import quote
  18 except ImportError:
  19     from urllib import quote
  20
  21 try:
  22     import pyuno
  23     import uno
  24     import unohelper
  25 except ImportError:
  26     print("pyuno not found: try to set PYTHONPATH and URE_BOOTSTRAP variables")
  27     print("PYTHONPATH=/installation/opt/program")
  28     print("URE_BOOTSTRAP=file:///installation/opt/program/fundamentalrc")
  29     raise
  30
  31 try:
  32     from com.sun.star.document import XDocumentEventListener
  33 except ImportError:
  34     print("UNO API class not found: try to set URE_BOOTSTRAP variable")
  35     print("URE_BOOTSTRAP=file:///installation/opt/program/fundamentalrc")
  36     raise
  37
  38 ### utilities ###
  39
  40 def partition(list, pred):
  41     left = []
  42     right = []
  43     for e in list:
  44         if pred(e):
  45             left.append(e)
  46         else:
  47             right.append(e)
  48     return (left, right)
  49
  50 def filelist(dir, suffix):
  51     if len(dir) == 0:
  52         raise Exception("filelist: empty directory")
  53     if not(dir[-1] == "/"):
  54         dir += "/"
  55     files = [dir + f for f in os.listdir(dir)]
  56 #    print(files)
  57     return [f for f in files
  58                     if os.path.isfile(f) and os.path.splitext(f)[1] == suffix]
  59
  60 def getFiles(dirs, suffix):
  61     files = []
  62     for dir in dirs:
  63         files += filelist(dir, suffix)
  64     return files
  65
  66 ### UNO utilities ###
  67
  68 class OfficeConnection:
  69     def __init__(self, args):
  70         self.args = args
  71         self.soffice = None
  72         self.socket = None
  73         self.xContext = None
  74     def setUp(self):
  75         (method, sep, rest) = self.args["--soffice"].partition(":")
  76         if sep != ":":
  77             raise Exception("soffice parameter does not specify method")
  78         if method == "path":
  79                 socket = "pipe,name=pytest" + str(uuid.uuid1())
  80                 try:
  81                     userdir = self.args["--userdir"]
  82                 except KeyError:
  83                     raise Exception("'path' method requires --userdir")
  84                 if not(userdir.startswith("file://")):
  85                     raise Exception("--userdir must be file URL")
  86                 self.soffice = self.bootstrap(rest, userdir, socket)
  87         elif method == "connect":
  88                 socket = rest
  89         else:
  90             raise Exception("unsupported connection method: " + method)
  91         self.xContext = self.connect(socket)
  92
  93     def bootstrap(self, soffice, userdir, socket):
  94         argv = [ soffice, "--accept=" + socket + ";urp",
  95                 "-env:UserInstallation=" + userdir,
  96                 "--quickstart=no", "--nofirststartwizard",
  97                 "--norestore", "--nologo", "--headless" ]
  98         if "--valgrind" in self.args:
  99             argv.append("--valgrind")
 100         return subprocess.Popen(argv)
 101
 102     def connect(self, socket):
 103         xLocalContext = uno.getComponentContext()
 104         xUnoResolver = xLocalContext.ServiceManager.createInstanceWithContext(
 105                 "com.sun.star.bridge.UnoUrlResolver", xLocalContext)
 106         url = "uno:" + socket + ";urp;StarOffice.ComponentContext"
 107         print("OfficeConnection: connecting to: " + url)
 108         while True:
 109             try:
 110                 xContext = xUnoResolver.resolve(url)
 111                 return xContext
 112 #            except com.sun.star.connection.NoConnectException
 113             except pyuno.getClass("com.sun.star.connection.NoConnectException"):
 114                 print("NoConnectException: sleeping...")
 115                 time.sleep(1)
 116
 117     def tearDown(self):
 118         if self.soffice:
 119             if self.xContext:
 120                 try:
 121                     print("tearDown: calling terminate()...")
 122                     xMgr = self.xContext.ServiceManager
 123                     xDesktop = xMgr.createInstanceWithContext(
 124                             "com.sun.star.frame.Desktop", self.xContext)
 125                     xDesktop.terminate()
 126                     print("...done")
 127 #                except com.sun.star.lang.DisposedException:
 128                 except pyuno.getClass("com.sun.star.beans.UnknownPropertyException"):
 129                     print("caught UnknownPropertyException")
 130                     pass # ignore, also means disposed
 131                 except pyuno.getClass("com.sun.star.lang.DisposedException"):
 132                     print("caught DisposedException")
 133                     pass # ignore
 134             else:
 135                 self.soffice.terminate()
 136             ret = self.soffice.wait()
 137             self.xContext = None
 138             self.socket = None
 139             self.soffice = None
 140             if ret != 0:
 141                 raise Exception("Exit status indicates failure: " + str(ret))
 142 #            return ret
 143
 144 class PerTestConnection:
 145     def __init__(self, args):
 146         self.args = args
 147         self.connection = None
 148     def getContext(self):
 149         return self.connection.xContext
 150     def setUp(self):
 151         assert(not(self.connection))
 152     def preTest(self):
 153         conn = OfficeConnection(self.args)
 154         conn.setUp()
 155         self.connection = conn
 156     def postTest(self):
 157         if self.connection:
 158             try:
 159                 self.connection.tearDown()
 160             finally:
 161                 self.connection = None
 162     def tearDown(self):
 163         assert(not(self.connection))
 164
 165 class PersistentConnection:
 166     def __init__(self, args):
 167         self.args = args
 168         self.connection = None
 169     def getContext(self):
 170         return self.connection.xContext
 171     def setUp(self):
 172         conn = OfficeConnection(self.args)
 173         conn.setUp()
 174         self.connection = conn
 175     def preTest(self):
 176         assert(self.connection)
 177     def postTest(self):
 178         assert(self.connection)
 179     def tearDown(self):
 180         if self.connection:
 181             try:
 182                 self.connection.tearDown()
 183             finally:
 184                 self.connection = None
 185
 186 def simpleInvoke(connection, test):
 187     try:
 188         connection.preTest()
 189         test.run(connection.getContext())
 190     finally:
 191         connection.postTest()
 192
 193 def retryInvoke(connection, test):
 194     tries = 5
 195     while tries > 0:
 196         try:
 197             tries -= 1
 198             try:
 199                 connection.preTest()
 200                 test.run(connection.getContext())
 201                 return
 202             finally:
 203                 connection.postTest()
 204         except KeyboardInterrupt:
 205             raise # Ctrl+C should work
 206         except:
 207             print("retryInvoke: caught exception")
 208     raise Exception("FAILED retryInvoke")
 209
 210 def runConnectionTests(connection, invoker, tests):
 211     try:
 212         connection.setUp()
 213         for test in tests:
 214             invoker(connection, test)
 215     finally:
 216         connection.tearDown()
 217
 218 class EventListener(XDocumentEventListener,unohelper.Base):
 219     def __init__(self):
 220         self.layoutFinished = False
 221     def documentEventOccured(self, event):
 222 #        print(str(event.EventName))
 223         if event.EventName == "OnLayoutFinished":
 224             self.layoutFinished = True
 225     def disposing(event):
 226         pass
 227
 228 def mkPropertyValue(name, value):
 229     return uno.createUnoStruct("com.sun.star.beans.PropertyValue",
 230             name, 0, value, 0)
 231
 232 ### tests ###
 233
 234 def loadFromURL(xContext, url):
 235     xDesktop = xContext.ServiceManager.createInstanceWithContext(
 236             "com.sun.star.frame.Desktop", xContext)
 237     props = [("Hidden", True), ("ReadOnly", True)] # FilterName?
 238     loadProps = tuple([mkPropertyValue(name, value) for (name, value) in props])
 239     xListener = EventListener()
 240     xGEB = xContext.ServiceManager.createInstanceWithContext(
 241         "com.sun.star.frame.GlobalEventBroadcaster", xContext)
 242     xGEB.addDocumentEventListener(xListener)
 243     try:
 244         xDoc = xDesktop.loadComponentFromURL(url, "_blank", 0, loadProps)
 245         time_ = 0
 246         while time_ < 30:
 247             if xListener.layoutFinished:
 248                 return xDoc
 249             print("delaying...")
 250             time_ += 1
 251             time.sleep(1)
 252         print("timeout: no OnLayoutFinished received")
 253         return xDoc
 254     except:
 255         if xDoc:
 256             print("CLOSING")
 257             xDoc.close(True)
 258         raise
 259     finally:
 260         if xListener:
 261             xGEB.removeDocumentEventListener(xListener)
 262
 263 def printDoc(xContext, xDoc, url):
 264     props = [ mkPropertyValue("FileName", url) ]
 265 # xDoc.print(props)
 266     uno.invoke(xDoc, "print", (tuple(props),)) # damn, that's a keyword!
 267     busy = True
 268     while busy:
 269         print("printing...")
 270         time.sleep(1)
 271         prt = xDoc.getPrinter()
 272         for value in prt:
 273             if value.Name == "IsBusy":
 274                 busy = value.Value
 275     print("...done printing")
 276
 277 class LoadPrintFileTest:
 278     def __init__(self, file, prtsuffix):
 279         self.file = file
 280         self.prtsuffix = prtsuffix
 281     def run(self, xContext):
 282         print("Loading document: " + self.file)
 283         try:
 284             url = "file://" + quote(self.file)
 285             xDoc = loadFromURL(xContext, url)
 286             printDoc(xContext, xDoc, url + self.prtsuffix)
 287         finally:
 288             if xDoc:
 289                 xDoc.close(True)
 290             print("...done with: " + self.file)
 291
 292 def runLoadPrintFileTests(opts, dirs, suffix, reference):
 293     if reference:
 294         prtsuffix = ".pdf.reference"
 295     else:
 296         prtsuffix = ".pdf"
 297     files = getFiles(dirs, suffix)
 298     tests = (LoadPrintFileTest(file, prtsuffix) for file in files)
 299     connection = PersistentConnection(opts)
 300 #    connection = PerTestConnection(opts)
 301     runConnectionTests(connection, simpleInvoke, tests)
 302
 303 def mkImages(file, resolution):
 304     argv = [ "gs", "-r" + resolution, "-sOutputFile=" + file + ".%04d.jpeg",
 305              "-dNOPROMPT", "-dNOPAUSE", "-dBATCH", "-sDEVICE=jpeg", file ]
 306     ret = subprocess.check_call(argv)
 307
 308 def mkAllImages(dirs, suffix, resolution, reference):
 309     if reference:
 310         prtsuffix = ".pdf.reference"
 311     else:
 312         prtsuffix = ".pdf"
 313     for dir in dirs:
 314         files = filelist(dir, suffix)
 315         print(files)
 316         for f in files:
 317             mkImages(f + prtsuffix, resolution)
 318
 319 def identify(imagefile):
 320     argv = ["identify", "-format", "%k", imagefile]
 321     process = subprocess.Popen(argv, stdout=subprocess.PIPE)
 322     result, _ = process.communicate()
 323     if process.wait() != 0:
 324         raise Exception("identify failed")
 325     if result.partition(b"\n")[0] != b"1":
 326         print("identify result: " + result)
 327         print("DIFFERENCE in " + imagefile)
 328
 329 def compose(refimagefile, imagefile, diffimagefile):
 330     argv = [ "composite", "-compose", "difference",
 331             refimagefile, imagefile, diffimagefile ]
 332     subprocess.check_call(argv)
 333
 334 def compareImages(file):
 335     allimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
 336                    if f.startswith(file)]
 337 #    refimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
 338 #                   if f.startswith(file + ".reference")]
 339 #    print("compareImages: allimages:" + str(allimages))
 340     (refimages, images) = partition(sorted(allimages),
 341             lambda f: f.startswith(file + ".pdf.reference"))
 342 #    print("compareImages: images" + str(images))
 343     for (image, refimage) in zip(images, refimages):
 344         compose(image, refimage, image + ".diff")
 345         identify(image + ".diff")
 346     if (len(images) != len(refimages)):
 347         print("DIFFERENT NUMBER OF IMAGES FOR: " + file)
 348
 349 def compareAllImages(dirs, suffix):
 350     print("compareAllImages...")
 351     for dir in dirs:
 352         files = filelist(dir, suffix)
 353 #        print("compareAllImages:" + str(files))
 354         for f in files:
 355             compareImages(f)
 356     print("...compareAllImages done")
 357
 358
 359 def parseArgs(argv):
 360     (optlist,args) = getopt.getopt(argv[1:], "hr",
 361             ["help", "soffice=", "userdir=", "reference", "valgrind"])
 362 #    print optlist
 363     return (dict(optlist), args)
 364
 365 def usage():
 366     message = """usage: {program} [option]... [directory]..."
 367  -h | --help:      print usage information
 368  -r | --reference: generate new reference files (otherwise: compare)
 369  --soffice=method:location
 370                    specify soffice instance to connect to
 371                    supported methods: 'path', 'connect'
 372  --userdir=URL     specify user installation directory for 'path' method
 373  --valgrind        pass --valgrind to soffice for 'path' method"""
 374     print(message.format(program = os.path.basename(sys.argv[0])))
 375
 376 def checkTools():
 377     try:
 378         subprocess.check_output(["gs", "--version"])
 379     except:
 380         print("Cannot execute 'gs'. Please install ghostscript.")
 381         sys.exit(1)
 382     try:
 383         subprocess.check_output(["composite", "-version"])
 384         subprocess.check_output(["identify", "-version"])
 385     except:
 386         print("Cannot execute 'composite' or 'identify'.")
 387         print("Please install ImageMagick.")
 388         sys.exit(1)
 389
 390 if __name__ == "__main__":
 391 #    checkTools()
 392     (opts,args) = parseArgs(sys.argv)
 393     if len(args) == 0:
 394         usage()
 395         sys.exit(1)
 396     if "-h" in opts or "--help" in opts:
 397         usage()
 398         sys.exit()
 399     elif "--soffice" in opts:
 400         reference = "-r" in opts or "--reference" in opts
 401         runLoadPrintFileTests(opts, args, ".odt", reference)
 402         mkAllImages(args, ".odt", "200", reference)
 403         if not(reference):
 404             compareAllImages(args, ".odt")
 405     else:
 406         usage()
 407         sys.exit(1)
 408
 409 # vim:set shiftwidth=4 softtabstop=4 expandtab: