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",
  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.getValueByName(
 241         "/singletons/com.sun.star.frame.theGlobalEventBroadcaster")
 242     xGEB.addDocumentEventListener(xListener)
 243     xDoc = None
 244     try:
 245         xDoc = xDesktop.loadComponentFromURL(url, "_blank", 0, loadProps)
 246         if xDoc is None:
 247             raise Exception("No document loaded?")
 248         time_ = 0
 249         while time_ < 30:
 250             if xListener.layoutFinished:
 251                 return xDoc
 252             print("delaying...")
 253             time_ += 1
 254             time.sleep(1)
 255         print("timeout: no OnLayoutFinished received")
 256         return xDoc
 257     except:
 258         if xDoc:
 259             print("CLOSING")
 260             xDoc.close(True)
 261         raise
 262     finally:
 263         if xListener:
 264             xGEB.removeDocumentEventListener(xListener)
 265
 266 def printDoc(xContext, xDoc, url):
 267     props = [ mkPropertyValue("FileName", url) ]
 268 # xDoc.print(props)
 269     uno.invoke(xDoc, "print", (tuple(props),)) # damn, that's a keyword!
 270     busy = True
 271     while busy:
 272         print("printing...")
 273         time.sleep(1)
 274         prt = xDoc.getPrinter()
 275         for value in prt:
 276             if value.Name == "IsBusy":
 277                 busy = value.Value
 278     print("...done printing")
 279
 280 class LoadPrintFileTest:
 281     def __init__(self, file, prtsuffix):
 282         self.file = file
 283         self.prtsuffix = prtsuffix
 284     def run(self, xContext):
 285         print("Loading document: " + self.file)
 286         xDoc = None
 287         try:
 288             url = "file://" + quote(self.file)
 289             xDoc = loadFromURL(xContext, url)
 290             printDoc(xContext, xDoc, url + self.prtsuffix)
 291         finally:
 292             if xDoc:
 293                 xDoc.close(True)
 294             print("...done with: " + self.file)
 295
 296 def runLoadPrintFileTests(opts, dirs, suffix, reference):
 297     if reference:
 298         prtsuffix = ".pdf.reference"
 299     else:
 300         prtsuffix = ".pdf"
 301     files = getFiles(dirs, suffix)
 302     tests = (LoadPrintFileTest(file, prtsuffix) for file in files)
 303     connection = PersistentConnection(opts)
 304 #    connection = PerTestConnection(opts)
 305     runConnectionTests(connection, simpleInvoke, tests)
 306
 307 def mkImages(file, resolution):
 308     argv = [ "gs", "-r" + resolution, "-sOutputFile=" + file + ".%04d.jpeg",
 309              "-dNOPROMPT", "-dNOPAUSE", "-dBATCH", "-sDEVICE=jpeg", file ]
 310     ret = subprocess.check_call(argv)
 311
 312 def mkAllImages(dirs, suffix, resolution, reference):
 313     if reference:
 314         prtsuffix = ".pdf.reference"
 315     else:
 316         prtsuffix = ".pdf"
 317     for dir in dirs:
 318         files = filelist(dir, suffix)
 319         print(files)
 320         for f in files:
 321             mkImages(f + prtsuffix, resolution)
 322
 323 def identify(imagefile):
 324     argv = ["identify", "-format", "%k", imagefile]
 325     process = subprocess.Popen(argv, stdout=subprocess.PIPE)
 326     result, _ = process.communicate()
 327     if process.wait() != 0:
 328         raise Exception("identify failed")
 329     if result.partition(b"\n")[0] != b"1":
 330         print("identify result: " + result.decode('utf-8'))
 331         print("DIFFERENCE in " + imagefile)
 332
 333 def compose(refimagefile, imagefile, diffimagefile):
 334     argv = [ "composite", "-compose", "difference",
 335             refimagefile, imagefile, diffimagefile ]
 336     subprocess.check_call(argv)
 337
 338 def compareImages(file):
 339     allimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
 340                    if f.startswith(file)]
 341 #    refimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
 342 #                   if f.startswith(file + ".reference")]
 343 #    print("compareImages: allimages:" + str(allimages))
 344     (refimages, images) = partition(sorted(allimages),
 345             lambda f: f.startswith(file + ".pdf.reference"))
 346 #    print("compareImages: images" + str(images))
 347     for (image, refimage) in zip(images, refimages):
 348         compose(image, refimage, image + ".diff")
 349         identify(image + ".diff")
 350     if (len(images) != len(refimages)):
 351         print("DIFFERENT NUMBER OF IMAGES FOR: " + file)
 352
 353 def compareAllImages(dirs, suffix):
 354     print("compareAllImages...")
 355     for dir in dirs:
 356         files = filelist(dir, suffix)
 357 #        print("compareAllImages:" + str(files))
 358         for f in files:
 359             compareImages(f)
 360     print("...compareAllImages done")
 361
 362
 363 def parseArgs(argv):
 364     (optlist,args) = getopt.getopt(argv[1:], "hr",
 365             ["help", "soffice=", "userdir=", "reference", "valgrind"])
 366 #    print optlist
 367     return (dict(optlist), args)
 368
 369 def usage():
 370     message = """usage: {program} [option]... [directory]..."
 371  -h | --help:      print usage information
 372  -r | --reference: generate new reference files (otherwise: compare)
 373  --soffice=method:location
 374                    specify soffice instance to connect to
 375                    supported methods: 'path', 'connect'
 376  --userdir=URL     specify user installation directory for 'path' method
 377  --valgrind        pass --valgrind to soffice for 'path' method"""
 378     print(message.format(program = os.path.basename(sys.argv[0])))
 379
 380 def checkTools():
 381     try:
 382         subprocess.check_output(["gs", "--version"])
 383     except:
 384         print("Cannot execute 'gs'. Please install ghostscript.")
 385         sys.exit(1)
 386     try:
 387         subprocess.check_output(["composite", "-version"])
 388         subprocess.check_output(["identify", "-version"])
 389     except:
 390         print("Cannot execute 'composite' or 'identify'.")
 391         print("Please install ImageMagick.")
 392         sys.exit(1)
 393
 394 if __name__ == "__main__":
 395 #    checkTools()
 396     (opts,args) = parseArgs(sys.argv)
 397     if len(args) == 0:
 398         usage()
 399         sys.exit(1)
 400     if "-h" in opts or "--help" in opts:
 401         usage()
 402         sys.exit()
 403     elif "--soffice" in opts:
 404         reference = "-r" in opts or "--reference" in opts
 405         runLoadPrintFileTests(opts, args, ".odt", reference)
 406         mkAllImages(args, ".odt", "200", reference)
 407         if not(reference):
 408             compareAllImages(args, ".odt")
 409     else:
 410         usage()
 411         sys.exit(1)
 412
 413 # vim:set shiftwidth=4 softtabstop=4 expandtab: