fuzzyworld

fuzzyworld-tools / files / tests / test_issues.py

test_issues.py

57.4 KB · history · blame

1"""Tests for server/issues.py (bug reports by email become issue pages, run records, replay links).
2
3Run with: python3 -m unittest discover tests
4"""
5
6import contextlib
7import importlib.util
8import io
9import itertools
10import json
11import mailbox
12import os
13import types
14import re
15import shutil
16import subprocess
17import tempfile
18import unittest
19from unittest import mock
20
21ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
22FIXTURES = os.path.join(ROOT, "tests", "fixtures", "issues")
23spec = importlib.util.spec_from_file_location("issues", os.path.join(ROOT, "server", "issues.py"))
24issues = importlib.util.module_from_spec(spec)
25spec.loader.exec_module(issues)
26
27AUTH = "mx; dkim=pass header.d=example.org; dmarc=pass header.from=example.org"
28_counter = itertools.count(1)
29
30
31def mail(subject="Bug: crash on start", body="It crashed.\n", repo="demo", sender="Rita <reporter@example.org>",
32 auth=(AUTH,), extra_headers=(), mid=None, to=None):
33 mid = mid or f"<{next(_counter)}@test>"
34 lines = [f"Message-ID: {mid}", *extra_headers]
35 lines += [f"Authentication-Results: {a}" for a in auth]
36 lines += [f"From: {sender}", f"To: {to or f'bugs-k7q4+{repo}@fuzzyworld.net'}", f"Subject: {subject}",
37 "Content-Type: text/plain; charset=utf-8", "", body]
38 return "\n".join(lines).encode()
39
40
41def record(**over):
42 r = {"simVersion": 43, "seed": 7, "startLevel": 1, "options": {"dev": False}, "actions": [{"id": "wait"}]}
43 r.update(over)
44 return json.dumps(r, separators=(",", ":"))
45
46
47def rd(path):
48 with open(path) as f:
49 return f.read()
50
51
52def read_tree(root):
53 for dirpath, _, names in os.walk(root):
54 for n in names:
55 p = os.path.join(dirpath, n)
56 with open(p, "rb") as f:
57 yield p, f.read().decode("utf-8", "replace")
58
59
60class Base(unittest.TestCase):
61 def setUp(self):
62 self.tmp = tempfile.mkdtemp()
63 self.addCleanup(shutil.rmtree, self.tmp)
64 j = lambda *p: os.path.join(self.tmp, *p)
65 self.git, self.site, self.apps, self.fwstate = j("git"), j("site"), j("apps"), j("fwstate")
66 self.state = j("issues")
67 self.maildir = j("Maildir", ".bugs")
68 with open(j("sites.json"), "w") as f:
69 json.dump({"authserv_id": "mx"}, f)
70 env = {"FW_ISSUES_DIR": self.state, "FW_COMMENTS_SITES": j("sites.json"), "FW_ISSUES_MAILDIR": self.maildir}
71 old = {k: os.environ.get(k) for k in env}
72 os.environ.update(env)
73 self.addCleanup(lambda: [os.environ.pop(k) if v is None else os.environ.__setitem__(k, v) for k, v in old.items()])
74 for name, value in (("GIT_ROOT", self.git), ("SITE", self.site), ("APPS", self.apps), ("STATE", self.fwstate)):
75 p = mock.patch.object(issues.gen, name, value)
76 p.start()
77 self.addCleanup(p.stop)
78 self.summaries = []
79 p = mock.patch.object(issues, "refresh_summary", lambda repo: self.summaries.append(repo))
80 p.start()
81 self.addCleanup(p.stop)
82 self.repo("demo", "public", "public")
83
84 def repo(self, name, visibility="public", mode=None):
85 path = os.path.join(self.git, name + ".git")
86 subprocess.run(["git", "init", "--bare", "--quiet", path], check=True)
87 subprocess.run(["git", "-C", path, "config", "fuzzyworld.visibility", visibility], check=True)
88 if mode is not None:
89 subprocess.run(["git", "-C", path, "config", "fuzzyworld.issues", mode], check=True)
90 return path
91
92 def ingest(self, *raws):
93 os.makedirs(os.path.dirname(self.maildir), exist_ok=True)
94 box = mailbox.Maildir(self.maildir, create=True)
95 for raw in raws:
96 box.add(raw)
97 out = io.StringIO()
98 with contextlib.redirect_stdout(out):
99 issues.cmd_ingest()
100 return out.getvalue()
101
102 def cli(self, *argv):
103 out, err = io.StringIO(), io.StringIO()
104 with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
105 code = issues.main(list(argv))
106 return code, out.getvalue(), err.getvalue()
107
108 def page(self, repo="demo", *parts):
109 return os.path.join(self.site, repo, "issues", *parts)
110
111 def count(self, outcome):
112 path = self.maildir + "." + outcome
113 return len(mailbox.Maildir(path, create=False).keys()) if os.path.isdir(path) else 0
114
115 def rejected(self):
116 try:
117 return rd(os.path.join(self.state, "rejected.log"))
118 except FileNotFoundError:
119 return ""
120
121
122class Modes(Base):
123 def test_public_publishes_at_once(self):
124 self.ingest(mail())
125 self.assertTrue(os.path.isfile(self.page("demo", "index.html")))
126 text = rd(self.page("demo", "1", "index.html"))
127 self.assertIn("crash on start", text)
128 self.assertIn("It crashed.", text)
129 self.assertIn('<pre class="issue-text">', text)
130 self.assertEqual(self.summaries, ["demo"]) # the index appeared: the summary link is refreshed
131
132 def test_reviewed_waits_for_approve_and_hide_removes(self):
133 self.repo("rev", "public", "reviewed")
134 self.ingest(mail(repo="rev"))
135 self.assertFalse(os.path.exists(self.page("rev")))
136 self.assertEqual(self.cli("approve", "rev", "1")[0], 0)
137 self.assertTrue(os.path.isfile(self.page("rev", "1", "index.html")))
138 self.assertEqual(self.cli("hide", "rev", "1")[0], 0)
139 self.assertFalse(os.path.exists(self.page("rev")))
140
141 def test_hide_beats_public_mode(self):
142 self.ingest(mail())
143 self.cli("hide", "demo", "1")
144 self.assertFalse(os.path.exists(self.page("demo")))
145 self.ingest(mail(subject="Another"))
146 self.assertFalse(os.path.exists(self.page("demo", "1")))
147 self.assertTrue(os.path.isfile(self.page("demo", "2", "index.html")))
148 self.assertNotIn("crash on start", rd(self.page("demo", "index.html")))
149
150 def test_private_and_unknown_modes_publish_nothing(self):
151 for name, mode in (("p1", "private"), ("p2", "pubic"), ("p3", ""), ("p4", None)):
152 self.repo(name, "public", mode)
153 self.ingest(mail(repo=name))
154 self.assertFalse(os.path.exists(self.page(name)), name)
155 self.assertEqual(issues.issue_mode(os.path.join(self.git, name + ".git")), "private")
156 self.assertTrue(os.path.isfile(os.path.join(self.state, "repos", name, "1", "issue.json")))
157
158 def test_private_repo_publishes_nothing_even_in_public_mode(self):
159 self.repo("secret", "private", "public")
160 self.ingest(mail(repo="secret"))
161 self.cli("approve", "secret", "1")
162 self.assertFalse(os.path.exists(os.path.join(self.site, "secret")))
163
164 def test_close_and_reopen(self):
165 self.ingest(mail())
166 self.cli("close", "demo", "1")
167 self.assertIn("closed", rd(self.page("demo", "index.html")))
168 self.cli("reopen", "demo", "1")
169 self.assertNotIn("closed", rd(self.page("demo", "index.html")))
170
171 def test_unauthenticated_mail_is_held_in_public_mode(self):
172 self.ingest(mail(auth=("mx; dkim=fail",)))
173 self.assertFalse(os.path.exists(self.page()))
174 stored = json.load(open(os.path.join(self.state, "repos", "demo", "1", "issue.json")))
175 self.assertFalse(stored["messages"][0]["authenticated"])
176 self.cli("approve", "demo", "1")
177 self.assertTrue(os.path.isfile(self.page("demo", "1", "index.html")))
178
179 def test_foreign_authentication_header_counts_as_unauthenticated(self):
180 self.ingest(mail(auth=("evil; dkim=pass header.d=example.org",)))
181 self.assertFalse(os.path.exists(self.page()))
182
183 def test_unauthenticated_reply_is_not_shown_until_approved(self):
184 first = mail(mid="<a@test>")
185 self.ingest(first, mail(mid="<b@test>", body="Spoofed reply.\n", auth=(), extra_headers=("In-Reply-To: <a@test>",)))
186 self.assertNotIn("Spoofed", rd(self.page("demo", "1", "index.html")))
187 self.cli("approve", "demo", "1")
188 self.assertIn("Spoofed", rd(self.page("demo", "1", "index.html")))
189
190 def test_sender_limit_in_public_mode(self):
191 self.ingest(*[mail(subject=f"Bug {i}") for i in range(issues.SENDER_LIMIT + 2)])
192 self.assertEqual(len(issues.issue_numbers("demo")), issues.SENDER_LIMIT)
193 self.assertEqual(self.count("bad"), 2)
194
195
196class Intake(Base):
197 def test_reply_joins_the_issue(self):
198 self.ingest(mail(mid="<first@test>"),
199 mail(mid="<second@test>", subject="Re: Bug", body="More detail.\n",
200 extra_headers=("In-Reply-To: <first@test>", "References: <first@test>")))
201 self.assertEqual(issues.issue_numbers("demo"), [1])
202 stored = json.load(open(os.path.join(self.state, "repos", "demo", "1", "issue.json")))
203 self.assertEqual(len(stored["messages"]), 2)
204 text = rd(self.page("demo", "1", "index.html"))
205 self.assertIn("It crashed.", text)
206 self.assertIn("More detail.", text)
207
208 def test_reply_via_references_only_and_unknown_parent(self):
209 self.ingest(mail(mid="<first@test>"), mail(mid="<x@test>", extra_headers=("References: <old@test> <first@test>",)),
210 mail(mid="<y@test>", extra_headers=("In-Reply-To: <nobody@test>",)))
211 self.assertEqual(issues.issue_numbers("demo"), [1, 2])
212 self.assertEqual(len(issues.load_issue("demo", 1)["messages"]), 2)
213
214 def test_numbers_count_up_per_repo(self):
215 self.repo("other", "public", "public")
216 self.ingest(mail(), mail(), mail(repo="other"))
217 self.assertEqual(issues.issue_numbers("demo"), [1, 2])
218 self.assertEqual(issues.issue_numbers("other"), [1])
219
220 def test_unknown_repo_and_bad_addresses_go_to_bad(self):
221 self.ingest(mail(repo="nosuchrepo"), mail(repo="-evil"), mail(to="somebody@example.org"),
222 mail(to="bugs-k7q4+demo@fuzzyworld.net, bugs-k7q4+other@fuzzyworld.net"))
223 self.assertEqual(self.count("bad"), 4)
224 self.assertEqual(issues.stored_repos(), [])
225 self.assertIn("no such repo", self.rejected())
226
227 def test_address_in_cc_or_x_original_to(self):
228 self.ingest(mail(to="list@example.org", extra_headers=("X-Original-To: bugs-k7q4+demo@fuzzyworld.net",)))
229 self.assertEqual(issues.issue_numbers("demo"), [1])
230
231 def test_repo_with_dot_git_in_the_address_is_refused(self):
232 self.ingest(mail(repo="demo.git"))
233 self.assertEqual(self.count("bad"), 1)
234
235 def test_same_mail_twice_is_stored_once(self):
236 raw = mail()
237 self.ingest(raw, raw)
238 self.assertEqual(issues.issue_numbers("demo"), [1])
239 self.assertEqual(len(issues.load_issue("demo", 1)["messages"]), 1)
240 self.assertEqual(self.count("done"), 2)
241
242 def test_oversize_mail_goes_to_bad(self):
243 with mock.patch.object(issues, "MAX_MAIL", 2000):
244 self.ingest(mail(body="x" * 5000))
245 self.assertEqual(self.count("bad"), 1)
246 self.assertEqual(issues.stored_repos(), [])
247
248 def test_empty_text_and_two_froms_go_to_bad(self):
249 self.ingest(mail(body=" \n"), mail(extra_headers=("From: Other <o@example.org>",)))
250 self.assertEqual(self.count("bad"), 2)
251
252 def test_no_address_reaches_any_file(self):
253 body = "Mail me at secret.person@example.net or <other@example.com>.\nDevice: Pixel (sam@example.org)\n"
254 self.ingest(mail(subject="Bug from boss@example.net", body=body, sender="Rita Reporter <reporter@example.org>",
255 mid="<id-1@example.org>"))
256 found = 0
257 for root in (self.site, self.state):
258 for path, text in read_tree(root):
259 found += 1
260 for needle in ("@example", "Rita", "id-1"):
261 self.assertNotIn(needle, text, path)
262 self.assertGreater(found, 3)
263 page = rd(self.page("demo", "1", "index.html"))
264 self.assertIn("[address hidden]", page)
265 self.assertNotIn("Device:", page)
266 self.assertNotIn("sender_hash", page)
267 stored = rd(os.path.join(self.state, "repos", "demo", "1", "issue.json"))
268 self.assertIn('"author": "Reporter"', stored)
269
270 def test_script_is_escaped(self):
271 self.ingest(mail(subject="<script>alert(1)</script>", body="<script>alert(2)</script> & <b>\n"))
272 for p in (self.page("demo", "index.html"), self.page("demo", "1", "index.html")):
273 text = rd(p)
274 self.assertNotIn("<script>alert", text)
275 self.assertIn("&lt;script&gt;alert", text)
276 self.assertNotIn("<script", rd(self.page("demo", "1", "index.html")).replace("&lt;script", ""))
277
278 def test_sender_hash_uses_its_own_secret(self):
279 self.ingest(mail())
280 self.assertTrue(os.path.isfile(os.path.join(self.state, "secret")))
281 self.assertEqual(oct(os.stat(os.path.join(self.state, "secret")).st_mode & 0o777), "0o600")
282
283 def test_publish_takes_the_repo_lock_unless_told(self):
284 self.ingest(mail())
285 self.assertTrue(os.path.isfile(os.path.join(self.fwstate, "locks", "demo.lock")))
286 self.assertEqual(self.cli("publish", "demo", "--locked")[0], 0)
287
288 def test_list_show_and_usage(self):
289 self.ingest(mail())
290 code, out, _ = self.cli("list")
291 self.assertIn("demo#1", out)
292 self.assertIn("PUBLIC", out)
293 self.assertIn("crash on start", self.cli("show", "demo", "1")[1])
294 self.assertEqual(self.cli("show", "demo", "9")[0], 1)
295 self.assertEqual(self.cli("show", "../x", "1")[0], 1)
296 self.assertEqual(self.cli("nonsense")[0], 2)
297 self.assertEqual(self.cli("publish-all")[0], 0)
298
299
300class Records(Base):
301 def eml(self):
302 with open(os.path.join(FIXTURES, "explodorama-report.eml"), "rb") as f:
303 return f.read()
304
305 def fixture(self):
306 with open(os.path.join(FIXTURES, "q101-record.json")) as f:
307 return json.load(f)
308
309 def setUp(self):
310 super().setUp()
311 self.repo("explodorama", "public", "reviewed")
312
313 def stored_record(self, repo="explodorama", n=1):
314 with open(os.path.join(self.state, "repos", repo, str(n), "record.json")) as f:
315 return json.load(f)
316
317 def test_record_from_the_real_mail_matches_the_fixture(self):
318 self.ingest(self.eml())
319 rec = self.stored_record()
320 self.assertEqual(rec, self.fixture())
321 self.assertEqual(rec["seed"], 1464004408)
322 self.assertEqual(len(rec["actions"]), 290)
323 self.assertEqual(len(rec["actions"]), len(self.fixture()["actions"]))
324 raw = rd(os.path.join(self.state, "repos", "explodorama", "1", "record.json"))
325 self.assertEqual(raw, json.dumps(self.fixture(), separators=(",", ":"))) # re-serialized, compact
326
327 def test_find_record_accepts_the_json_on_the_next_line(self):
328 # seen in the first real IMAP mail: the mail client broke the line after "Run record: "
329 for text in ("x\nRun record: \n" + record() + " \n\n", "x\r\nRun record:\r\n" + record() + "\r\n"):
330 rec, why = issues.find_record(text)
331 self.assertEqual(rec["seed"], 7, why)
332 # but not across a blank line, and not when the next line is prose
333 self.assertEqual(issues.find_record("Run record:\n\n" + record() + "\n"), (None, ""))
334 self.assertEqual(issues.find_record("Run record:\nit crashed {sorry}\n"), (None, ""))
335
336 def test_record_json_on_the_next_line_is_not_shown_as_text(self):
337 text = issues.report_text("Hi\n\nSeed 1\nRun record: \n" + record() + " \n\nbye\n")
338 self.assertNotIn("simVersion", text)
339 self.assertIn("Seed 1", text)
340 self.assertIn("bye", text)
341
342 def test_find_record_skips_the_too_long_footer(self):
343 text = "x\nRun record: " + record() + "\n\n---\nRun record: too long for an email link. Paste it.\n"
344 rec, why = issues.find_record(text)
345 self.assertEqual(rec["seed"], 7)
346 rec, why = issues.find_record("Run record: too long for an email link.\n")
347 self.assertEqual((rec, why), (None, ""))
348 # an invalid line first, a valid one later: the valid one wins
349 rec, _ = issues.find_record("Run record: {bad}\nRun record: " + record(seed=9) + "\n")
350 self.assertEqual(rec["seed"], 9)
351
352 def test_page_leaves_out_device_and_record_lines(self):
353 self.ingest(self.eml())
354 stored = rd(os.path.join(self.state, "repos", "explodorama", "1", "issue.json"))
355 self.cli("approve", "explodorama", "1")
356 for text in (stored, rd(self.page("explodorama", "1", "index.html"))):
357 self.assertNotIn("Device:", text)
358 self.assertNotIn("Firefox", text)
359 self.assertNotIn("TestPhone", text)
360 self.assertNotIn("Run record", text)
361 self.assertIn("autopilot walked me into death", text) # the reporter's words come after the 9 KB record
362 self.assertIn("Bug report: version 0.2.0 (rules 43), level 4", rd(self.page("explodorama", "index.html")))
363
364 def test_record_json_appears_only_after_approve_in_reviewed_mode(self):
365 self.ingest(self.eml())
366 self.assertFalse(os.path.exists(self.page("explodorama")))
367 self.cli("approve", "explodorama", "1")
368 with open(self.page("explodorama", "1", "record.json")) as f:
369 self.assertEqual(json.load(f), self.fixture())
370 self.cli("hide", "explodorama", "1")
371 self.assertFalse(os.path.exists(self.page("explodorama", "1", "record.json")))
372
373 def test_record_is_published_at_once_in_public_mode(self):
374 self.repo("pub", "public", "public")
375 self.ingest(mail(repo="pub", body="It broke.\nRun record: " + record() + "\n"))
376 self.assertTrue(os.path.isfile(self.page("pub", "1", "record.json")))
377
378 def test_bad_records_are_refused_but_the_issue_is_kept(self):
379 too_big = record(note="x" * (issues.RECORD_MAX + 10))
380 cases = {
381 "over 64 KB": too_big,
382 "invalid JSON": '{"seed":1,"simVersion":1,"actions":[}',
383 "bool seed": record(seed=True),
384 "float seed": record(seed=1.5),
385 "missing actions": '{"seed":1,"simVersion":1}',
386 "actions not a list": record(actions={}),
387 "missing simVersion": '{"seed":1,"actions":[]}',
388 "NaN": '{"seed":1,"simVersion":1,"actions":NaN}',
389 "deep nesting": '{"seed":1,"simVersion":1,"actions":' + "[" * 30000 + "]" * 30000 + "}",
390 }
391 for label, line in cases.items():
392 rec, why = issues.find_record("Run record: " + line + "\n")
393 self.assertIsNone(rec, label)
394 self.assertTrue(why, label)
395 self.assertEqual(issues.check_record("[1]")[0], None)
396 self.repo("bad", "public", "public")
397 self.ingest(mail(repo="bad", body="Hello.\nRun record: " + record(seed=True) + "\n"))
398 self.assertEqual(issues.issue_numbers("bad"), [1])
399 self.assertFalse(os.path.exists(os.path.join(self.state, "repos", "bad", "1", "record.json")))
400 self.assertFalse(issues.load_issue("bad", 1)["record"])
401 self.assertIn("integer seed", self.rejected())
402
403 def test_quoted_printable_raw_fallback(self):
404 raw = b"Content-Transfer-Encoding: quoted-printable\n\nRun record: " + record().replace(":", "=3A").encode() + b"\n"
405 rec, why = issues.find_record("nothing useful here", raw)
406 self.assertEqual(rec["seed"], 7)
407 self.assertEqual(issues.find_record("nothing useful here", b"plain mail")[0], None)
408
409 def test_only_the_first_record_is_kept(self):
410 self.repo("pub", "public", "public")
411 self.ingest(mail(repo="pub", mid="<r1@test>", body="First.\nRun record: " + record(seed=1) + "\n"),
412 mail(repo="pub", mid="<r2@test>", body="Again.\nRun record: " + record(seed=2) + "\n",
413 extra_headers=("In-Reply-To: <r1@test>",)))
414 self.assertEqual(self.stored_record("pub")["seed"], 1)
415 self.assertIn("second run record ignored", self.rejected())
416
417 def test_record_from_an_unauthenticated_reply_is_ignored(self):
418 self.repo("pub", "public", "public")
419 self.ingest(mail(repo="pub", mid="<r1@test>"),
420 mail(repo="pub", mid="<r2@test>", auth=(), body="Hi.\nRun record: " + record() + "\n",
421 extra_headers=("In-Reply-To: <r1@test>",)))
422 self.assertFalse(issues.load_issue("pub", 1)["record"])
423
424
425class Review(Base):
426 """Fixes from the review: per-message approval, text and record privacy, setup checks."""
427
428 def test_nonfinite_numbers_are_refused(self):
429 for line in ('{"seed":1,"simVersion":1,"actions":[1e400]}', '{"seed":1,"simVersion":1,"actions":[-1e999]}',
430 '{"seed":1,"simVersion":1,"actions":[Infinity]}'):
431 self.assertIsNone(issues.find_record("Run record: " + line + "\n")[0], line)
432 rec, _ = issues.check_record('{"seed":1,"simVersion":1,"actions":[1.5e10]}')
433 self.assertEqual(json.dumps(rec, allow_nan=False), '{"simVersion": 1, "seed": 1, "actions": [15000000000.0]}')
434
435 def test_later_reply_after_approve_is_not_public_in_reviewed_mode(self):
436 self.repo("rev", "public", "reviewed")
437 self.ingest(mail(repo="rev", mid="<a@test>"))
438 self.cli("approve", "rev", "1")
439 self.ingest(mail(repo="rev", mid="<b@test>", body="Sneaky follow-up.\n", extra_headers=("In-Reply-To: <a@test>",)))
440 self.assertNotIn("Sneaky", rd(self.page("rev", "1", "index.html")))
441 self.assertIn("It crashed.", rd(self.page("rev", "1", "index.html")))
442 self.cli("approve", "rev", "1")
443 self.assertIn("Sneaky", rd(self.page("rev", "1", "index.html")))
444
445 def test_public_mode_shows_later_authenticated_reply_but_never_an_unauthenticated_one(self):
446 self.ingest(mail(mid="<a@test>"))
447 self.cli("approve", "demo", "1")
448 self.ingest(mail(mid="<b@test>", body="Authentic follow-up.\n", extra_headers=("In-Reply-To: <a@test>",)),
449 mail(mid="<c@test>", body="Forged follow-up.\n", auth=(), extra_headers=("In-Reply-To: <a@test>",)))
450 text = rd(self.page("demo", "1", "index.html"))
451 self.assertIn("Authentic follow-up.", text)
452 self.assertNotIn("Forged", text)
453
454 def test_hide_clears_message_approvals(self):
455 self.repo("rev", "public", "reviewed")
456 self.ingest(mail(repo="rev"))
457 self.cli("approve", "rev", "1")
458 self.cli("hide", "rev", "1")
459 self.assertFalse(issues.load_issue("rev", 1)["messages"][0]["approved"])
460
461 def test_sender_limit_applies_in_reviewed_mode(self):
462 self.repo("rev", "public", "reviewed")
463 self.ingest(*[mail(repo="rev", subject=f"Bug {i}") for i in range(issues.SENDER_LIMIT + 2)])
464 self.assertEqual(len(issues.issue_numbers("rev")), issues.SENDER_LIMIT)
465
466 def test_flowed_text_is_rejoined(self):
467 text = "Sorry, this was reported by boss@exa \nmple.net and it hurts. \nNew paragraph.\n"
468 out = issues.report_text(text, flowed=True)
469 self.assertIn("[address hidden]", out)
470 self.assertNotIn("exa", out)
471 self.assertIn("New paragraph.", out)
472 self.assertNotIn("boss@exa", issues.report_text(text, flowed=False).replace("[address hidden]", ""))
473
474 def test_device_continuation_line_is_dropped(self):
475 ua = "Device: Android 378x694; Mozilla/5.0 (Android 17; Mobile; rv:159.0) Gecko/"
476 out = issues.report_text(f"Hello.\n{ua}\n159.0 Firefox/159.0\nSeed 12, level 4\nBye.\n")
477 self.assertNotIn("Firefox", out)
478 self.assertNotIn("Android", out)
479 self.assertIn("Seed 12, level 4", out)
480 out = issues.report_text("Device: Pixel\n Firefox/1.0\nafter\n")
481 self.assertEqual(out, "after")
482
483 def test_quoted_replies_are_cut_at_the_attribution(self):
484 for cut in ("On Tue, 3 Nov 2026 at 10:00, Jane Roe <jane@example.net> wrote:",
485 "On Tue, 3 Nov 2026 at 10:00, Jane Roe\nwrote:",
486 "-----Original Message-----",
487 "From: Jane Roe\nSent: Tuesday\nTo: Someone"):
488 out = issues.report_text(f"My answer.\n\n{cut}\nJane Roe says secrets.\n")
489 self.assertEqual(out, "My answer.", cut)
490
491 def test_fullwidth_at_sign_is_scrubbed(self):
492 out = issues.report_text("write to boss\uff20example.net now")
493 self.assertNotIn("boss", out)
494 self.assertIn("[address hidden]", out)
495 self.ingest(mail(subject="Mail me\uff1a boss\uff20example.net", body="boss\uff20example.net\n"))
496 for p in (self.page("demo", "index.html"), self.page("demo", "1", "index.html")):
497 self.assertNotIn("boss", rd(p))
498
499 def test_record_keeps_only_the_allowlist_and_refuses_addresses(self):
500 rec, _ = issues.check_record(record(deviceId="abc", notes="x", startLevel=2, startCharge=None))
501 self.assertEqual(sorted(rec), sorted(["simVersion", "seed", "actions", "startLevel", "startCharge", "options"]))
502 for bad in (record(options={"name": "me@example.net"}), record(actions=[{"id": "say", "text": "a\uff20b.org"}]),
503 record(options={"x@example.org": 1})):
504 self.assertIsNone(issues.check_record(bad)[0], bad)
505 with open(os.path.join(FIXTURES, "q101-record.json")) as f:
506 real = json.load(f)
507 self.assertEqual(issues.check_record(json.dumps(real))[0], real) # nothing of the real record is lost
508 self.ingest(mail(body="Hi.\nRun record: " + record(secretKey="zzz") + "\n"))
509 self.assertNotIn("secretKey", rd(os.path.join(self.state, "repos", "demo", "1", "record.json")))
510 self.assertNotIn("secretKey", rd(self.page("demo", "1", "record.json")))
511
512 def imap_config(self, path, **over):
513 cfg = {"host": "imap.example.org", "user": "u", "password": "pw", "folder": "INBOX", **over}
514 os.makedirs(os.path.dirname(path), exist_ok=True)
515 with open(path, "w") as f:
516 json.dump(cfg, f)
517 os.chmod(path, 0o600)
518
519 def test_same_mailbox_as_comments_is_refused(self):
520 theirs = os.path.join(self.tmp, "comments", "imap.json")
521 with mock.patch.dict(os.environ, {"FW_COMMENTS_IMAP": theirs}):
522 self.imap_config(theirs)
523 self.imap_config(issues.imap_path())
524 with self.assertRaisesRegex(issues.Error, "own account or its own"):
525 issues.cmd_ingest()
526 self.assertEqual(self.cli("imap-check")[0], 1)
527 self.imap_config(issues.imap_path(), folder="INBOX.bugs")
528 issues.check_mailbox_separate() # a different folder is fine
529 self.imap_config(issues.imap_path(), user="bugs")
530 issues.check_mailbox_separate()
531
532 def test_check_n_refuses_non_ascii_digits(self):
533 for bad in ("\u0661", "\uff11", "0", "-1", "1 ", ""):
534 with self.assertRaises(issues.Error, msg=bad):
535 issues.check_n(bad)
536 self.assertEqual(issues.check_n("12"), 12)
537 self.ingest(mail())
538 self.assertEqual(self.cli("show", "demo", "\u0661")[0], 1)
539
540 def test_replay_template_rules(self):
541 for ok in ("replay.html?report={record}", "r/p.html?x=%20&report={record}"):
542 self.assertTrue(issues.replay_ok(ok), ok)
543 for bad in ("a%2e%2e/x.html?r={record}", "a%00.html?r={record}", "a b.html?r={record}", "a\tb.html?r={record}",
544 "a\nb.html?r={record}", "a\x7f.html?r={record}", "x.html", "/x?r={record}", "javascript:1?{record}"):
545 self.assertFalse(issues.replay_ok(bad), bad)
546
547 def test_same_message_id_is_stored_once(self):
548 self.ingest(mail(mid="<same@test>", body="One.\n"), mail(mid="<same@test>", body="Two, different bytes.\n"))
549 self.assertEqual(issues.issue_numbers("demo"), [1])
550 self.assertEqual(len(issues.load_issue("demo", 1)["messages"]), 1)
551
552 def test_a_crash_while_saving_creates_no_duplicate_and_loses_no_report(self):
553 with mock.patch.object(issues, "save_issue", side_effect=OSError("disk full")):
554 with self.assertRaises(OSError):
555 self.ingest(mail(mid="<crash@test>"))
556 self.assertEqual(self.count("done"), 0) # the mail is still in the inbox
557 self.assertIn("keys.json", os.listdir(self.state)) # remembered before the issue was saved
558 self.ingest() # next run: stored once
559 self.ingest()
560 self.assertEqual(len(issues.issue_numbers("demo")), 1)
561 self.assertEqual(len(issues.load_issue("demo", 1)["messages"]), 1)
562
563
564class Review2(Base):
565 def test_reply_record_needs_approval_in_reviewed_mode(self):
566 self.repo("rv", "public", "reviewed")
567 self.ingest(mail(repo="rv", body="First, no record.\n", mid="<a@x>"))
568 self.cli("approve", "rv", "1")
569 self.ingest(mail(repo="rv", body="Reply.\nRun record: " + record(options={"note": "unmoderated"}) + "\n",
570 mid="<b@x>", extra_headers=("In-Reply-To: <a@x>",)))
571 self.assertTrue(issues.load_issue("rv", 1)["record"])
572 self.assertFalse(os.path.exists(self.page("rv", "1", "record.json")))
573 self.assertNotIn("Watch what happened", rd(self.page("rv", "1", "index.html")))
574 self.cli("approve", "rv", "1")
575 self.assertTrue(os.path.isfile(self.page("rv", "1", "record.json")))
576
577 def test_authenticated_reply_record_is_live_in_public_mode(self):
578 self.ingest(mail(body="No record.\n", mid="<a@x>"),
579 mail(body="Reply.\nRun record: " + record() + "\n", mid="<b@x>", extra_headers=("In-Reply-To: <a@x>",)))
580 self.assertTrue(os.path.isfile(self.page("demo", "1", "record.json")))
581
582 ATTRIBUTIONS = (
583 "Am Mo., 3. Nov. 2026 um 10:00 Uhr schrieb Pavel Safronov <p@x.net>:",
584 "Le lun. 3 nov. 2026 \u00e0 10:00, Pavel Safronov <p@x.net> a \u00e9crit :",
585 "Op ma 3 nov 2026 om 10:00 schreef Pavel Safronov <p@x.net>:",
586 "Pavel Safronov <p@x.net> wrote:",
587 "El lun, 3 nov 2026 a las 10:00, Pavel Safronov escribi\u00f3:",
588 )
589
590 def test_attributions_in_any_language_cut_the_text(self):
591 for cut in self.ATTRIBUTIONS:
592 out = issues.report_text(f"My answer.\n\n{cut}\n> Pavel Safronov wrote this\nPavel Safronov again\n")
593 self.assertEqual(out, "My answer.", cut)
594 out = issues.report_text("My answer.\n________________________________\nFrom: Pavel\nSent: Monday\nTo: me\n\nPavel text")
595 self.assertEqual(out, "My answer.")
596 self.assertIn("Steps to reproduce:", issues.report_text("Steps to reproduce:\n1. tap\n")) # a plain colon line stays
597
598 def test_device_block(self):
599 d65 = "Device: " + "x" * 57
600 out = issues.report_text(f"{d65}\nAppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148\nSeed 5, level 2\n")
601 self.assertEqual(out, "Seed 5, level 2")
602 d83 = "Device: Android 378x694; Mozilla/5.0 (Android 17; Mobile; rv:159.0) Gecko/159.0 Fir"
603 self.assertEqual(len(d83), 83)
604 out = issues.report_text(f"{d83}\n\nThe game froze when I tapped twice on level 3.\n")
605 self.assertEqual(out, "The game froze when I tapped twice on level 3.")
606 out = issues.report_text(f"{d83}\nThe game froze when I tapped twice on level 3.\n") # prose right after: kept
607 self.assertEqual(out, "The game froze when I tapped twice on level 3.")
608 out = issues.report_text("Device: A\n Firefox 1\n more\nVersion 1.0\n")
609 self.assertEqual(out, "Version 1.0")
610
611 def test_unflow_keeps_pending_text_before_a_quote(self):
612 out = issues.report_text("Thanks for the fix \nit works now \n> quoted\n", flowed=True)
613 self.assertEqual(out, "Thanks for the fix it works now")
614
615 def test_unflow_signature_and_delsp_and_device(self):
616 out = issues.report_text("Hello there \n-- \nSignature name\n", flowed=True)
617 self.assertEqual(out, "Hello there")
618 self.assertEqual(issues.report_text("long wo \nrd end\n", flowed=True, delsp=True), "long word end")
619 self.assertEqual(issues.report_text("long wo \nrd end\n", flowed=True), "long wo rd end")
620 out = issues.report_text("Device: Pixel (Mobile; x) \n\nMy own paragraph.\n", flowed=True)
621 self.assertEqual(out, "My own paragraph.")
622 out = issues.report_text("Device: Pixel \nSeed 5, level 2\n", flowed=True)
623 self.assertEqual(out, "Seed 5, level 2")
624
625 def test_flowed_mail_end_to_end_with_delsp(self):
626 raw = mail(body="Thanks for the fix \nit works now \n> quoted\n").replace(
627 b"Content-Type: text/plain; charset=utf-8", b"Content-Type: text/plain; charset=utf-8; format=flowed; delsp=yes")
628 self.ingest(raw)
629 self.assertIn("Thanks for the fixit works now", issues.load_issue("demo", 1)["messages"][0]["text"])
630
631 def test_crash_recovery_does_not_confuse_a_reused_slot(self):
632 real = issues.save_issue
633 calls = []
634
635 def boom(issue):
636 calls.append(1)
637 if len(calls) == 1:
638 raise OSError("disk full")
639 real(issue)
640 a, b = mail(body="Report A\n", mid="<A@x>"), mail(body="Report B\n", mid="<B@x>")
641 with mock.patch.object(issues, "save_issue", boom):
642 with self.assertRaises(OSError):
643 self.ingest(a)
644 box = mailbox.Maildir(self.maildir, create=False)
645 k = list(box.keys())[0]
646 a_raw = box.get_bytes(k)
647 box.remove(k)
648 self.ingest(b) # takes issue 1 while A's key still points at it
649 box.add(a_raw)
650 self.ingest()
651 texts = sorted(m["text"] for i in issues.load_issues("demo") for m in i["messages"])
652 self.assertEqual(texts, ["Report A", "Report B"])
653
654 def test_no_numbered_folder_without_an_issue_after_a_crash(self):
655 real = issues.save_issue
656 calls = []
657
658 def boom(issue):
659 calls.append(1)
660 if len(calls) == 1:
661 raise OSError("x")
662 real(issue)
663 with mock.patch.object(issues, "save_issue", boom):
664 with self.assertRaises(OSError):
665 self.ingest(mail(body="A\nRun record: " + record() + "\n"))
666 self.assertEqual(issues.issue_numbers("demo"), [])
667 self.ingest()
668 self.assertEqual(issues.issue_numbers("demo"), [1])
669 self.assertTrue(os.path.isfile(self.page("demo", "1", "record.json")))
670
671 def test_mailbox_comparison_ignores_case(self):
672 theirs = os.path.join(self.tmp, "comments", "imap.json")
673 with mock.patch.dict(os.environ, {"FW_COMMENTS_IMAP": theirs}):
674 Review.imap_config(self, theirs)
675 Review.imap_config(self, issues.imap_path(), host="IMAP.Example.ORG", user="U", folder="inbox")
676 with self.assertRaises(issues.Error):
677 issues.check_mailbox_separate()
678
679
680class Review3(Base):
681 def test_real_text_with_colons_and_brackets_is_not_cut(self):
682 for text in ("I pressed <Space> and\nthen this happened:\nthe bomb vanished",
683 "The game crashed when I pressed\nthe <Enter> key:\nnothing else\n",
684 "What happened:\nI pressed <Enter> on the menu:\nnothing",
685 "He wrote notes and\nthen it broke:\nboom"):
686 self.assertEqual(issues.report_text(text), text.strip(), text)
687
688 def test_attribution_right_after_a_text_line(self):
689 text = ("Still happens in 0.3.0\nOn Tue, Nov 3, 2026 at 7:51 PM Code <bugs-k7q4+demo@fuzzyworld.net> wrote:\n> x\n"
690 "Code again\n")
691 self.assertEqual(issues.report_text(text), "Still happens in 0.3.0")
692
693 def test_weak_attribution_never_empties_the_text(self):
694 self.assertEqual(issues.report_text("Steps:\n> quoted by the app"), "Steps:")
695 self.assertEqual(issues.report_text("Hello\nSteps:\n> quoted"), "Hello")
696 self.assertEqual(issues.report_text("Pavel <p@x.net> wrote:\n> hi"), "") # a real attribution may leave nothing
697
698 def test_wrapped_user_agent_starting_with_version_slash_is_dropped(self):
699 text = ("Bug here\nDevice: Mozilla/5.0 (iPhone; CPU iPhone OS 17_0) AppleWebKit/605.1.15\n"
700 "Version/17.0 Mobile/15E148 Safari/604.1\nSeed 1\n")
701 self.assertEqual(issues.report_text(text), "Bug here\nSeed 1")
702 self.assertIsNotNone(issues.KNOWN_LINE_RE.match("Version 0.2.0 (rules 43)"))
703 self.assertIsNone(issues.KNOWN_LINE_RE.match("Version/17.0 Mobile"))
704
705 def test_flowed_device_line_is_not_joined_onto_the_previous_line(self):
706 text = "Bug \nDevice: Pixel (Mobile; x)\nSeed 1\n"
707 self.assertEqual(issues.report_text(text, flowed=True), "Bug \nSeed 1")
708 self.assertEqual(issues.report_text("Bug \nRun record: {}\nafter\n", flowed=True), "Bug \nafter")
709
710
711class Review4(Base):
712 R = staticmethod(issues.report_text)
713
714 def test_blank_line_ends_a_flowed_paragraph(self):
715 self.assertEqual(issues.unflow(["Para one. ", "", "Para two."]), ["Para one. ", "", "Para two."])
716 self.assertEqual(self.R("Para one. \n\nPara two.", flowed=True), "Para one. \n\nPara two.")
717 for text in ("It broke. \n\nOn Mon, 3 Nov 2026, Bob Smith <bob@x.com> wrote:\n> old",
718 "It broke.\n\nOn Mon, 3 Nov 2026, Bob Smith <bob@x.com> wrote:\n> old"):
719 self.assertEqual(self.R(text, flowed=True), "It broke.")
720
721 def test_wrapped_attributions_do_not_leak_names(self):
722 cases = {
723 "en_wrap_noaddr": "Bug text.\n\nOn Mon, 3 Nov 2026 at 10:00, Bob Smith\nwrote:\n> old",
724 "en_wrap_mid": "Bug text.\n\nOn Mon, 3 Nov 2026 at 10:00, Bob\nSmith <bob@x.com> wrote:\n> old",
725 "en_name_nl": "Bug text.\n\nOn Mon, 3 Nov 2026 at 10:00 Bob Smith <\nbob@x.com> wrote:\n> old",
726 "en_no_blank": "Bug text.\nOn Mon, 3 Nov 2026 at 10:00, Bob Smith <bob@x.com> wrote:\n> old",
727 "de_wrap": "Bug text.\n\nAm 03.11.2026 um 10:00 schrieb Hans M\u00fcller\n<hans@x.de>:\n> alt",
728 "de_wrap2": "Bug text.\n\nAm Mo., 3. Nov. 2026 um 10:00 Uhr schrieb Hans M\u00fcller <\nhans@x.de>:\n> alt",
729 "fr_wrap": "Bug text.\n\nLe lun. 3 nov. 2026 \u00e0 10:00, Jean Dupont <jean@x.fr> a\n\u00e9crit :\n> vieux",
730 "fr_nbsp": "Bug text.\n\nLe lun. 3 nov. 2026 \u00e0 10:00, Jean Dupont <jean@x.fr> a \u00e9crit\u00a0:\n> vieux",
731 "nl_wrap": "Bug text.\n\nOp ma 3 nov 2026 om 10:00 schreef Jan Jansen\n<jan@x.nl>:\n> oud",
732 "es_wrap": "Bug text.\n\nEl lun, 3 nov 2026 a las 10:00, Juan P\u00e9rez (<juan@x.es>)\nescribi\u00f3:\n> viejo",
733 "gmail_blank_before_quote": "Bug text.\n\nOn 3 Nov 2026, at 10:00, Bob Smith <bob@x.com> wrote:\n\n> old",
734 "schrieb_am": "Bug text.\n\nBob Smith <bob@x.com> schrieb am Mo., 3. Nov. 2026, 10:00:\n> alt",
735 "schrieb_am_wrapped": "Bug text.\n\nBob Smith <bob@x.com> schrieb am Mo., 3. Nov. 2026,\n10:00:\n> alt",
736 "wrote_plain_name": "Bug text.\n\nBob Smith wrote:\n> old",
737 "flowed": "Bug text. \n\nOn Mon, 3 Nov 2026 at 10:00, Bob Smith \n<bob@x.com> wrote:\n> old",
738 }
739 for label, text in cases.items():
740 out = self.R(text, flowed=(label == "flowed"))
741 self.assertEqual(out.strip(), "Bug text.", label)
742
743 def test_non_english_outlook_header_blocks_are_cut(self):
744 blocks = (
745 "________________________________\nVon: Bob Smith <bob@x.com>\nGesendet: Montag, 3. November 2026 10:00\nAn: bugs\nBetreff: Re: x\n\nalt",
746 "De : Bob Smith <bob@x.com>\nEnvoy\u00e9 : lundi 3 novembre 2026 10:00\n\u00c0 : bugs\nObjet : Re: x\n\nvieux",
747 "Van: Bob Smith\nVerzonden: maandag 3 november 2026 10:00\nAan: bugs\nOnderwerp: Re: x\n\noud",
748 "From: Bob Smith\nSent: Monday, November 3, 2026 10:00\nTo: bugs\nSubject: Re: x\n\nold",
749 )
750 for b in blocks:
751 self.assertEqual(self.R("Bug text.\n\n" + b), "Bug text.", b)
752
753 def test_header_block_rule_leaves_the_games_own_template_and_plain_lists_alone(self):
754 with open(os.path.join(FIXTURES, "explodorama-report.eml"), "rb") as f:
755 raw = f.read()
756 msg = issues.message_from_bytes(raw, policy=issues.policy.default)
757 text = self.R(msg.get_body(("plain",)).get_content())
758 self.assertIn("autopilot walked me into death", text)
759 self.assertIn("Seed 1464004408", text)
760 self.assertIn("Errors caught: none", text)
761 lst = "Expected: the bomb\nActual: nothing <Enter>\nSteps: tap\nNote: twice"
762 self.assertEqual(self.R(lst), lst)
763
764 def test_bare_verb_in_prose_is_not_an_attribution(self):
765 for text in ("I pressed <Enter> and wrote:\nnothing happened", "He wrote:\nthe thing broke",
766 "On the menu screen I typed and wrote:\nnothing"):
767 self.assertEqual(self.R(text), text)
768
769 def test_weak_match_needs_everything_after_it_to_be_quoted(self):
770 self.assertEqual(self.R("It crashed.\nError shown:\n> TypeError: x\nThat was it."),
771 "It crashed.\nError shown:\nThat was it.")
772 self.assertEqual(self.R("Error shown:\n> TypeError: x\nThat was it."), "Error shown:\nThat was it.")
773 self.assertEqual(self.R("It crashed.\nError shown:\n> TypeError: x\n\n> more"), "It crashed.")
774
775 def test_nfkc_comes_before_the_device_filter(self):
776 self.assertEqual(self.R("\uff24evice: Pavel's iPhone\nbroke"), "broke")
777 self.assertEqual(self.R("\uff32un record: {}\nafter"), "after")
778
779
780class PageLayout(Base):
781 def test_state_badge_in_header_and_list(self):
782 self.ingest(mail())
783 self.assertIn('<span class="badge badge-success">open</span>', rd(self.page("demo", "1", "index.html")))
784 self.assertIn('<span class="badge badge-success">open</span>', rd(self.page("demo", "index.html")))
785 self.cli("close", "demo", "1")
786 self.assertIn('<span class="badge">closed</span>', rd(self.page("demo", "1", "index.html")))
787 self.assertIn(">closed</span>", rd(self.page("demo", "index.html")))
788
789 def test_details_table_from_the_game_template(self):
790 body = ("What happened:\nIt broke.\n\n---\nVersion 0.2.0 (rules 43), commit 805e7a9\n"
791 "Seed 1464004408, level 4, 66 turns taken, 3 lives, playing; modes: none\nErrors caught: none\n")
792 self.ingest(mail(body=body))
793 page = rd(self.page("demo", "1", "index.html"))
794 for want in ("<th>Version</th><td>0.2.0</td>", "<th>Rules</th><td>43</td>", "<th>Commit</th><td>805e7a9</td>",
795 "<th>Seed</th><td>1464004408</td>", "<th>Level</th><td>4</td>", "<th>Turns</th><td>66</td>",
796 "<th>Lives</th><td>3</td>", "<th>Status</th><td>playing</td>", "<th>Modes</th><td>none</td>"):
797 self.assertIn(want, page)
798 self.assertIn("Errors caught: none", page) # the full text stays below
799 self.assertNotIn("issue-details", rd(self.page("demo", "index.html")))
800
801 def test_details_table_only_for_matched_fields_escaped_and_capped(self):
802 t = issues.details_table("Seed 5, level 2, 9 turns taken, 1 lives, dead; modes: <b>" + "x" * 80)
803 self.assertNotIn("<th>Version</th>", t)
804 self.assertIn("<th>Seed</th><td>5</td>", t)
805 self.assertIn("&lt;b&gt;", t)
806 self.assertNotIn("<b>", t)
807 self.assertNotIn("x" * 41, t)
808 self.assertEqual(issues.details_table("Seed five, level two"), "")
809 self.ingest(mail(body="Plain report with no template.\n"))
810 self.assertNotIn("issue-details", rd(self.page("demo", "1", "index.html")))
811
812 def test_dash_line_becomes_a_thin_rule(self):
813 out = issues.text_blocks("First part.\n------------\nSecond <b>part</b>.\n---\nThird.")
814 self.assertEqual(out.count("<hr"), 2)
815 self.assertEqual(out.count('<pre class="issue-text">'), 3)
816 self.assertNotIn("---", out)
817 self.assertIn("Second &lt;b&gt;part&lt;/b&gt;.", out)
818 self.assertEqual(issues.text_blocks("no separator"), '<pre class="issue-text">no separator</pre>')
819 self.assertNotIn("<hr", issues.text_blocks("a-b -- c\n- item"))
820 self.ingest(mail(body="Above.\n---\nBelow.\n"))
821 page = rd(self.page("demo", "1", "index.html"))
822 self.assertIn('<hr class="issue-sep">', page)
823 self.assertNotIn("\n---\n", page)
824
825 def test_reporters_are_labelled_by_first_appearance_without_hashes(self):
826 self.ingest(mail(mid="<a@x>", sender="A <alice@example.org>", auth=("mx; dkim=pass header.d=example.org",)),
827 mail(mid="<b@x>", sender="B <bob@example.org>", auth=("mx; dkim=pass header.d=example.org",),
828 body="Bob here.\n", extra_headers=("In-Reply-To: <a@x>",)),
829 mail(mid="<c@x>", sender="A <alice@example.org>", auth=("mx; dkim=pass header.d=example.org",),
830 body="Alice again.\n", extra_headers=("In-Reply-To: <a@x>",)))
831 page = rd(self.page("demo", "1", "index.html"))
832 self.assertLess(page.index("Reporter A"), page.index("Reporter B"))
833 self.assertEqual(page.count("Reporter A"), 2)
834 self.assertEqual(page.count("Reporter B"), 1)
835 stored = issues.load_issue("demo", 1)
836 for m in stored["messages"]:
837 self.assertEqual(m["author"], "Reporter")
838 self.assertNotIn(m["sender_hash"], page)
839 self.assertEqual(issues.reporter_labels([{"sender_hash": "h1"}, {"sender_hash": "h2"}, {"sender_hash": "h1"}, {}]),
840 ["Reporter A", "Reporter B", "Reporter A", "Reporter"])
841
842 def test_previous_and_next_links_skip_hidden_issues(self):
843 self.ingest(mail(subject="One"), mail(subject="Two"), mail(subject="Three"), mail(subject="Four"))
844 self.cli("hide", "demo", "3")
845 p1, p2, p4 = (rd(self.page("demo", str(n), "index.html")) for n in (1, 2, 4))
846 self.assertNotIn("\u2190 #", p1)
847 self.assertIn("#2 \u2192", p1)
848 self.assertIn("\u2190 #1", p2)
849 self.assertIn("#4 \u2192", p2) # #3 is hidden
850 self.assertIn("\u2190 #2", p4)
851 self.assertNotIn("#5", p4)
852 self.assertNotIn("#3", p2.split('<p class="issue-nav">')[1])
853
854 def test_single_issue_has_no_nav_and_css_wraps_text(self):
855 self.ingest(mail())
856 self.assertNotIn("issue-nav", rd(self.page("demo", "1", "index.html")))
857 css = rd(os.path.join(ROOT, "server", "site", "style.css"))
858 self.assertIn("pre.message, pre.issue-text {\n white-space: pre-wrap;\n overflow-wrap: anywhere;", css)
859 self.assertIn("max-width: 100%", css)
860
861 def test_pages_still_hold_no_script_or_hash(self):
862 self.ingest(mail(subject="<script>x</script>", body="<script>y</script>\n---\n<script>z</script>\n"))
863 for p in (self.page("demo", "index.html"), self.page("demo", "1", "index.html")):
864 self.assertNotIn("<script", rd(p).replace("&lt;script", ""))
865
866
867class FwScript(Base):
868 """server/fw publish_one and `fw issues <repo> <mode>` against temp directories (flock is faked: macOS has none)."""
869
870 def setUp(self):
871 super().setUp()
872 self.bin = os.path.join(self.tmp, "bin")
873 os.makedirs(self.bin)
874 shim = os.path.join(self.bin, "flock")
875 with open(shim, "w") as f:
876 f.write("#!/bin/sh\nexit 0\n")
877 os.chmod(shim, 0o755)
878 self.env = {**os.environ, "PATH": self.bin + os.pathsep + os.environ["PATH"], "HOME": self.tmp, "FW_GIT_ROOT": self.git,
879 "FW_CODE_SITE": self.site, "FW_STATE": self.fwstate, "FW_LOGS": os.path.join(self.tmp, "logs"),
880 "FW_APPS_SITE": self.apps, "FW_TOOLS": ROOT}
881 os.makedirs(self.site)
882
883 def fw(self, *args):
884 return subprocess.run(["bash", os.path.join(ROOT, "server", "fw"), *args], env=self.env, capture_output=True, text=True)
885
886 def config(self, repo="demo"):
887 return subprocess.run(["git", "-C", os.path.join(self.git, repo + ".git"), "config", "--get", "fuzzyworld.issues"],
888 capture_output=True, text=True).stdout.strip()
889
890 def test_repo_tabs_carry_an_issues_tab_only_when_there_are_public_issues(self):
891 view = types.SimpleNamespace(name="demo", url=lambda kind: f"/demo/{kind}")
892 def tabs():
893 return re.search(r'<nav class="repo-nav">.*?</nav>', issues.gen.page("t", "b", view=view), re.S).group(0)
894 self.assertNotIn(">Issues</a>", tabs())
895 self.ingest(mail())
896 self.assertIn(">Issues</a>", tabs())
897
898 def test_mode_setter_and_publish_one(self):
899 self.ingest(mail())
900 r = self.fw("issues", "demo", "reviewed")
901 self.assertEqual(r.returncode, 0, r.stderr)
902 self.assertEqual(self.config(), "reviewed")
903 self.assertFalse(os.path.exists(self.page()))
904 r = self.fw("issues", "demo", "public")
905 self.assertEqual(r.returncode, 0, r.stderr)
906 self.assertTrue(os.path.isfile(self.page("demo", "1", "index.html")))
907 shutil.rmtree(self.page())
908 r = self.fw("publish", "demo") # publish_one runs `issues.py publish --locked` before gen.py repo
909 self.assertEqual(r.returncode, 0, r.stderr)
910 self.assertTrue(os.path.isfile(self.page("demo", "index.html")))
911 self.assertIn('/demo/issues/', rd(os.path.join(self.site, "demo", "index.html"))) # the summary links to Issues
912 r = self.fw("issues", "demo", "private")
913 self.assertEqual(r.returncode, 0, r.stderr)
914 self.assertFalse(os.path.exists(self.page()))
915
916 def test_other_words_and_unknown_repos_are_refused(self):
917 for args in (("issues", "demo", "bogus"), ("issues", "demo", "Public"), ("issues", "nosuch", "public"), ("issues", "demo")):
918 r = self.fw(*args)
919 self.assertNotEqual(r.returncode, 0, args)
920 self.assertEqual(self.config(), "public") # unchanged
921
922
923class Replay(Base):
924 def app(self, name, repo="demo", replay="replay.html?report={record}", index=True):
925 d = os.path.join(self.apps, name)
926 os.makedirs(d)
927 if index:
928 with open(os.path.join(d, "index.html"), "w") as f:
929 f.write("x")
930 with open(os.path.join(d, ".fuzzyworld.json"), "w") as f:
931 json.dump({"app": name, "repo": repo, "replay": replay}, f)
932
933 def test_no_link_without_a_replay_key_or_app(self):
934 self.assertEqual(issues.replay_link("demo", 1), "")
935 self.app("demo", replay="")
936 self.assertEqual(issues.replay_link("demo", 1), "")
937
938 def test_link_is_encoded_and_prefers_the_app_named_like_the_repo(self):
939 self.app("demo-next", replay="next.html?r={record}")
940 self.app("demo")
941 link = issues.replay_link("demo", 3)
942 self.assertIn('href="https://apps.fuzzyworld.net/demo/replay.html?report='
943 'https%3A%2F%2Fcode.fuzzyworld.net%2Fdemo%2Fissues%2F3%2Frecord.json"', link)
944 self.assertIn('rel="noopener"', link)
945 self.assertNotIn("next.html", link)
946 self.assertNotIn("<iframe", link)
947
948 def test_link_needs_an_index_page_and_a_safe_template(self):
949 self.app("demo", index=False)
950 self.assertEqual(issues.replay_link("demo", 1), "")
951 self.app("other", repo="demo", replay="javascript:alert({record})")
952 self.assertEqual(issues.replay_link("demo", 1), "")
953
954 def test_other_repos_apps_are_ignored(self):
955 self.app("game", repo="somebody-else")
956 self.assertEqual(issues.replay_link("demo", 1), "")
957
958 def test_page_shows_the_link_only_with_a_published_record(self):
959 self.app("demo")
960 self.ingest(mail(mid="<a@test>"), mail(subject="With record", body="Boom.\nRun record: " + record() + "\n"))
961 self.assertNotIn("Watch what happened", rd(self.page("demo", "1", "index.html")))
962 page = rd(self.page("demo", "2", "index.html"))
963 self.assertIn("Watch what happened", page)
964 self.assertIn("replay.html?report=", page)
965
966 def test_summary_link_refresh_follows_the_index(self):
967 self.ingest(mail())
968 self.cli("hide", "demo", "1")
969 self.assertEqual(self.summaries, ["demo", "demo"]) # appeared, then disappeared
970
971
972class AdminJson(Base):
973 """The machine-readable commands `fw admin` runs: list --json, show --json, record."""
974 PRIVATE = ("sender_hash", "mkey", "mid")
975
976 def js(self, *argv):
977 code, out, err = self.cli(*argv)
978 self.assertEqual((code, err), (0, ""))
979 for word in self.PRIVATE:
980 self.assertNotIn(word, out)
981 return json.loads(out)
982
983 def test_list_json_fields_and_waiting(self):
984 self.repo("rev", "public", "reviewed")
985 self.repo("priv", "public", "private")
986 self.ingest(mail(repo="rev", body="Boom.\nRun record: " + record() + "\n"), mail(), mail(repo="priv"))
987 data = self.js("list", "--json")
988 self.assertEqual(data["repos"]["rev"], {"mode": "reviewed", "public_repo": True})
989 self.assertEqual(data["repos"]["priv"]["mode"], "private")
990 by = {(i["repo"], i["n"]): i for i in data["issues"]}
991 rev, pub, priv = by[("rev", 1)], by[("demo", 1)], by[("priv", 1)]
992 self.assertEqual((rev["public"], rev["waiting"], rev["record"], rev["messages"], rev["shown"]), (False, True, True, 1, 0))
993 self.assertEqual((pub["public"], pub["waiting"], pub["shown"]), (True, False, 1))
994 self.assertFalse(priv["waiting"])
995 self.assertEqual(set(rev), {"repo", "n", "title", "created", "updated", "state", "approved", "hidden", "public",
996 "record", "messages", "shown", "waiting"})
997 self.cli("approve", "rev", "1")
998 rev = {(i["repo"], i["n"]): i for i in self.js("list", "rev", "--json")["issues"]}[("rev", 1)]
999 self.assertEqual((rev["public"], rev["waiting"], rev["approved"]), (True, False, True))
1000 self.cli("hide", "rev", "1")
1001 hidden = {(i["repo"], i["n"]): i for i in self.js("list", "--json")["issues"]}[("rev", 1)]
1002 self.assertEqual((hidden["hidden"], hidden["waiting"]), (True, False))
1003
1004 def test_list_json_new_reply_waits_in_public_mode_when_unauthenticated(self):
1005 self.ingest(mail(mid="<a@test>"))
1006 self.ingest(mail(subject="Re: Bug: crash on start", body="more", auth=("mx; dkim=fail",),
1007 extra_headers=("In-Reply-To: <a@test>",)))
1008 issue = self.js("list", "--json")["issues"][0]
1009 self.assertEqual((issue["messages"], issue["shown"], issue["waiting"]), (2, 1, True))
1010
1011 def test_show_json_full_text_of_unpublished_report(self):
1012 self.repo("rev", "public", "reviewed")
1013 self.ingest(mail(repo="rev", mid="<a@test>", body="Version 1.2 (rules 43), commit abcdef1\nSeed 7, level 2, 30 turns taken, 3 lives, alive; modes: none\nSecret words\n"),
1014 mail(repo="rev", subject="Second"))
1015 data = self.js("show", "rev", "1", "--json")
1016 self.assertEqual((data["mode"], data["public_repo"], data["public"], data["page_url"]), ("reviewed", True, False, None))
1017 self.assertIn("Secret words", data["messages"][0]["text"])
1018 self.assertEqual(data["messages"][0], {"label": "Reporter A", "date": data["messages"][0]["date"], "authenticated": True,
1019 "approved": False, "shown": False, "text": data["messages"][0]["text"]})
1020 self.assertIn(["Commit", "abcdef1"], data["details"])
1021 self.assertEqual((data["prev"], data["next"]), (None, 2))
1022 self.assertEqual(self.js("show", "rev", "2", "--json")["prev"], 1)
1023 self.cli("approve", "rev", "1")
1024 data = self.js("show", "rev", "1", "--json")
1025 self.assertEqual((data["public"], data["page_url"], data["messages"][0]["shown"]), (True, "https://code.fuzzyworld.net/rev/issues/1/", True))
1026
1027 def test_show_json_labels_and_watch_url(self):
1028 d = os.path.join(self.apps, "demo")
1029 os.makedirs(d)
1030 open(os.path.join(d, "index.html"), "w").write("x")
1031 json.dump({"app": "demo", "repo": "demo", "replay": "replay.html?report={record}"}, open(os.path.join(d, ".fuzzyworld.json"), "w"))
1032 self.ingest(mail(mid="<a@test>", body="Boom.\nRun record: " + record() + "\n"))
1033 self.ingest(mail(sender="Other <other@example.org>", subject="Re: x", body="me too", extra_headers=("In-Reply-To: <a@test>",)))
1034 data = self.js("show", "demo", "1", "--json")
1035 self.assertEqual([m["label"] for m in data["messages"]], ["Reporter A", "Reporter B"])
1036 self.assertTrue(data["record_shown"])
1037 self.assertIn("replay.html?report=", data["watch_url"])
1038 self.assertEqual(issues.replay_href("demo", 1), data["watch_url"])
1039 self.assertEqual(issues.replay_href("nothing", 1), "")
1040
1041 def test_show_json_errors_and_plain_show_unchanged(self):
1042 self.ingest(mail())
1043 self.assertEqual(self.cli("show", "demo", "9", "--json")[0], 1)
1044 code, out, _ = self.cli("show", "demo", "1")
1045 self.assertIn('"sender_hash"', out) # the plain command still prints the stored file
1046 self.assertEqual(self.cli("show", "demo", "1", "--bogus")[0], 2)
1047
1048 def test_details_rows_match_the_table(self):
1049 text = "Version 1.2 (rules 43), commit abcdef1\nSeed 7, level 2, 30 turns taken, 3 lives, alive; modes: none"
1050 self.assertEqual(dict(issues.details_rows(text))["Seed"], "7")
1051 self.assertIn("<th>Seed</th><td>7</td>", issues.details_table(text))
1052 self.assertEqual(issues.details_rows("nothing"), [])
1053 self.assertEqual(issues.details_table("nothing"), "")
1054
1055 def test_record_prints_the_stored_file_even_when_unpublished(self):
1056 self.repo("rev", "public", "reviewed")
1057 self.ingest(mail(repo="rev", body="Boom.\nRun record: " + record() + "\n"), mail(repo="rev", subject="No record"))
1058 code, out, _ = self.cli("record", "rev", "1")
1059 self.assertEqual(code, 0)
1060 self.assertEqual(out, rd(os.path.join(self.state, "repos", "rev", "1", "record.json")))
1061 self.assertFalse(os.path.exists(self.page("rev")))
1062 code, out, err = self.cli("record", "rev", "2")
1063 self.assertEqual((code, out), (1, ""))
1064 self.assertIn("no record for rev#2", err)
1065
1066
1067class ImapParameters(unittest.TestCase):
1068 def test_defaults_are_the_comments_values_and_issues_can_override(self):
1069 c = issues.comments.IMAPSource()
1070 c.cfg, c.prefix = {"done_folder": "", "bad_folder": ""}, "INBOX."
1071 self.assertEqual(c._folder("bad"), "INBOX.comments-bad")
1072 self.assertEqual(c.max_size, issues.comments.IMAP_MAX_SIZE)
1073 i = issues.comments.IMAPSource(cfg_path="/x/imap.json", stem="issues", max_size=5, log=print)
1074 i.cfg, i.prefix = {"done_folder": "", "bad_folder": ""}, "INBOX."
1075 self.assertEqual(i._folder("done"), "INBOX.issues-done")
1076 self.assertFalse(i.available())
1077
1078
1079if __name__ == "__main__":
1080 unittest.main()