1#!/usr/bin/env python3
2"""Unit tests for measure_cef_startup.py (stdlib only).
3
4Run headless: python3 -m unittest discover -s scripts -p 'test_measure_cef_startup.py'
5Set DUMBER_MEASURE_CUTOFF_SECONDS=0.2 for fast streaming tests (default 5.0).
6"""
7
8import os
9import subprocess
10import sys
11import tempfile
12import threading
13import unittest
14import urllib.error
15import urllib.request
16from http.server import ThreadingHTTPServer
17
18sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
19
20import measure_cef_startup as harness
21
22
23class BeaconTest(unittest.TestCase):
24 def setUp(self):
25 self.state = harness.FixtureState()
26 harness.Handler.state = self.state
27 harness.Handler.fixture = "static"
28 harness.Handler.run_token = "a" * 16
29 harness.Handler.nav_token = "b" * 16
30 self.server = ThreadingHTTPServer(("127.0.0.1", 0), harness.Handler)
31 self.port = self.server.server_address[1]
32 self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.05})
33 self.thread.daemon = True
34 self.thread.start()
35
36 def tearDown(self):
37 self.server.shutdown()
38 self.server.server_close()
39 self.thread.join(timeout=5)
40
41 def get(self, query):
42 url = "http://127.0.0.1:%d/__beacon?%s" % (self.port, query)
43 try:
44 with urllib.request.urlopen(url, timeout=5) as response:
45 return response.status
46 except urllib.error.HTTPError as exc:
47 exc.read()
48 return exc.code
49
50 def test_accepts_valid_beacon(self):
51 status = self.get("run=%s&nav=%s&event=%s&value=12" % ("a" * 16, "b" * 16, "first-contentful-paint"))
52 self.assertEqual(status, 204)
53 self.assertEqual(len(self.state.beacons), 1)
54 self.assertEqual(self.state.beacons[0]["value_ms"], 12)
55
56 def test_rejects_wrong_token(self):
57 status = self.get("run=%s&nav=%s&event=%s&value=12" % ("c" * 16, "d" * 16, "first-contentful-paint"))
58 self.assertEqual(status, 400)
59 self.assertEqual(self.state.beacons, [])
60
61 def test_rejects_malformed_token(self):
62 status = self.get("run=short&nav=%s&event=%s&value=12" % ("b" * 16, "first-contentful-paint"))
63 self.assertEqual(status, 400)
64 self.assertEqual(self.state.beacons, [])
65
66 def test_rejects_non_numeric_value(self):
67 status = self.get("run=%s&nav=%s&event=%s&value=abc" % ("a" * 16, "b" * 16, "first-contentful-paint"))
68 self.assertEqual(status, 400)
69 self.assertEqual(self.state.beacons, [])
70
71 def test_rejects_unknown_event(self):
72 status = self.get("run=%s&nav=%s&event=%s&value=1" % ("a" * 16, "b" * 16, "bogus"))
73 self.assertEqual(status, 400)
74 self.assertEqual(self.state.beacons, [])
75
76 def test_milestones_still_parsed(self):
77 lines = ['{"message":"startup_trace: milestone","milestone":"process_entry"}', "not json"]
78 self.assertEqual(len(harness.parse_child_output(lines)), 1)
79
80 def test_image_endpoint_serves_decodable_png(self):
81 import struct
82 import zlib
83 url = "http://127.0.0.1:%d/pixel.png" % self.port
84 with urllib.request.urlopen(url, timeout=5) as response:
85 self.assertEqual(response.status, 200)
86 self.assertEqual(response.headers.get_content_type(), "image/png")
87 body = response.read()
88 self.assertEqual(body[:8], bytes.fromhex("89504e470d0a1a0a"))
89 width, height = struct.unpack(">II", body[16:24])
90 self.assertEqual((width, height), (1, 1))
91 idat_len = struct.unpack(">I", body[33:37])[0]
92 self.assertEqual(body[37:41], b"IDAT")
93 zlib.decompress(body[41:41 + idat_len])
94 self.assertEqual(body[-8:-4], b"IEND")
95
96
97class CliValidationTest(unittest.TestCase):
98 def setUp(self):
99 self.script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "measure_cef_startup.py")
100 self.temp = tempfile.mkdtemp()
101 self.binary = os.path.join(self.temp, "fakebin")
102 with open(self.binary, "w") as handle:
103 handle.write("#!/bin/sh\nexit 0\n")
104 os.chmod(self.binary, 0o755)
105 self.cef_dir = os.path.join(self.temp, "cef")
106 os.makedirs(self.cef_dir)
107
108 def test_rejects_existing_output(self):
109 existing = os.path.join(self.temp, "exists")
110 os.makedirs(existing)
111 result = subprocess.run(
112 [sys.executable, self.script, "--binary", self.binary, "--cef-dir", self.cef_dir,
113 "--scenario", "profile-fresh", "--fixture", "static",
114 "--runs", "1", "--output", existing],
115 capture_output=True, text=True)
116 self.assertNotEqual(result.returncode, 0)
117 self.assertIn("fresh absolute directory", result.stderr)
118
119 def test_rejects_relative_output(self):
120 result = subprocess.run(
121 [sys.executable, self.script, "--binary", self.binary, "--cef-dir", self.cef_dir,
122 "--scenario", "profile-fresh", "--fixture", "static",
123 "--runs", "1", "--output", "relative/path"],
124 capture_output=True, text=True)
125 self.assertNotEqual(result.returncode, 0)
126
127 def test_rejects_missing_binary(self):
128 result = subprocess.run(
129 [sys.executable, self.script, "--binary", os.path.join(self.temp, "missing"),
130 "--cef-dir", self.cef_dir, "--scenario", "profile-fresh", "--fixture", "static",
131 "--runs", "1", "--output", os.path.join(self.temp, "out")],
132 capture_output=True, text=True)
133 self.assertNotEqual(result.returncode, 0)
134
135 def test_rejects_bad_runs(self):
136 result = subprocess.run(
137 [sys.executable, self.script, "--binary", self.binary, "--cef-dir", self.cef_dir,
138 "--scenario", "profile-fresh", "--fixture", "static",
139 "--runs", "0", "--output", os.path.join(self.temp, "out2")],
140 capture_output=True, text=True)
141 self.assertNotEqual(result.returncode, 0)
142
143 def test_rejects_bad_idle_timeout(self):
144 for bad in ("-1", "300001"):
145 result = subprocess.run(
146 [sys.executable, self.script, "--binary", self.binary, "--cef-dir", self.cef_dir,
147 "--scenario", "profile-fresh", "--fixture", "static",
148 "--runs", "1", "--idle-timeout-ms", bad,
149 "--output", os.path.join(self.temp, "out-idle-" + bad)],
150 capture_output=True, text=True)
151 self.assertNotEqual(result.returncode, 0)
152 self.assertIn("idle-timeout-ms", result.stderr)
153
154
155class FakeBinaryLifetimeTest(unittest.TestCase):
156 def setUp(self):
157 self.temp = tempfile.mkdtemp()
158 self.cef_dir = os.path.join(self.temp, "cef")
159 os.makedirs(self.cef_dir)
160 self.counter = 0
161
162 def make_binary(self, body):
163 self.counter += 1
164 path = os.path.join(self.temp, "fake-%d" % self.counter)
165 with open(path, "w") as handle:
166 handle.write("#!/bin/sh\n" + body + "\n")
167 os.chmod(path, 0o755)
168 return path
169
170 def make_fetch_binary(self):
171 # Fetches the fixture URL passed as argv[2] (argv: browse <url>),
172 # then exits: exercises document counting without a real browser.
173 path = os.path.join(self.temp, "fetch-%d" % self.counter)
174 self.counter += 1
175 with open(path, "w") as handle:
176 handle.write("#!/usr/bin/env python3\nimport sys, urllib.request\n"
177 "urllib.request.urlopen(sys.argv[2], timeout=10).read()\n")
178 os.chmod(path, 0o755)
179 return path
180
181 def test_duplicate_documents_marked_incomplete(self):
182 binary = self.make_binary("exit 0")
183 result = harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
184 self.assertFalse(result["complete"])
185 self.assertIsNone(result["target_fcp_ms"])
186
187 def test_absent_fcp_is_incomplete(self):
188 binary = self.make_fetch_binary()
189 result = harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
190 self.assertEqual(result["document_requests"], 1)
191 self.assertIsNone(result["target_fcp_ms"])
192 self.assertFalse(result["complete"])
193 # Same-clock spawn-to-request latency is still reported.
194 self.assertIsNotNone(result["first_document_observation_ms"])
195 self.assertGreaterEqual(result["first_document_observation_ms"], 0)
196
197 def test_no_navigation_means_no_document_observation(self):
198 binary = self.make_binary("exit 0")
199 result = harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
200 self.assertEqual(result["document_requests"], 0)
201 self.assertIsNone(result["first_document_observation_ms"])
202
203 def test_relay_fallback_recorded_without_owner(self):
204 binary = self.make_binary("exit 0")
205 result = harness.run_sample(binary, self.cef_dir, "relay-window", "static", 1)
206 self.assertTrue(result["fallback_spawn"])
207 self.assertFalse(result["relayed"])
208
209 def test_second_window_uses_observation_not_process_summary(self):
210 binary = self.make_binary("exit 0")
211 first = harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
212 second = harness.run_sample(binary, self.cef_dir, "relay-window", "static", 2)
213 self.assertIn("observation_spawn_to_summary_ms", first)
214 self.assertIn("observation_spawn_to_summary_ms", second)
215 self.assertNotIn("total_ms", second)
216
217 def test_run_dir_cleaned_up(self):
218 before = set(os.listdir(tempfile.gettempdir()))
219 binary = self.make_binary("exit 0")
220 harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
221 after = set(os.listdir(tempfile.gettempdir()))
222 leaked = [name for name in (after - before) if name.startswith("cef-startup-run-")]
223 self.assertEqual(leaked, [])
224
225 def test_streaming_returns_without_process_exit(self):
226 # A hanging GUI must not pin the harness: with a short cutoff the
227 # sample decides from streamed output, then terminates the child.
228 binary = self.make_binary("exec sleep 30")
229 result = harness.run_sample(binary, self.cef_dir, "profile-fresh", "static", 1)
230 self.assertFalse(result["complete"])
231 self.assertLess(result["observation_spawn_to_summary_ms"], 15000)
232
233
234class ResidencyContractTest(unittest.TestCase):
235 def setUp(self):
236 self.temp = tempfile.mkdtemp()
237 self.cef_dir = os.path.join(self.temp, "cef")
238 os.makedirs(self.cef_dir)
239 self.dirs = {
240 "config": os.path.join(self.temp, "config"),
241 "data": os.path.join(self.temp, "data"),
242 "state": os.path.join(self.temp, "state"),
243 "cache": os.path.join(self.temp, "cache"),
244 "runtime": os.path.join(self.temp, "runtime"),
245 "root_cache": os.path.join(self.temp, "cef-root-cache"),
246 }
247
248 def test_scenario_env_forces_json_debug_config(self):
249 env = harness.scenario_env(self.cef_dir, self.dirs, idle_timeout_ms=120000)
250 self.assertEqual(env["DUMBER_CEF_DIR"], self.cef_dir)
251 with open(os.path.join(self.dirs["config"], "dumber", "config.toml")) as handle:
252 content = handle.read()
253 self.assertIn('format = "json"', content)
254 self.assertIn('level = "debug"', content)
255 self.assertIn(self.cef_dir, content)
256 self.assertIn("idle_runtime_timeout_ms = 120000", content)
257
258 def test_scenario_env_defaults_to_no_residency(self):
259 harness.scenario_env(self.cef_dir, self.dirs)
260 with open(os.path.join(self.dirs["config"], "dumber", "config.toml")) as handle:
261 content = handle.read()
262 self.assertIn("idle_runtime_timeout_ms = 0", content)
263
264 def test_scenario_env_passes_display_vars_only(self):
265 os.environ["DUMBER_MEASURE_TEST_MARKER"] = "must-not-propagate"
266 try:
267 env = harness.scenario_env(self.cef_dir, self.dirs)
268 finally:
269 del os.environ["DUMBER_MEASURE_TEST_MARKER"]
270 self.assertNotIn("DUMBER_MEASURE_TEST_MARKER", env)
271 for name in harness.DISPLAY_PASSTHROUGH_VARS:
272 if os.environ.get(name):
273 self.assertEqual(env[name], os.environ[name])
274
275 def test_wayland_absolute_display_needs_nothing(self):
276 previous = os.environ.get("WAYLAND_DISPLAY")
277 runtime = os.environ.get("XDG_RUNTIME_DIR", "/run/user/%d" % os.getuid())
278 display = previous or "wayland-1"
279 absolute = display if "/" in display else os.path.join(runtime, os.path.basename(display))
280 env = {"WAYLAND_DISPLAY": absolute}
281 os.makedirs(self.dirs["runtime"], exist_ok=True)
282 self.assertTrue(harness.ensure_wayland_socket(self.dirs["runtime"], env))
283 self.assertEqual(env["WAYLAND_DISPLAY"], absolute)
284 self.assertEqual(os.listdir(self.dirs["runtime"]), [])
285
286 def test_wayland_relative_display_resolves_to_absolute(self):
287 import socket as stdlib_socket
288 fake_runtime = os.path.join(self.temp, "real-runtime")
289 os.makedirs(fake_runtime)
290 sock = stdlib_socket.socket(stdlib_socket.AF_UNIX, stdlib_socket.SOCK_STREAM)
291 sock.bind(os.path.join(fake_runtime, "wayland-99"))
292 previous_runtime = os.environ.get("XDG_RUNTIME_DIR")
293 os.environ["XDG_RUNTIME_DIR"] = fake_runtime
294 env = {"WAYLAND_DISPLAY": "wayland-99"}
295 try:
296 self.assertTrue(harness.ensure_wayland_socket(self.dirs["runtime"], env))
297 self.assertEqual(env["WAYLAND_DISPLAY"],
298 os.path.join(fake_runtime, "wayland-99"))
299 # No symlink: deep isolated paths exceed the unix-socket limit.
300 self.assertNotIn("wayland-99", os.listdir(self.dirs["runtime"]))
301 finally:
302 if previous_runtime is None:
303 os.environ.pop("XDG_RUNTIME_DIR", None)
304 else:
305 os.environ["XDG_RUNTIME_DIR"] = previous_runtime
306 sock.close()
307
308 def test_wayland_missing_socket_reports_false(self):
309 previous_runtime = os.environ.get("XDG_RUNTIME_DIR")
310 os.environ["XDG_RUNTIME_DIR"] = os.path.join(self.temp, "empty-runtime")
311 env = {"WAYLAND_DISPLAY": "wayland-does-not-exist"}
312 try:
313 self.assertFalse(harness.ensure_wayland_socket(self.dirs["runtime"], env))
314 finally:
315 if previous_runtime is None:
316 os.environ.pop("XDG_RUNTIME_DIR", None)
317 else:
318 os.environ["XDG_RUNTIME_DIR"] = previous_runtime
319 self.assertEqual(env["WAYLAND_DISPLAY"], "wayland-does-not-exist")
320
321 def test_count_cef_inits(self):
322 init = '{"message":"startup_trace: milestone","milestone":"cef_initialized"}'
323 other = '{"message":"startup_trace: milestone","milestone":"process_entry"}'
324 noise = '{"message":"cef: InitWithApp returned OK"}'
325 self.assertEqual(harness.count_cef_inits([]), 0)
326 self.assertEqual(harness.count_cef_inits([other, "not json"]), 0)
327 self.assertEqual(harness.count_cef_inits([other, init]), 1)
328 self.assertEqual(harness.count_cef_inits([init, noise, init]), 2)
329
330 def test_second_window_records_owner_init_count(self):
331 binary = os.path.join(self.temp, "fakebin")
332 with open(binary, "w") as handle:
333 handle.write("#!/bin/sh\nexit 0\n")
334 os.chmod(binary, 0o755)
335 log_path = os.path.join(self.temp, "owner.log")
336 with open(log_path, "w") as handle:
337 handle.write('{"message":"startup_trace: milestone","milestone":"cef_initialized"}\n')
338 owner_dirs = dict(self.dirs)
339 owner_dirs["owner_log"] = log_path
340 result = harness.run_sample(binary, self.cef_dir, "relay-window", "static", 1,
341 owner=None, owner_dirs=owner_dirs, owner_socket=None)
342 self.assertTrue(result["fallback_spawn"])
343 self.assertEqual(result["owner_cef_init_count"], 1)
344
345 def test_second_window_missing_owner_log_is_none(self):
346 binary = os.path.join(self.temp, "fakebin")
347 with open(binary, "w") as handle:
348 handle.write("#!/bin/sh\nexit 0\n")
349 os.chmod(binary, 0o755)
350 owner_dirs = dict(self.dirs)
351 owner_dirs["owner_log"] = os.path.join(self.temp, "absent.log")
352 result = harness.run_sample(binary, self.cef_dir, "relay-window", "static", 1,
353 owner=None, owner_dirs=owner_dirs, owner_socket=None)
354 self.assertIsNone(result["owner_cef_init_count"])
355
356
357class DocumentTokenTest(unittest.TestCase):
358 # Document routes must only record and serve correlated requests;
359 # beacons already validate tokens, subresources stay ungated.
360 def setUp(self):
361 self.state = harness.FixtureState()
362 cls = harness.make_sample_handler(self.state, "static", "a" * 16, "b" * 16)
363 self.server = ThreadingHTTPServer(("127.0.0.1", 0), cls)
364 self.port = self.server.server_address[1]
365 self.thread = threading.Thread(target=self.server.serve_forever, kwargs={"poll_interval": 0.05})
366 self.thread.daemon = True
367 self.thread.start()
368
369 def tearDown(self):
370 self.server.shutdown()
371 self.server.server_close()
372 self.thread.join(timeout=5)
373
374 def get(self, path):
375 url = "http://127.0.0.1:%d%s" % (self.port, path)
376 try:
377 with urllib.request.urlopen(url, timeout=5) as response:
378 return response.status, response.read()
379 except urllib.error.HTTPError as exc:
380 exc.read()
381 return exc.code, b""
382
383 def test_correlated_document_recorded_and_served(self):
384 status, body = self.get("/?run=%s&nav=%s" % ("a" * 16, "b" * 16))
385 self.assertEqual(status, 200)
386 self.assertIn(b"static", body)
387 self.assertEqual(len(self.state.document_requests), 1)
388 self.assertIsNotNone(self.state.first_document_monotonic)
389
390 def test_wrong_tokens_rejected_without_recording(self):
391 status, _ = self.get("/?run=%s&nav=%s" % ("c" * 16, "d" * 16))
392 self.assertEqual(status, 404)
393 self.assertEqual(self.state.document_requests, [])
394 self.assertIsNone(self.state.first_document_monotonic)
395
396 def test_missing_tokens_rejected_without_recording(self):
397 status, _ = self.get("/")
398 self.assertEqual(status, 404)
399 self.assertEqual(self.state.document_requests, [])
400
401 def test_redirect_with_bad_tokens_rejected(self):
402 status, _ = self.get("/redirect?run=%s&nav=%s" % ("c" * 16, "d" * 16))
403 self.assertEqual(status, 404)
404 self.assertEqual(self.state.document_requests, [])
405
406 def test_redirect_with_tokens_redirects_once(self):
407 opener = urllib.request.build_opener(NoRedirect())
408 url = "http://127.0.0.1:%d/redirect?run=%s&nav=%s" % (self.port, "a" * 16, "b" * 16)
409 try:
410 opener.open(url, timeout=5)
411 self.fail("expected redirect")
412 except Redirected as exc:
413 self.assertEqual(exc.code, 302)
414 self.assertIn("/target?run=%s" % ("a" * 16), exc.headers["Location"])
415 self.assertEqual(len(self.state.document_requests), 1)
416
417 def test_handler_state_is_per_sample(self):
418 other_state = harness.FixtureState()
419 other_cls = harness.make_sample_handler(other_state, "static", "c" * 16, "d" * 16)
420 self.assertIsNot(other_cls.state, self.state)
421 self.assertNotEqual(other_cls.run_token, "a" * 16)
422
423
424class NoRedirect(urllib.request.HTTPRedirectHandler):
425 def redirect_request(self, req, fp, code, msg, headers, newurl):
426 raise Redirected(code, headers)
427
428
429class Redirected(Exception):
430 def __init__(self, code, headers):
431 super().__init__(code)
432 self.code = code
433 self.headers = headers
434
435
436class ResidencyValidityTest(unittest.TestCase):
437 def base_result(self):
438 return {"relayed": True, "fallback_spawn": False, "launcher_returncode": 0,
439 "owner_cef_init_count": 1, "launcher_cef_init_count": 0}
440
441 def test_non_resident_scenario_is_none(self):
442 self.assertIsNone(harness.is_residency_valid(self.base_result(), "profile-fresh"))
443
444 def test_valid_relay(self):
445 self.assertTrue(harness.is_residency_valid(self.base_result(), "relay-window"))
446
447 def test_fallback_is_invalid(self):
448 result = self.base_result()
449 result.update({"relayed": False, "fallback_spawn": True})
450 self.assertFalse(harness.is_residency_valid(result, "relay-window"))
451
452 def test_launcher_failure_is_invalid(self):
453 result = self.base_result()
454 result["launcher_returncode"] = 1
455 self.assertFalse(harness.is_residency_valid(result, "relay-window"))
456
457 def test_launcher_cold_start_is_invalid(self):
458 result = self.base_result()
459 result["launcher_cef_init_count"] = 1
460 self.assertFalse(harness.is_residency_valid(result, "relay-window"))
461
462 def test_missing_owner_init_is_invalid(self):
463 result = self.base_result()
464 result["owner_cef_init_count"] = None
465 self.assertFalse(harness.is_residency_valid(result, "relay-window"))
466
467 def test_reopen_requires_settled_close_and_same_process(self):
468 valid = self.base_result()
469 valid["reopen"] = {"close_acknowledged": True, "quiescence_settled": True,
470 "same_process": True, "owner_alive_after": True}
471 self.assertTrue(harness.is_residency_valid(valid, "reopen-window"))
472 for key in ("close_acknowledged", "quiescence_settled", "same_process", "owner_alive_after"):
473 broken = self.base_result()
474 reopen = {"close_acknowledged": True, "quiescence_settled": True,
475 "same_process": True, "owner_alive_after": True}
476 reopen[key] = False
477 broken["reopen"] = reopen
478 self.assertFalse(harness.is_residency_valid(broken, "reopen-window"), key)
479 self.assertFalse(harness.is_residency_valid(self.base_result(), "reopen-window"))
480
481 def test_toml_escaping_round_trips(self):
482 try:
483 import tomllib
484 except ImportError:
485 self.skipTest("tomllib unavailable")
486 nasty = "C:\\we\"ird\npath\twith\x01controls\x7f"
487 escaped = harness._escape_toml_basic(nasty)
488 parsed = tomllib.loads('key = "%s"' % escaped)
489 self.assertEqual(parsed["key"], nasty)
490
491
492class LateOwnerRequestTest(unittest.TestCase):
493 # Finding 2: a relay launcher may exit right after forwarding while
494 # the owner requests the document later. Collection must survive the
495 # launcher exit until the bounded navigation deadline.
496 def setUp(self):
497 self.temp = tempfile.mkdtemp()
498 self.cef_dir = os.path.join(self.temp, "cef")
499 os.makedirs(self.cef_dir)
500 self.saved = {key: os.environ.get(key) for key in
501 ("DUMBER_MEASURE_CUTOFF_SECONDS", "DUMBER_MEASURE_NAV_DEADLINE_SECONDS")}
502 os.environ["DUMBER_MEASURE_CUTOFF_SECONDS"] = "0.2"
503 os.environ["DUMBER_MEASURE_NAV_DEADLINE_SECONDS"] = "4"
504 self.owner = subprocess.Popen(["sleep", "30"])
505 self.addCleanup(self.owner.kill)
506
507 def tearDown(self):
508 for key, value in self.saved.items():
509 if value is None:
510 os.environ.pop(key, None)
511 else:
512 os.environ[key] = value
513 try:
514 self.owner.wait(timeout=5)
515 except subprocess.TimeoutExpired:
516 pass
517
518 def test_late_document_survives_launcher_exit(self):
519 binary = os.path.join(self.temp, "late-launcher")
520 with open(binary, "w") as handle:
521 handle.write("#!/bin/sh\n"
522 'url="$2"\n'
523 '(sleep 1; python3 -c "import sys,urllib.request;'
524 ' urllib.request.urlopen(sys.argv[1], timeout=10).read()" "$url") >/dev/null 2>&1 &\n'
525 "exit 0\n")
526 os.chmod(binary, 0o755)
527 owner_dirs = {key: os.path.join(self.temp, key)
528 for key in ("config", "data", "state", "cache", "runtime", "root_cache")}
529 owner_dirs["owner_log"] = os.path.join(self.temp, "owner.log")
530 with open(owner_dirs["owner_log"], "w") as handle:
531 handle.write('{"message":"startup_trace: milestone","milestone":"cef_initialized"}\n')
532 result = harness.run_sample(binary, self.cef_dir, "relay-window", "static", 1,
533 owner=self.owner, owner_dirs=owner_dirs, owner_socket=None)
534 self.assertEqual(result["document_requests"], 1)
535 self.assertIsNotNone(result["first_document_observation_ms"])
536 self.assertGreaterEqual(result["first_document_observation_ms"], 900)
537
538
539class ReopenWindowTest(unittest.TestCase):
540 def setUp(self):
541 self.temp = tempfile.mkdtemp()
542 self.cef_dir = os.path.join(self.temp, "cef")
543 os.makedirs(self.cef_dir)
544
545 def test_diagnostic_close_acknowledged(self):
546 import json
547 import socket
548 socket_path = os.path.join(self.temp, "test.sock")
549 listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
550 listener.bind(socket_path)
551 listener.listen(1)
552 listener.settimeout(5)
553
554 def serve_once():
555 conn, _ = listener.accept()
556 with conn:
557 conn.settimeout(5)
558 data = b""
559 while b"\n" not in data:
560 chunk = conn.recv(4096)
561 if not chunk:
562 break
563 data += chunk
564 request = json.loads(data.decode())
565 assert request["action"] == "close-all-windows"
566 conn.sendall(json.dumps({"request_id": request["request_id"],
567 "accepted": True}).encode() + b"\n")
568
569 thread = threading.Thread(target=serve_once)
570 thread.daemon = True
571 thread.start()
572 try:
573 self.assertTrue(harness.send_diagnostic_close(socket_path))
574 finally:
575 thread.join(timeout=5)
576 listener.close()
577
578 def test_diagnostic_close_refused_without_server(self):
579 self.assertFalse(harness.send_diagnostic_close(os.path.join(self.temp, "missing.sock"), 1.0))
580
581 def test_proc_helpers_observe_current_process(self):
582 self.assertIsInstance(harness.child_pids(os.getpid()), list)
583 rss = harness.proc_rss_kb(os.getpid())
584 self.assertIsNotNone(rss)
585 self.assertGreater(rss, 0)
586 self.assertIsNone(harness.proc_rss_kb(2 ** 30))
587
588 def test_proc_starttime_identifies_current_process(self):
589 first = harness.proc_starttime(os.getpid())
590 self.assertIsNotNone(first)
591 self.assertGreaterEqual(first, 0)
592 self.assertEqual(first, harness.proc_starttime(os.getpid()))
593 self.assertIsNone(harness.proc_starttime(2 ** 30))
594
595 def test_child_quiescence_returns_stable_count(self):
596 count, settled = harness.wait_for_child_quiescence(os.getpid(), settle_seconds=0.2, timeout_seconds=5.0)
597 self.assertIsInstance(count, int)
598 self.assertGreaterEqual(count, 0)
599 self.assertTrue(settled)
600
601 def test_child_quiescence_timeout_reports_unsettled(self):
602 count, settled = harness.wait_for_child_quiescence(
603 os.getpid(), settle_seconds=60.0, timeout_seconds=0.0)
604 self.assertIsInstance(count, int)
605 self.assertFalse(settled)
606
607 def test_reopen_without_owner_falls_back(self):
608 binary = os.path.join(self.temp, "fakebin")
609 with open(binary, "w") as handle:
610 handle.write("#!/bin/sh\nexit 0\n")
611 os.chmod(binary, 0o755)
612 result = harness.run_sample(binary, self.cef_dir, "reopen-window", "static", 1,
613 owner=None, owner_dirs=None, owner_socket=None)
614 self.assertTrue(result["fallback_spawn"])
615 self.assertFalse(result["relayed"])
616 self.assertNotIn("reopen", result)
617
618
619if __name__ == "__main__":
620 unittest.main()