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"""Import twins: the same source saved as ODF 1.2 and as ODF 1.3 by another producer (plain and extended) must import alike. 21The corpus is a directory with v12, v13, v12ext and v13ext subdirectories holding the same file names. Each document is 22loaded once, dumped (aootest_worker.ops.dump) and the dumps of every pair are compared.""" 23import json 24import re 25import zipfile 26from pathlib import Path 27 28from aootest.checkers import diff7 29from aootest.client import WorkerError, WorkerFailure, WorkerHang 30 31PAIRS = (("v12", "v13"), ("v12ext", "v13ext")) 32MODES = sorted({m for p in PAIRS for m in p}) 33SUFFIXES = (".odt", ".ods", ".odp", ".odg") 34 35 36def corpus_problem(root): 37 """Why `root` is not a usable twin corpus, or None.""" 38 for m in MODES: 39 if not (Path(root) / m).is_dir(): 40 return "no %s directory in %s" % (m, root) 41 names = [set(d.name for d in (Path(root) / m).iterdir() if d.suffix in SUFFIXES) for m in MODES] 42 if not names[0]: 43 return "no documents in %s" % (Path(root) / MODES[0]) 44 if any(n != names[0] for n in names): 45 return "the four directories do not hold the same file names" 46 return None 47 48 49def document_names(root, mode="v12"): 50 return sorted(p.name for p in (Path(root) / mode).iterdir() if p.suffix in SUFFIXES) 51 52 53def mutate(src, dst): 54 """Copy of an ODF text package with one word appended to the first paragraph that has text; False if there is none.""" 55 done = False 56 with zipfile.ZipFile(src) as zi, zipfile.ZipFile(dst, "w") as zo: 57 for it in zi.infolist(): 58 data = zi.read(it.filename) 59 if it.filename == "content.xml" and not done: 60 text = data.decode("utf-8") 61 text2, n = re.subn(r">([A-Za-z][^<>]{2,})</text:p>", r">\1 mutated</text:p>", text, count=1) 62 if n: 63 data, done = text2.encode("utf-8"), True 64 zo.writestr(it, data, compress_type=it.compress_type) 65 return done 66 67 68def dump_one(session, path, timeout=300): 69 """The dump of one document, or {"error": text} when it did not load, hung or killed the office.""" 70 try: 71 return session.call("dump_document", {"path": str(path)}, timeout=timeout) 72 except WorkerError as e: 73 return {"error": "%s: %s" % (e.kind, e.message)} 74 except WorkerHang as e: 75 return {"error": "HANG: %s" % e} 76 except WorkerFailure as e: 77 return {"error": "CRASH: %s" % e} 78 79 80class TwinRun: 81 def __init__(self, session, root, workdir): 82 self.session, self.root, self.dir = session, Path(root), Path(workdir) 83 self._dumps, self._diffs, self._controls = {}, {}, None 84 85 def dumps(self, mode, source_dir=None, tag=None): 86 """Directory of <document>.json dumps for the documents of `mode` (or of `source_dir`), written on first use.""" 87 tag = tag or mode 88 if tag not in self._dumps: 89 out = self.dir / "dump" / tag 90 out.mkdir(parents=True, exist_ok=True) 91 src = Path(source_dir) if source_dir else self.root / mode 92 for p in sorted(src.iterdir()): 93 if p.suffix in SUFFIXES: 94 (out / (p.name + ".json")).write_text(json.dumps(dump_one(self.session, p), sort_keys=True, indent=0, default=str)) 95 self._dumps[tag] = out 96 return self._dumps[tag] 97 98 def load(self, mode, name): 99 return json.loads((self.dumps(mode) / (name + ".json")).read_text()) 100 101 def diff(self, pair): 102 """{document: {"ndiff", "diffs", ...}} for the dumps of a pair of modes.""" 103 if pair not in self._diffs: 104 self._diffs[pair] = diff7.diff_dirs(str(self.dumps(pair[0])), str(self.dumps(pair[1]))) 105 return self._diffs[pair] 106 107 def controls(self): 108 """Compares an unchanged copy and a copy with one changed word of a text document: ({"same", "mutated"} ndiff, base name), 109 or (None, reason). The comparison can only be trusted if the first is 0 and the second is not.""" 110 if self._controls is None: 111 base = None 112 for name in document_names(self.root): 113 d = self.load("v12", name) 114 if name.endswith(".odt") and d.get("type") == "writer" and (d.get("body") or {}).get("chars", 0) > 20: 115 base = self.root / "v12" / name 116 break 117 if base is None: 118 self._controls = (None, "no text document with characters in the corpus") 119 else: 120 old, new = self.dir / "ctl" / "old", self.dir / "ctl" / "new" 121 old.mkdir(parents=True), new.mkdir(parents=True) 122 for d in (old, new): 123 (d / "ctl-same.odt").write_bytes(base.read_bytes()) 124 (old / "ctl-mut.odt").write_bytes(base.read_bytes()) 125 if not mutate(base, new / "ctl-mut.odt"): 126 self._controls = (None, "could not change a word in %s" % base.name) 127 else: 128 res = diff7.diff_dirs(str(self.dumps(None, old, "ctl-old")), str(self.dumps(None, new, "ctl-new"))) 129 self._controls = ({"same": res["ctl-same.odt"]["ndiff"], "mutated": res["ctl-mut.odt"]["ndiff"]}, base.name) 130 return self._controls 131