424 lines
14 KiB
Python
Executable file
424 lines
14 KiB
Python
Executable file
# -*- tab-width: 4; indent-tabs-mode: nil; py-indent-offset: 4 -*-
|
|
# Version: MPL 1.1 / GPLv3+ / LGPLv3+
|
|
#
|
|
# The contents of this file are subject to the Mozilla Public License Version
|
|
# 1.1 (the "License"); you may not use this file except in compliance with
|
|
# the License or as specified alternatively below. You may obtain a copy of
|
|
# the License at http://www.mozilla.org/MPL/
|
|
#
|
|
# Software distributed under the License is distributed on an "AS IS" basis,
|
|
# WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
|
|
# for the specific language governing rights and limitations under the
|
|
# License.
|
|
#
|
|
# Major Contributor(s):
|
|
# Copyright (C) 2012 Red Hat, Inc., Michael Stahl <mstahl@redhat.com>
|
|
# (initial developer)
|
|
#
|
|
# All Rights Reserved.
|
|
#
|
|
# For minor contributions see the git repository.
|
|
#
|
|
# Alternatively, the contents of this file may be used under the terms of
|
|
# either the GNU General Public License Version 3 or later (the "GPLv3+"), or
|
|
# the GNU Lesser General Public License Version 3 or later (the "LGPLv3+"),
|
|
# in which case the provisions of the GPLv3+ or the LGPLv3+ are applicable
|
|
# instead of those above.
|
|
|
|
import getopt
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
try:
|
|
from urllib.parse import quote
|
|
except ImportError:
|
|
from urllib import quote
|
|
|
|
try:
|
|
import pyuno
|
|
import uno
|
|
import unohelper
|
|
except ImportError:
|
|
print("pyuno not found: try to set PYTHONPATH and URE_BOOTSTRAP variables")
|
|
print("PYTHONPATH=/installation/opt/program")
|
|
print("URE_BOOTSTRAP=file:///installation/opt/program/fundamentalrc")
|
|
raise
|
|
|
|
try:
|
|
from com.sun.star.document import XDocumentEventListener
|
|
except ImportError:
|
|
print("UNO API class not found: try to set URE_BOOTSTRAP variable")
|
|
print("URE_BOOTSTRAP=file:///installation/opt/program/fundamentalrc")
|
|
raise
|
|
|
|
### utilities ###
|
|
|
|
def partition(list, pred):
|
|
left = []
|
|
right = []
|
|
for e in list:
|
|
if pred(e):
|
|
left.append(e)
|
|
else:
|
|
right.append(e)
|
|
return (left, right)
|
|
|
|
def filelist(dir, suffix):
|
|
if len(dir) == 0:
|
|
raise Exception("filelist: empty directory")
|
|
if not(dir[-1] == "/"):
|
|
dir += "/"
|
|
files = [dir + f for f in os.listdir(dir)]
|
|
# print(files)
|
|
return [f for f in files
|
|
if os.path.isfile(f) and os.path.splitext(f)[1] == suffix]
|
|
|
|
def getFiles(dirs, suffix):
|
|
files = []
|
|
for dir in dirs:
|
|
files += filelist(dir, suffix)
|
|
return files
|
|
|
|
### UNO utilities ###
|
|
|
|
class OfficeConnection:
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.soffice = None
|
|
self.socket = None
|
|
self.xContext = None
|
|
def setUp(self):
|
|
(method, sep, rest) = self.args["--soffice"].partition(":")
|
|
if sep != ":":
|
|
raise Exception("soffice parameter does not specify method")
|
|
if method == "path":
|
|
socket = "pipe,name=pytest" + str(uuid.uuid1())
|
|
try:
|
|
userdir = self.args["--userdir"]
|
|
except KeyError:
|
|
raise Exception("'path' method requires --userdir")
|
|
if not(userdir.startswith("file://")):
|
|
raise Exception("--userdir must be file URL")
|
|
self.soffice = self.bootstrap(rest, userdir, socket)
|
|
elif method == "connect":
|
|
socket = rest
|
|
else:
|
|
raise Exception("unsupported connection method: " + method)
|
|
self.xContext = self.connect(socket)
|
|
|
|
def bootstrap(self, soffice, userdir, socket):
|
|
argv = [ soffice, "--accept=" + socket + ";urp",
|
|
"-env:UserInstallation=" + userdir,
|
|
"--quickstart=no", "--nofirststartwizard",
|
|
"--norestore", "--nologo", "--headless" ]
|
|
if "--valgrind" in self.args:
|
|
argv.append("--valgrind")
|
|
return subprocess.Popen(argv)
|
|
|
|
def connect(self, socket):
|
|
xLocalContext = uno.getComponentContext()
|
|
xUnoResolver = xLocalContext.ServiceManager.createInstanceWithContext(
|
|
"com.sun.star.bridge.UnoUrlResolver", xLocalContext)
|
|
url = "uno:" + socket + ";urp;StarOffice.ComponentContext"
|
|
print("OfficeConnection: connecting to: " + url)
|
|
while True:
|
|
try:
|
|
xContext = xUnoResolver.resolve(url)
|
|
return xContext
|
|
# except com.sun.star.connection.NoConnectException
|
|
except pyuno.getClass("com.sun.star.connection.NoConnectException"):
|
|
print("NoConnectException: sleeping...")
|
|
time.sleep(1)
|
|
|
|
def tearDown(self):
|
|
if self.soffice:
|
|
if self.xContext:
|
|
try:
|
|
print("tearDown: calling terminate()...")
|
|
xMgr = self.xContext.ServiceManager
|
|
xDesktop = xMgr.createInstanceWithContext(
|
|
"com.sun.star.frame.Desktop", self.xContext)
|
|
xDesktop.terminate()
|
|
print("...done")
|
|
# except com.sun.star.lang.DisposedException:
|
|
except pyuno.getClass("com.sun.star.beans.UnknownPropertyException"):
|
|
print("caught UnknownPropertyException")
|
|
pass # ignore, also means disposed
|
|
except pyuno.getClass("com.sun.star.lang.DisposedException"):
|
|
print("caught DisposedException")
|
|
pass # ignore
|
|
else:
|
|
self.soffice.terminate()
|
|
ret = self.soffice.wait()
|
|
self.xContext = None
|
|
self.socket = None
|
|
self.soffice = None
|
|
if ret != 0:
|
|
raise Exception("Exit status indicates failure: " + str(ret))
|
|
# return ret
|
|
|
|
class PerTestConnection:
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.connection = None
|
|
def getContext(self):
|
|
return self.connection.xContext
|
|
def setUp(self):
|
|
assert(not(self.connection))
|
|
def preTest(self):
|
|
conn = OfficeConnection(self.args)
|
|
conn.setUp()
|
|
self.connection = conn
|
|
def postTest(self):
|
|
if self.connection:
|
|
try:
|
|
self.connection.tearDown()
|
|
finally:
|
|
self.connection = None
|
|
def tearDown(self):
|
|
assert(not(self.connection))
|
|
|
|
class PersistentConnection:
|
|
def __init__(self, args):
|
|
self.args = args
|
|
self.connection = None
|
|
def getContext(self):
|
|
return self.connection.xContext
|
|
def setUp(self):
|
|
conn = OfficeConnection(self.args)
|
|
conn.setUp()
|
|
self.connection = conn
|
|
def preTest(self):
|
|
assert(self.connection)
|
|
def postTest(self):
|
|
assert(self.connection)
|
|
def tearDown(self):
|
|
if self.connection:
|
|
try:
|
|
self.connection.tearDown()
|
|
finally:
|
|
self.connection = None
|
|
|
|
def simpleInvoke(connection, test):
|
|
try:
|
|
connection.preTest()
|
|
test.run(connection.getContext())
|
|
finally:
|
|
connection.postTest()
|
|
|
|
def retryInvoke(connection, test):
|
|
tries = 5
|
|
while tries > 0:
|
|
try:
|
|
tries -= 1
|
|
try:
|
|
connection.preTest()
|
|
test.run(connection.getContext())
|
|
return
|
|
finally:
|
|
connection.postTest()
|
|
except KeyboardInterrupt:
|
|
raise # Ctrl+C should work
|
|
except:
|
|
print("retryInvoke: caught exception")
|
|
raise Exception("FAILED retryInvoke")
|
|
|
|
def runConnectionTests(connection, invoker, tests):
|
|
try:
|
|
connection.setUp()
|
|
for test in tests:
|
|
invoker(connection, test)
|
|
finally:
|
|
connection.tearDown()
|
|
|
|
class EventListener(XDocumentEventListener,unohelper.Base):
|
|
def __init__(self):
|
|
self.layoutFinished = False
|
|
def documentEventOccured(self, event):
|
|
# print(str(event.EventName))
|
|
if event.EventName == "OnLayoutFinished":
|
|
self.layoutFinished = True
|
|
def disposing(event):
|
|
pass
|
|
|
|
def mkPropertyValue(name, value):
|
|
return uno.createUnoStruct("com.sun.star.beans.PropertyValue",
|
|
name, 0, value, 0)
|
|
|
|
### tests ###
|
|
|
|
def loadFromURL(xContext, url):
|
|
xDesktop = xContext.ServiceManager.createInstanceWithContext(
|
|
"com.sun.star.frame.Desktop", xContext)
|
|
props = [("Hidden", True), ("ReadOnly", True)] # FilterName?
|
|
loadProps = tuple([mkPropertyValue(name, value) for (name, value) in props])
|
|
xListener = EventListener()
|
|
xGEB = xContext.ServiceManager.createInstanceWithContext(
|
|
"com.sun.star.frame.GlobalEventBroadcaster", xContext)
|
|
xGEB.addDocumentEventListener(xListener)
|
|
try:
|
|
xDoc = xDesktop.loadComponentFromURL(url, "_blank", 0, loadProps)
|
|
time_ = 0
|
|
while time_ < 30:
|
|
if xListener.layoutFinished:
|
|
return xDoc
|
|
print("delaying...")
|
|
time_ += 1
|
|
time.sleep(1)
|
|
print("timeout: no OnLayoutFinished received")
|
|
return xDoc
|
|
except:
|
|
if xDoc:
|
|
print("CLOSING")
|
|
xDoc.close(True)
|
|
raise
|
|
finally:
|
|
if xListener:
|
|
xGEB.removeDocumentEventListener(xListener)
|
|
|
|
def printDoc(xContext, xDoc, url):
|
|
props = [ mkPropertyValue("FileName", url) ]
|
|
# xDoc.print(props)
|
|
uno.invoke(xDoc, "print", (tuple(props),)) # damn, that's a keyword!
|
|
busy = True
|
|
while busy:
|
|
print("printing...")
|
|
time.sleep(1)
|
|
prt = xDoc.getPrinter()
|
|
for value in prt:
|
|
if value.Name == "IsBusy":
|
|
busy = value.Value
|
|
print("...done printing")
|
|
|
|
class LoadPrintFileTest:
|
|
def __init__(self, file, prtsuffix):
|
|
self.file = file
|
|
self.prtsuffix = prtsuffix
|
|
def run(self, xContext):
|
|
print("Loading document: " + self.file)
|
|
try:
|
|
url = "file://" + quote(self.file)
|
|
xDoc = loadFromURL(xContext, url)
|
|
printDoc(xContext, xDoc, url + self.prtsuffix)
|
|
finally:
|
|
if xDoc:
|
|
xDoc.close(True)
|
|
print("...done with: " + self.file)
|
|
|
|
def runLoadPrintFileTests(opts, dirs, suffix, reference):
|
|
if reference:
|
|
prtsuffix = ".pdf.reference"
|
|
else:
|
|
prtsuffix = ".pdf"
|
|
files = getFiles(dirs, suffix)
|
|
tests = (LoadPrintFileTest(file, prtsuffix) for file in files)
|
|
connection = PersistentConnection(opts)
|
|
# connection = PerTestConnection(opts)
|
|
runConnectionTests(connection, simpleInvoke, tests)
|
|
|
|
def mkImages(file, resolution):
|
|
argv = [ "gs", "-r" + resolution, "-sOutputFile=" + file + ".%04d.jpeg",
|
|
"-dNOPROMPT", "-dNOPAUSE", "-dBATCH", "-sDEVICE=jpeg", file ]
|
|
ret = subprocess.check_call(argv)
|
|
|
|
def mkAllImages(dirs, suffix, resolution, reference):
|
|
if reference:
|
|
prtsuffix = ".pdf.reference"
|
|
else:
|
|
prtsuffix = ".pdf"
|
|
for dir in dirs:
|
|
files = filelist(dir, suffix)
|
|
print(files)
|
|
for f in files:
|
|
mkImages(f + prtsuffix, resolution)
|
|
|
|
def identify(imagefile):
|
|
argv = ["identify", "-format", "%k", imagefile]
|
|
result = subprocess.check_output(argv)
|
|
if result.partition("\n")[0] != "1":
|
|
print("identify result: " + result)
|
|
print("DIFFERENCE in " + imagefile)
|
|
|
|
def compose(refimagefile, imagefile, diffimagefile):
|
|
argv = [ "composite", "-compose", "difference",
|
|
refimagefile, imagefile, diffimagefile ]
|
|
subprocess.check_call(argv)
|
|
|
|
def compareImages(file):
|
|
allimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
|
|
if f.startswith(file)]
|
|
# refimages = [f for f in filelist(os.path.dirname(file), ".jpeg")
|
|
# if f.startswith(file + ".reference")]
|
|
# print("compareImages: allimages:" + str(allimages))
|
|
(refimages, images) = partition(sorted(allimages),
|
|
lambda f: f.startswith(file + ".pdf.reference"))
|
|
# print("compareImages: images" + str(images))
|
|
for (image, refimage) in zip(images, refimages):
|
|
compose(image, refimage, image + ".diff")
|
|
identify(image + ".diff")
|
|
if (len(images) != len(refimages)):
|
|
print("DIFFERENT NUMBER OF IMAGES FOR: " + file)
|
|
|
|
def compareAllImages(dirs, suffix):
|
|
print "compareAllImages..."
|
|
for dir in dirs:
|
|
files = filelist(dir, suffix)
|
|
# print("compareAllImages:" + str(files))
|
|
for f in files:
|
|
compareImages(f)
|
|
print "...compareAllImages done"
|
|
|
|
|
|
def parseArgs(argv):
|
|
(optlist,args) = getopt.getopt(argv[1:], "hr",
|
|
["help", "soffice=", "userdir=", "reference", "valgrind"])
|
|
# print optlist
|
|
return (dict(optlist), args)
|
|
|
|
def usage():
|
|
message = """usage: {program} [option]... [directory]..."
|
|
-h | --help: print usage information
|
|
-r | --reference: generate new reference files (otherwise: compare)
|
|
--soffice=method:location
|
|
specify soffice instance to connect to
|
|
supported methods: 'path', 'connect'
|
|
--userdir=URL specify user installation directory for 'path' method
|
|
--valgrind pass --valgrind to soffice for 'path' method"""
|
|
print(message.format(program = os.path.basename(sys.argv[0])))
|
|
|
|
def checkTools():
|
|
try:
|
|
subprocess.check_output(["gs", "--version"])
|
|
except:
|
|
print("Cannot execute 'gs'. Please install ghostscript.")
|
|
sys.exit(1)
|
|
try:
|
|
subprocess.check_output(["composite", "-version"])
|
|
subprocess.check_output(["identify", "-version"])
|
|
except:
|
|
print("Cannot execute 'composite' or 'identify'.")
|
|
print("Please install ImageMagick.")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
# checkTools()
|
|
(opts,args) = parseArgs(sys.argv)
|
|
if len(args) == 0:
|
|
usage()
|
|
sys.exit(1)
|
|
if "-h" in opts or "--help" in opts:
|
|
usage()
|
|
sys.exit()
|
|
elif "--soffice" in opts:
|
|
reference = "-r" in opts or "--reference" in opts
|
|
runLoadPrintFileTests(opts, args, ".odt", reference)
|
|
mkAllImages(args, ".odt", "200", reference)
|
|
if not(reference):
|
|
compareAllImages(args, ".odt")
|
|
else:
|
|
usage()
|
|
sys.exit(1)
|
|
|
|
# vim:set shiftwidth=4 softtabstop=4 expandtab:
|