1################################################################ 2# 3# Licensed to the Apache Software Foundation (ASF) under one 4# or more contributor license agreements. See the NOTICE file 5# distributed with this work for additional information 6# regarding copyright ownership. The ASF licenses this file 7# to you under the Apache License, Version 2.0 (the 8# "License"); you may not use this file except in compliance 9# with the License. You may obtain a copy of the License at 10# 11# http://www.apache.org/licenses/LICENSE-2.0 12# 13# Unless required by applicable law or agreed to in writing, 14# software distributed under the License is distributed on an 15# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 16# KIND, either express or implied. See the License for the 17# specific language governing permissions and limitations 18# under the License. 19# 20################################################################ 21 22"""The scenarios of test/smoketestdoc in the openoffice tree (new, save and reopen per app, chart, Math, Java loader, 23database, extension), as UNO steps. Each op returns {"steps": [{"name", "ok", "detail"}]} so the host can say which step failed.""" 24import os 25 26import uno 27import unohelper 28from com.sun.star.lang import XServiceInfo 29from com.sun.star.task import XInteractionHandler 30from com.sun.star.ucb import XCommandEnvironment 31from com.sun.star.util import XCloseListener 32 33from aootest_worker.ops import core 34from aootest_worker.ops.core import pv 35 36 37class Steps: 38 def __init__(self): 39 self.steps = [] 40 41 def check(self, name, fn): 42 """Runs fn; the step passes when it returns anything but False (or a (False, detail) pair) and raises nothing. Returns fn's value, or None if it failed.""" 43 try: 44 value = fn() 45 except Exception as e: 46 self.steps.append({"name": name, "ok": False, "detail": "%s: %s" % (type(e).__name__, e)}) 47 return None 48 detail = "" 49 if isinstance(value, tuple) and len(value) == 2 and isinstance(value[0], bool): 50 value, detail = value 51 ok = value is not False 52 self.steps.append({"name": name, "ok": ok, "detail": detail if detail or ok else "returned False"}) 53 return value if ok else None 54 55 def result(self): 56 return {"steps": self.steps} 57 58 59class CloseListener(unohelper.Base, XCloseListener): 60 def __init__(self): 61 self.closed = False 62 63 def queryClosing(self, source, gets_ownership): 64 pass 65 66 def notifyClosing(self, source): 67 self.closed = True 68 69 def disposing(self, source): 70 pass 71 72 73def _url(path): 74 return uno.systemPathToFileUrl(os.path.abspath(path)) 75 76 77def _new(factory): 78 doc = core._desktop().loadComponentFromURL(factory, "_blank", 0, (pv("Hidden", True),)) 79 if doc is None: 80 raise RuntimeError("could not create %s" % factory) 81 return doc 82 83 84def _close_with_listener(doc): 85 listener = CloseListener() 86 doc.addCloseListener(listener) 87 doc.close(True) 88 return listener.closed 89 90 91def _filter_exists(name): 92 return core._create("com.sun.star.document.FilterFactory").hasByName(name) 93 94 95def doc_cycle(factory, formats, workdir, base="smoketest_file"): 96 """New document from `factory`, stored once per (label, filter, extension) in `formats`, closed (the close listener must 97 be called), then every stored file loaded again and closed.""" 98 st = Steps() 99 doc = st.check("new", lambda: _new(factory)) 100 if doc is None: 101 return st.result() 102 paths = {} 103 for label, flt, ext in formats: 104 path = os.path.join(workdir, "%s.%s" % (base, ext)) 105 paths[label] = path 106 st.check("filter %s exists" % label, lambda: _filter_exists(flt)) 107 st.check("save %s" % label, lambda: doc.storeAsURL(_url(path), (pv("FilterName", flt),)) or True) 108 st.check("close", lambda: _close_with_listener(doc)) 109 for label, flt, ext in formats: 110 def reopen(): 111 d = core._desktop().loadComponentFromURL(_url(paths[label]), "_blank", 0, (pv("Hidden", True), pv("MacroExecutionMode", 0))) 112 if d is None: 113 return False 114 d.close(True) 115 st.check("open %s" % label, reopen) 116 return st.result() 117 118 119def chart(): 120 """A chart object added to the first sheet of a new spreadsheet.""" 121 st = Steps() 122 doc = st.check("new", lambda: _new("private:factory/scalc")) 123 if doc is None: 124 return st.result() 125 126 def add(): 127 charts = doc.Sheets.getByIndex(0).Charts 128 charts.addNewByName("TestChart", uno.createUnoStruct("com.sun.star.awt.Rectangle"), 129 (uno.createUnoStruct("com.sun.star.table.CellRangeAddress"),), True, True) 130 return charts.hasByName("TestChart") 131 st.check("add chart", add) 132 st.check("close", lambda: _close_with_listener(doc)) 133 return st.result() 134 135 136def math(): 137 st = Steps() 138 doc = st.check("new", lambda: _new("private:factory/smath")) 139 if doc is not None: 140 st.check("close", lambda: _close_with_listener(doc)) 141 return st.result() 142 143 144def java_loader(): 145 """The Java loader service can be created. It does not start a JVM, so this holds without a JRE.""" 146 st = Steps() 147 st.check("create com.sun.star.loader.Java", lambda: core._create("com.sun.star.loader.Java") is not None) 148 return st.result() 149 150 151def _scan(conn, table, text): 152 """(number of rows, number of rows whose 5th column is `text`) by reading the whole table through a fresh RowSet.""" 153 rs = core._create("com.sun.star.sdb.RowSet") 154 rs.ActiveConnection = conn 155 rs.CommandType = uno.getConstantByName("com.sun.star.sdb.CommandType.COMMAND") 156 rs.Command = "SELECT * FROM " + table 157 rs.execute() 158 rows = found = 0 159 while rs.next(): 160 rows += 1 161 found += rs.getString(5) == text 162 rs.dispose() 163 return rows, found 164 165 166def database(text="Automatical Test"): 167 """Open the first registered data source, insert a row through a RowSet, read it back, delete it, read another row. 168 Row counts come from a full scan: RowCount of this driver counts what has been fetched so far.""" 169 st = Steps() 170 ctx = st.check("create DatabaseContext", lambda: core._create("com.sun.star.sdb.DatabaseContext")) 171 if ctx is None: 172 return st.result() 173 names = ctx.getElementNames() 174 if not st.check("a data source is registered", lambda: len(names) > 0): 175 return st.result() 176 source = ctx.getByName(names[0]) 177 table = st.check("a table exists", lambda: source.Tables.getElementNames()[0]) 178 conn = st.check("open connection", lambda: source.getConnection("", "")) 179 if table is None or conn is None: 180 return st.result() 181 try: 182 rs = st.check("create RowSet", lambda: core._create("com.sun.star.sdb.RowSet")) 183 if rs is None: 184 return st.result() 185 rs.ActiveConnection = conn 186 rs.CommandType = uno.getConstantByName("com.sun.star.sdb.CommandType.COMMAND") 187 rs.Command = "SELECT * FROM " + table 188 if st.check("run SELECT * FROM " + table, lambda: rs.execute() or True) is None: 189 return st.result() 190 before = _scan(conn, table, text) 191 st.check("table has no leftover test row", lambda: (before[1] == 0, "rows %d, leftovers %d" % before)) 192 193 def insert(): 194 rs.moveToInsertRow() 195 rs.updateString(5, text) 196 rs.insertRow() 197 rs.moveToCurrentRow() 198 after = _scan(conn, table, text) 199 return (rs.getString(5) == text and after == (before[0] + 1, before[1] + 1)), "rows %d, matches %d" % after 200 inserted = st.check("insert row", insert) 201 if inserted: 202 def delete(): 203 rs.deleteRow() 204 after = _scan(conn, table, text) 205 return after == before, "rows %d, matches %d (expected %d, %d)" % (after + before) 206 st.check("delete row", delete) 207 208 def seek(): 209 rs.first() 210 return rs.getString(5) != text 211 st.check("read another row", seek) 212 finally: 213 conn.dispose() 214 return st.result() 215 216 217class ApprovingHandler(unohelper.Base, XInteractionHandler): 218 def handle(self, request): 219 approve = uno.getTypeByName("com.sun.star.task.XInteractionApprove") 220 for c in request.getContinuations(): 221 a = c.queryInterface(approve) 222 if a is not None: 223 a.select() 224 return 225 226 227class CommandEnvironment(unohelper.Base, XCommandEnvironment): 228 def getInteractionHandler(self): 229 return ApprovingHandler() 230 231 def getProgressHandler(self): 232 return None 233 234 235def extension(path, identifier, service, implementation): 236 """Add the extension at `path` for the user, check that its service exists, remove it, check that the service is gone.""" 237 st = Steps() 238 ctx = core._ctx() 239 mgr = st.check("get ExtensionManager", lambda: ctx.getValueByName("/singletons/com.sun.star.deployment.ExtensionManager")) 240 if mgr is None: 241 return st.result() 242 env = CommandEnvironment() 243 name = os.path.basename(path) 244 if st.check("add extension", lambda: mgr.addExtension(_url(path), (), "user", None, env) is not None) is None: 245 return st.result() 246 obj = st.check("service created", lambda: core._create(service)) 247 st.check("implementation name", lambda: obj is not None and obj.getImplementationName() == implementation) 248 st.check("remove extension", lambda: mgr.removeExtension(identifier, name, "user", None, env) or True) 249 st.check("service gone", lambda: core._create(service) is None) 250 return st.result() 251 252 253OPS = {name: core.guarded(fn) for name, fn in {"doc_cycle": doc_cycle, "chart": chart, "math": math, "java_loader": java_loader, 254 "database": database, "extension": extension}.items()} 255