fuzzyworld

fuzzyworld-tools / files / tests / test_admin.py

test_admin.py

28.0 KB · history · blame

1"""Tests for lib/admin.py (the local `fw admin` web page).
2
3Run with: python3 -m unittest discover tests
4"""
5
6import http.client
7import importlib.util
8import json
9import os
10import re
11import shutil
12import subprocess
13import tempfile
14import threading
15import unittest
16from unittest import mock
17
18ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")
19spec = importlib.util.spec_from_file_location("admin", os.path.join(ROOT, "lib", "admin.py"))
20admin = importlib.util.module_from_spec(spec)
21spec.loader.exec_module(admin)
22
23CID = "20241231T101010Z-abcdef"
24EVIL = "<script>alert(1)</script>"
25
26
27def issue_list(**over):
28 i = {"repo": "demo", "n": 3, "title": "Crash", "created": "2026-11-01T10:00:00Z", "updated": "2026-11-02T10:00:00Z",
29 "state": "open", "approved": False, "hidden": False, "public": False, "record": True, "messages": 2, "shown": 1,
30 "waiting": True}
31 i.update(over)
32 return {"repos": {"demo": {"mode": "reviewed", "public_repo": True}}, "issues": [i]}
33
34
35def issue_show(**over):
36 d = {"repo": "demo", "n": 3, "title": "Crash", "created": "2026-11-01T10:00:00Z", "state": "open", "approved": False,
37 "hidden": False, "mode": "reviewed", "public_repo": True, "public": False, "page_url": None, "record": True,
38 "record_shown": False, "watch_url": None, "details": [["Version", "1.2"], ["Commit", "abcdef1"]], "prev": 2, "next": None,
39 "messages": [{"label": "Reporter A", "date": "2026-11-01T10:00:00Z", "authenticated": True, "approved": False,
40 "shown": False, "text": "It crashed."}]}
41 d.update(over)
42 return d
43
44
45def comment_list(**over):
46 c = {"id": CID, "site": "writing", "path": "/2024/05/01/hello/", "post_url": "https://writing.fuzzyworld.net/2024/05/01/hello/",
47 "author": "Ann", "date": "2026-11-01T10:00:00Z", "source": "email", "parent": None, "text": "Nice post.", "trusted": True}
48 c.update(over)
49 return {"pending": [c], "approved": [{**c, "id": "20241230T101010Z-123456", "source": None, "text": "Old one"}]}
50
51
52def deploys_reply(**over):
53 d = {"repos": [{"repo": "tools", "head": "abc1234", "targets": [
54 {"target": "server", "description": "the <b>server</b>", "paths": ["server"], "per_machine": False,
55 "entries": [{"key": "tools/server", "machine": None, "commit": "1111111", "at": "2026-11-04 14:22", "state": "pending",
56 "behind": {"count": 10, "commits": [["2222222", "Fix <i>it</i>"]] * 8}}]},
57 {"target": "laptop-runner", "description": "", "paths": ["runner"], "per_machine": True,
58 "entries": [{"key": "tools/laptop-runner@mac-a", "machine": "mac-a", "commit": "3333333+uncommitted", "at": "t",
59 "state": "pending", "behind": None}]},
60 {"target": "vscode", "description": "", "paths": ["vscode-web"], "per_machine": False,
61 "entries": [{"key": "tools/vscode", "machine": None, "commit": "", "at": "", "state": "never", "behind": None}]}]}],
62 "apps": [{"app": "zed", "title": "Zed", "repo": "zed", "commit": "4444444", "deployed": "2026-11-03T10:00:00Z", "head": "4444444",
63 "state": "current", "behind": {"count": 0, "commits": []}},
64 {"app": "alpha", "title": EVIL, "repo": "alpha", "commit": "5555555", "deployed": "2026-11-03T10:00:00Z", "head": "6666666",
65 "state": "pending", "behind": {"count": 1, "commits": [["6666666", "More"]]}}]}
66 d.update(over)
67 return d
68
69
70class FakeRunner:
71 host = "fake"
72
73 def __init__(self):
74 self.calls, self.replies, self.errors = [], {}, {}
75
76 def run(self, args):
77 self.calls.append(list(args))
78 if tuple(args) in self.errors:
79 raise admin.FwError(self.errors[tuple(args)])
80 reply = self.replies.get(tuple(args), "ok")
81 return reply if isinstance(reply, str) else json.dumps(reply)
82
83 def json(self, args):
84 return json.loads(self.run(args))
85
86
87class Base(unittest.TestCase):
88 def setUp(self):
89 self.tmp = tempfile.mkdtemp()
90 self.addCleanup(shutil.rmtree, self.tmp)
91 p = mock.patch.object(admin.Handler, "log_request")
92 p.start()
93 self.addCleanup(p.stop)
94 self.fake = FakeRunner()
95 self.fake.replies[("issues", "list", "--json")] = issue_list()
96 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show()
97 self.fake.replies[("comments", "list", "--json")] = comment_list()
98 self.fake.replies[("deploys", "--json")] = deploys_reply()
99 self.server = admin.make_server(self.fake, 0, code_dir=self.tmp)
100 self.port = self.server.port
101 t = threading.Thread(target=self.server.serve_forever, daemon=True)
102 t.start()
103 self.addCleanup(t.join)
104 self.addCleanup(self.server.server_close)
105 self.addCleanup(self.server.shutdown)
106 self.cookie = f"fwadmin-{self.port}={self.server.session}"
107
108 def req(self, method, path, body=None, headers=None, login=True, token=True):
109 h = {"Host": f"127.0.0.1:{self.port}"}
110 if login:
111 h["Cookie"] = self.cookie
112 if body is not None:
113 body = dict(body)
114 if token:
115 body.setdefault("token", self.server.csrf)
116 from urllib.parse import urlencode
117 body = urlencode(body)
118 h["Content-Type"] = "application/x-www-form-urlencoded"
119 h.update(headers or {})
120 c = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
121 try:
122 c.request(method, path, body=body, headers=h)
123 r = c.getresponse()
124 return r.status, dict(r.getheaders()), r.read().decode()
125 finally:
126 c.close()
127
128 def get(self, path, **kw):
129 return self.req("GET", path, **kw)
130
131 def post(self, path, body, **kw):
132 return self.req("POST", path, body=body, **kw)
133
134
135class Access(Base):
136 def test_no_cookie_is_forbidden(self):
137 self.assertEqual(self.get("/reports", login=False)[0], 403)
138 self.assertEqual(self.get("/deploy", login=False)[0], 403)
139 self.assertEqual(self.get("/style.css", login=False)[0], 403)
140 self.assertEqual(self.fake.calls, [])
141
142 def test_login_works_once(self):
143 key = self.server.login_key
144 status, headers, _ = self.get(f"/login?k={key}", login=False)
145 self.assertEqual((status, headers["Location"]), (303, "/reports"))
146 cookie = headers["Set-Cookie"]
147 self.assertIn(f"fwadmin-{self.port}={self.server.session}", cookie)
148 for word in ("HttpOnly", "SameSite=Strict", "Path=/"):
149 self.assertIn(word, cookie)
150 self.assertEqual(self.get(f"/login?k={key}", login=False)[0], 403)
151 self.assertEqual(self.get("/login?k=wrong", login=False)[0], 403)
152 self.assertEqual(self.get("/reports")[0], 200)
153
154 def test_wrong_login_key_does_not_burn_the_real_one(self):
155 self.assertEqual(self.get("/login?k=nope", login=False)[0], 403)
156 self.assertEqual(self.get(f"/login?k={self.server.login_key}", login=False)[0], 303)
157
158 def test_wrong_host_is_421(self):
159 self.assertEqual(self.get("/reports", headers={"Host": "evil.example:80"})[0], 421)
160 self.assertEqual(self.get("/reports", headers={"Host": f"localhost:{self.port}"})[0], 421)
161 self.assertEqual(self.post("/comments", {"id": CID, "action": "approve"}, headers={"Host": "evil.example"})[0], 421)
162 self.assertEqual(self.fake.calls, [])
163
164 def test_post_without_token_or_with_foreign_origin_is_forbidden(self):
165 body = {"action": "approve"}
166 self.assertEqual(self.post("/reports/demo/3", body, token=False)[0], 403)
167 self.assertEqual(self.post("/reports/demo/3", {**body, "token": "wrong"})[0], 403)
168 self.assertEqual(self.post("/reports/demo/3", body, headers={"Origin": "http://evil.example"})[0], 403)
169 self.assertEqual(self.post("/reports/demo/3", body, headers={"Origin": "null"})[0], 403)
170 self.assertEqual(self.post("/reports/demo/3", body, headers={"Sec-Fetch-Site": "cross-site"})[0], 403)
171 self.assertEqual(self.post("/reports/demo/3", body, login=False)[0], 403)
172 self.assertEqual(self.fake.calls, [])
173 ok = self.post("/reports/demo/3", body, headers={"Origin": f"http://127.0.0.1:{self.port}", "Sec-Fetch-Site": "same-origin"})
174 self.assertEqual(ok[0], 303)
175
176 def test_csrf_token_is_not_the_cookie_value(self):
177 self.assertNotEqual(self.server.csrf, self.server.session)
178 self.assertIn(self.server.csrf, self.get("/reports/demo/3")[2])
179 self.assertNotIn(self.server.session, self.get("/reports/demo/3")[2])
180 self.assertEqual(self.post("/reports/demo/3", {"action": "approve", "token": self.server.session})[0], 403)
181 self.fake.calls.clear()
182 self.assertEqual(self.get("/reports", headers={"Cookie": f"fwadmin={self.server.session}"})[0], 403) # not the per-port name
183 self.assertEqual(self.fake.calls, [])
184
185 def test_an_idle_connection_does_not_stall_others(self):
186 import socket
187 import time
188 idle = socket.create_connection(("127.0.0.1", self.port))
189 self.addCleanup(idle.close)
190 half = socket.create_connection(("127.0.0.1", self.port)) # headers promise a body that never comes
191 self.addCleanup(half.close)
192 half.sendall(f"POST /comments HTTP/1.0\r\nHost: 127.0.0.1:{self.port}\r\nContent-Length: 50\r\n\r\n".encode())
193 start = time.time()
194 self.assertEqual(self.get("/reports")[0], 200)
195 self.assertLess(time.time() - start, 2)
196
197 def test_security_headers(self):
198 _, h, _ = self.get("/reports")
199 self.assertIn("default-src 'none'", h["Content-Security-Policy"])
200 self.assertIn("form-action 'self'", h["Content-Security-Policy"])
201 self.assertIn("frame-ancestors 'none'", h["Content-Security-Policy"])
202 self.assertEqual((h["X-Content-Type-Options"], h["Referrer-Policy"], h["Cache-Control"]), ("nosniff", "no-referrer", "no-store"))
203
204 def test_unknown_route_is_404_and_no_javascript(self):
205 self.assertEqual(self.get("/nothing")[0], 404)
206 body = self.get("/reports")[2] + self.get("/comments")[2] + self.get("/reports/demo/3")[2] + self.get("/deploy")[2]
207 self.assertNotIn("<script", body)
208 self.assertNotIn("onclick", body)
209 self.assertIn("<title>fw admin</title>", body)
210
211 def test_style_is_served(self):
212 status, h, body = self.get("/style.css")
213 self.assertEqual((status, h["Content-Type"].split(";")[0]), (200, "text/css"))
214 self.assertIn(".badge", body)
215
216
217class Actions(Base):
218 def test_approve_runs_the_command_and_redirects(self):
219 status, h, _ = self.post("/reports/demo/3", {"action": "approve"})
220 self.assertEqual((status, h["Location"]), (303, "/reports/demo/3"))
221 self.assertEqual(self.fake.calls, [["issues", "approve", "demo", "3"]])
222
223 def test_close_reopen_and_mode(self):
224 self.post("/reports/demo/3", {"action": "close"})
225 self.post("/reports/demo/3", {"action": "reopen"})
226 self.assertEqual(self.fake.calls, [["issues", "close", "demo", "3"], ["issues", "reopen", "demo", "3"]])
227
228 def test_hide_needs_confirm(self):
229 status, _, body = self.post("/reports/demo/3", {"action": "hide"})
230 self.assertEqual(status, 200)
231 self.assertEqual(self.fake.calls, [])
232 self.assertIn('name="confirm" value="1"', body)
233 self.assertIn('name="action" value="hide"', body)
234 self.assertIn("Cancel", body)
235 self.assertEqual(self.post("/reports/demo/3", {"action": "hide", "confirm": "1"})[0], 303)
236 self.assertEqual(self.fake.calls, [["issues", "hide", "demo", "3"]])
237
238 def test_mode_change_needs_confirm(self):
239 status, _, body = self.post("/repos/demo/mode", {"mode": "public", "n": "3"})
240 self.assertEqual((status, self.fake.calls), (200, []))
241 self.assertIn('name="mode" value="public"', body)
242 status, h, _ = self.post("/repos/demo/mode", {"mode": "public", "n": "3", "confirm": "1"})
243 self.assertEqual((status, h["Location"]), (303, "/reports/demo/3"))
244 self.assertEqual(self.fake.calls, [["issues", "demo", "public"]])
245
246 def test_invalid_input_is_400_without_a_call(self):
247 for path, body in (("/reports/a;b/3", {"action": "approve"}), ("/reports/demo/1x", {"action": "approve"}),
248 ("/reports/demo/0", {"action": "approve"}), ("/reports/demo/3", {"action": "delete"}),
249 ("/reports/demo/3", {"action": "--json"}), ("/reports/demo.git/3", {"action": "approve"}),
250 ("/repos/demo/mode", {"mode": "open"}), ("/repos/list/mode", {"mode": "public", "confirm": "1"}),
251 ("/repos/demo/mode", {"mode": "public", "n": "x", "confirm": "1"}),
252 ("/comments", {"id": "../etc", "action": "approve"}), ("/comments", {"id": CID + "\n", "action": "approve"}), ("/comments", {"id": CID, "action": "nuke"})):
253 self.assertEqual(self.post(path, body)[0], 400, (path, body))
254 self.assertEqual(self.get("/reports/a;b/3")[0], 400)
255 self.assertEqual(self.get("/reports/demo/1x")[0], 400)
256 self.assertEqual(self.get("/reports/demo/1x/record.json")[0], 400)
257 self.assertEqual(self.get("/reports?show=bogus")[0], 400)
258 self.assertEqual(self.get("/reports?repo=a;b")[0], 400)
259 self.assertEqual(self.fake.calls, [])
260
261 def test_comment_actions(self):
262 self.post("/comments", {"id": CID, "action": "approve"})
263 self.post("/comments", {"id": CID, "action": "trust"})
264 self.assertEqual(self.fake.calls, [["comments", "approve", CID], ["comments", "approve", CID, "--trust"]])
265
266 def test_comment_reject_untrust_remove_need_confirm(self):
267 for action in ("reject", "untrust", "remove"):
268 self.fake.calls.clear()
269 status, _, body = self.post("/comments", {"id": CID, "action": action})
270 self.assertEqual((status, self.fake.calls), (200, []), action)
271 self.assertIn('name="confirm" value="1"', body)
272 self.assertEqual(self.post("/comments", {"id": CID, "action": action, "confirm": "1"})[0], 303)
273 self.assertEqual(self.fake.calls, [["comments", action, CID]])
274
275 def test_flash_shows_output_once_and_escapes_errors(self):
276 self.fake.replies[("issues", "approve", "demo", "3")] = "approve: demo#3 (public)"
277 self.post("/reports/demo/3", {"action": "approve"})
278 page = self.get("/reports/demo/3")[2]
279 self.assertIn("approve: demo#3 (public)", page)
280 self.assertNotIn("approve: demo#3 (public)", self.get("/reports/demo/3")[2])
281 self.fake.errors[("issues", "close", "demo", "3")] = f"boom {EVIL}"
282 self.post("/reports/demo/3", {"action": "close"})
283 page = self.get("/reports/demo/3")[2]
284 self.assertIn("boom &lt;script&gt;", page)
285 self.assertNotIn(EVIL, page)
286
287 def test_unexpected_error_in_a_post_still_redirects_with_a_flash(self):
288 def boom(args):
289 raise ValueError("<b>bad</b>")
290 self.fake.run = boom
291 status, h, _ = self.post("/reports/demo/3", {"action": "approve"})
292 self.assertEqual(status, 303)
293 self.assertIn("unexpected error: ValueError", self.get("/reports/demo/3")[2])
294 self.assertNotIn("<b>bad</b>", self.get("/reports/demo/3")[2])
295
296 def test_failed_listing_is_shown_not_crashed(self):
297 self.fake.errors[("issues", "list", "--json")] = "could not reach dh"
298 status, _, body = self.get("/reports")
299 self.assertEqual(status, 502)
300 self.assertIn("could not reach dh", body)
301
302
303class Pages(Base):
304 def test_reports_list(self):
305 body = self.get("/reports")[2]
306 for text in ("Crash", "demo", "reviewed", "1/2", "waiting", 'href="/reports/demo/3"', "Has replay"):
307 self.assertIn(text, body)
308 self.assertEqual(self.fake.calls, [["issues", "list", "--json"]])
309
310 def test_default_filter_is_waiting_else_all_and_filters_apply(self):
311 self.fake.replies[("issues", "list", "--json")] = issue_list(waiting=False, record=False)
312 self.assertIn("Crash", self.get("/reports")[2]) # nothing waiting: all
313 self.assertNotIn("Crash", self.get("/reports?show=waiting")[2])
314 self.assertNotIn("Crash", self.get("/reports?show=replay")[2])
315 self.assertIn("Crash", self.get("/reports?show=all")[2])
316
317 def test_report_page(self):
318 body = self.get("/reports/demo/3")[2]
319 for text in ("It crashed.", '<pre class="issue-text">', "Reporter A", "authenticated: yes", "not shown", "Commit", "abcdef1",
320 "Approve", "Hide", "Close", 'name="mode" value="reviewed" checked', "Download record", 'href="/reports/demo/2"'):
321 self.assertIn(text, body)
322 self.assertNotIn("Prepare dev replay", body) # no package.json in code_dir
323 self.assertEqual(self.fake.calls, [["issues", "show", "demo", "3", "--json"]])
324 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show(state="closed")
325 self.assertIn("Reopen", self.get("/reports/demo/3")[2])
326
327 def test_external_links_only_when_https(self):
328 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show(public=True, page_url="https://code.fuzzyworld.net/demo/issues/3/",
329 watch_url="javascript:alert(1)")
330 body = self.get("/reports/demo/3")[2]
331 self.assertIn('href="https://code.fuzzyworld.net/demo/issues/3/" rel="noreferrer noopener"', body)
332 self.assertNotIn("javascript:", body.replace("javascript:alert(1)</", ""))
333 self.assertNotIn('href="javascript', body)
334 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show(watch_url="https://apps.fuzzyworld.net/demo/replay.html?r=1&x=2")
335 body = self.get("/reports/demo/3")[2]
336 self.assertIn("Watch what happened", body)
337 self.assertIn("r=1&amp;x=2", body)
338
339 def test_hostile_text_is_escaped_everywhere(self):
340 self.fake.replies[("issues", "list", "--json")] = issue_list(title=EVIL)
341 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show(
342 title=EVIL, details=[[EVIL, EVIL]], messages=[{**issue_show()["messages"][0], "text": EVIL, "label": EVIL}])
343 self.fake.replies[("comments", "list", "--json")] = comment_list(text=EVIL, author=EVIL, site=EVIL, post_url="javascript:" + EVIL)
344 for path in ("/reports", "/reports/demo/3", "/comments", "/deploy"):
345 body = self.get(path)[2]
346 self.assertNotIn("<script>", body, path)
347 self.assertIn("&lt;script&gt;alert(1)&lt;/script&gt;", self.get("/reports/demo/3")[2])
348 self.assertNotIn('href="javascript', self.get("/comments")[2])
349
350 def test_deploy_page_is_read_only_and_escapes(self):
351 status, _, body = self.get("/deploy")
352 self.assertEqual(status, 200)
353 self.assertIn('aria-current=page>Deploy</a>', body)
354 self.assertNotIn("<form", body) # nothing to click yet
355 self.assertNotIn(EVIL, body)
356 self.assertIn("the &lt;b&gt;server&lt;/b&gt;", body)
357 self.assertIn("Fix &lt;i&gt;it&lt;/i&gt;", body)
358 self.assertIn("and 2 more", body) # 10 commits behind, 8 listed
359 self.assertIn("on mac-a", body)
360 self.assertIn("not on the server (unpushed commit, or unknown repo)", body)
361 self.assertIn("never deployed", body)
362 self.assertLess(body.index("&lt;script&gt;"), body.index("Zed")) # a pending app is listed before a current one
363 self.assertEqual(self.fake.calls[-1], ["deploys", "--json"])
364
365 def test_deploy_page_with_nothing_deployed(self):
366 self.fake.replies[("deploys", "--json")] = {"repos": [], "apps": []}
367 self.assertIn("No deployed apps.", self.get("/deploy")[2])
368
369 def test_comments_page(self):
370 body = self.get("/comments")[2]
371 for text in ("Nice post.", "Approve + trust", "Reject", "Untrust", "Recently approved", "Old one", "Remove",
372 'href="https://writing.fuzzyworld.net/2024/05/01/hello/" rel="noreferrer noopener"', "20241230T101010Z-123456"):
373 self.assertIn(text, body)
374 self.assertEqual(self.fake.calls, [["comments", "list", "--json"]])
375
376 def test_comments_buttons_depend_on_source_and_trust(self):
377 self.fake.replies[("comments", "list", "--json")] = comment_list(source="form", trusted=False)
378 pending = self.get("/comments")[2].split("Recently approved")[0]
379 self.assertNotIn("Approve + trust", pending)
380 self.assertNotIn("Untrust", pending)
381
382 def test_download_record(self):
383 self.fake.replies[("issues", "record", "demo", "3")] = '{"seed":7}'
384 status, h, body = self.get("/reports/demo/3/record.json")
385 self.assertEqual((status, body), (200, '{"seed":7}'))
386 self.assertIn("attachment", h["Content-Disposition"])
387 self.fake.errors[("issues", "record", "demo", "3")] = "no record for demo#3"
388 status, h, _ = self.get("/reports/demo/3/record.json")
389 self.assertEqual((status, h["Location"]), (303, "/reports/demo/3"))
390 self.assertIn("no record for demo#3", self.get("/reports/demo/3")[2])
391
392
393class Replay(Base):
394 def app(self, scripts=None):
395 os.makedirs(os.path.join(self.tmp, "demo"))
396 with open(os.path.join(self.tmp, "demo", "package.json"), "w") as f:
397 json.dump({"scripts": scripts if scripts is not None else {"report-replay": "node x.mjs"}}, f)
398
399 def test_button_only_with_the_script(self):
400 self.app({"dev": "vite"})
401 self.assertNotIn("Prepare dev replay", self.get("/reports/demo/3")[2])
402 shutil.rmtree(os.path.join(self.tmp, "demo"))
403 self.app()
404 self.assertIn("Prepare dev replay", self.get("/reports/demo/3")[2])
405
406 def test_replay_runs_npm_with_the_record_and_rules_commit(self):
407 self.app()
408 self.fake.replies[("issues", "record", "demo", "3")] = '{"seed":7}'
409 done = subprocess.CompletedProcess([], 0, stdout="open http://localhost:5173/?replay=/tmp/x/record.json&rules=abcdef1 now\n", stderr="")
410 with mock.patch.object(admin.subprocess, "run", return_value=done) as run:
411 self.assertEqual(self.post("/reports/demo/3", {"action": "replay"})[0], 303)
412 cmd = run.call_args.args[0]
413 self.assertEqual(cmd[:6], ["npm", "run", "--silent", "report-replay", "--", cmd[5]])
414 self.assertEqual(cmd[6:], ["--rules", "abcdef1"])
415 self.assertTrue(cmd[5].endswith("record.json"))
416 with open(cmd[5]) as f:
417 self.assertEqual(f.read(), '{"seed":7}')
418 shutil.rmtree(os.path.dirname(cmd[5]))
419 self.assertEqual(run.call_args.kwargs["cwd"], os.path.join(self.tmp, "demo"))
420 self.assertIn("fw-admin-demo-3-", cmd[5])
421 self.assertEqual(run.call_args.kwargs["errors"], "replace")
422 self.assertEqual(run.call_args.kwargs["timeout"], 60)
423 page = self.get("/reports/demo/3")[2]
424 self.assertIn('<a href="http://localhost:5173/?replay=/tmp/x/record.json&amp;rules=abcdef1">', page)
425 self.assertIn("needs `npm run dev` running in demo", page)
426
427 def test_replay_without_commit_or_with_a_bad_one(self):
428 self.app()
429 self.fake.replies[("issues", "show", "demo", "3", "--json")] = issue_show(details=[["Commit", "not-hex"]])
430 done = subprocess.CompletedProcess([], 0, stdout="ok", stderr="")
431 with mock.patch.object(admin.subprocess, "run", return_value=done) as run:
432 self.post("/reports/demo/3", {"action": "replay"})
433 shutil.rmtree(os.path.dirname(run.call_args.args[0][5]))
434 self.assertEqual(len(run.call_args.args[0]), 6)
435
436 def test_replay_without_the_script_is_an_error(self):
437 with mock.patch.object(admin.subprocess, "run") as run:
438 self.post("/reports/demo/3", {"action": "replay"})
439 run.assert_not_called()
440 self.assertIn("report-replay", self.get("/reports/demo/3")[2])
441
442 def test_code_dir_comes_from_the_environment(self):
443 with mock.patch.dict(os.environ, {"FW_CODE_DIR": "/somewhere/else"}):
444 s = admin.make_server(self.fake, 0)
445 self.addCleanup(s.server_close)
446 self.assertEqual(s.code_dir, "/somewhere/else")
447 s2 = admin.make_server(self.fake, 0, code_dir="/given")
448 self.addCleanup(s2.server_close)
449 self.assertEqual(s2.code_dir, "/given")
450
451
452class Ssh(unittest.TestCase):
453 def done(self, code=0, out="", err=""):
454 return subprocess.CompletedProcess([], code, stdout=out, stderr=err)
455
456 def test_exact_argv(self):
457 with mock.patch.object(admin.subprocess, "run", return_value=self.done(out="hi")) as run:
458 self.assertEqual(admin.SshRunner("dh").run(["issues", "show", "demo", "3", "--json"]), "hi")
459 self.assertEqual(run.call_args.args[0], ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "dh",
460 "~/bin/fw issues show demo 3 --json"])
461 kw = run.call_args.kwargs
462 self.assertEqual((kw["stdin"], kw["capture_output"], kw["text"], kw["timeout"]), (subprocess.DEVNULL, True, True, 60))
463 self.assertEqual((kw["encoding"], kw["errors"]), ("utf-8", "replace"))
464
465 def test_unsafe_arguments_are_refused_before_anything_is_sent(self):
466 r = admin.SshRunner("dh")
467 for bad in ("a;b", "$(id)", "-x", "--rm", "a b", "", "x\n", "a/b", "--json ", "`ls`"):
468 with mock.patch.object(admin.subprocess, "run") as run:
469 with self.assertRaises(admin.FwError):
470 r.run(["issues", bad])
471 run.assert_not_called()
472 with mock.patch.object(admin.subprocess, "run", return_value=self.done()) as run:
473 r.run(["comments", "approve", CID, "--trust"])
474 run.assert_called_once()
475
476 def test_host_is_checked(self):
477 for bad in ("-oProxyCommand=x", "a b", "a;b", "", "x@y"):
478 with self.assertRaises(admin.FwError):
479 admin.SshRunner(bad)
480 admin.SshRunner("my-host.example")
481
482 def test_timeout_unreachable_and_failure(self):
483 r = admin.SshRunner("dh", timeout=5)
484 with mock.patch.object(admin.subprocess, "run", side_effect=subprocess.TimeoutExpired("ssh", 5)):
485 with self.assertRaisesRegex(admin.FwError, "timed out after 5 s"):
486 r.run(["issues", "list"])
487 with mock.patch.object(admin.subprocess, "run", return_value=self.done(255, err="ssh: Could not resolve")):
488 with self.assertRaisesRegex(admin.FwError, r"could not reach dh \(try `ssh dh true`\)"):
489 r.run(["issues", "list"])
490 with mock.patch.object(admin.subprocess, "run", return_value=self.done(1, err="issues: no issue demo#9\n")):
491 with self.assertRaisesRegex(admin.FwError, "no issue demo#9"):
492 r.run(["issues", "list"])
493 with mock.patch.object(admin.subprocess, "run", side_effect=FileNotFoundError("ssh")):
494 with self.assertRaises(admin.FwError):
495 r.run(["issues", "list"])
496
497 def test_json(self):
498 r = admin.SshRunner("dh")
499 with mock.patch.object(admin.subprocess, "run", return_value=self.done(out='{"a": 1}')):
500 self.assertEqual(r.json(["issues", "list", "--json"]), {"a": 1})
501 for out in ("not json", "[1]"):
502 with mock.patch.object(admin.subprocess, "run", return_value=self.done(out=out)):
503 with self.assertRaises(admin.FwError):
504 r.json(["issues", "list", "--json"])
505
506
507class Cli(unittest.TestCase):
508 def test_main_prints_the_login_url_without_opening(self):
509 started = {}
510
511 def fake_serve(self):
512 started["url"] = self.login_url()
513
514 import io
515 out = io.StringIO()
516 with mock.patch.object(admin.AdminServer, "serve_forever", fake_serve), mock.patch.object(admin.webbrowser, "open") as opened, \
517 mock.patch("sys.stdout", out), mock.patch.dict(os.environ, {"FW_HOST": "dh"}):
518 self.assertEqual(admin.main(["--no-open", "--code-dir", "/x"]), 0)
519 opened.assert_not_called()
520 self.assertTrue(re.match(r"^http://127\.0\.0\.1:\d+/login\?k=[\w-]+$", started["url"]))
521 self.assertIn(started["url"], out.getvalue())
522
523 def test_bad_host_env_is_reported(self):
524 with mock.patch.dict(os.environ, {"FW_HOST": "-oEvil"}):
525 self.assertEqual(admin.main(["--no-open"]), 1)
526
527
528if __name__ == "__main__":
529 unittest.main()