fuzzyworld

fuzzyworld-tools / files / tests / test_issues.py

test_issues.py

Blame: the commit that last changed each line · file · history

55e5fc3e 2026-11-04 Pavel Safronov1"""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
85702b14 2026-11-04 Pavel Safronov13import types
14import re
55e5fc3e 2026-11-04 Pavel Safronov15import 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
859a0fd4 2026-11-04 Pavel Safronov327 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
55e5fc3e 2026-11-04 Pavel Safronov342 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
09c576d4 2026-11-04 Pavel Safronov780class 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
55e5fc3e 2026-11-04 Pavel Safronov867class 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
85702b14 2026-11-04 Pavel Safronov890 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
55e5fc3e 2026-11-04 Pavel Safronov898 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
85702b14 2026-11-04 Pavel Safronov972class 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
55e5fc3e 2026-11-04 Pavel Safronov1067class 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()