448 lines
18 KiB
Python
448 lines
18 KiB
Python
"""Read-only-ish timing probe for the desktop update commit sequence.
|
|
|
|
The probe imports production code and replaces only its external download/apply
|
|
edges in memory. It does not modify production sources or existing tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import gc
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import weakref
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
|
|
|
|
from PySide6.QtCore import QCoreApplication, QEvent, QObject, QThreadPool, QTimer
|
|
from PySide6.QtWidgets import QApplication
|
|
|
|
from doctor_workstation.services import app_update as update_service
|
|
from doctor_workstation.services.app_update import (
|
|
PACKAGE_TYPE_INNO_SETUP,
|
|
UpdateOffer,
|
|
UpdatePackage,
|
|
)
|
|
from doctor_workstation.ui.dialogs import app_update as update_ui
|
|
|
|
|
|
def _record(events: list[dict[str, Any]], name: str, started: float) -> None:
|
|
events.append(
|
|
{
|
|
"event": name,
|
|
"elapsed_ms": round((time.perf_counter() - started) * 1000, 3),
|
|
"thread_id": threading.get_ident(),
|
|
}
|
|
)
|
|
|
|
|
|
def exercise_session(iterations: int = 25) -> dict[str, Any]:
|
|
app = QApplication.instance() or QApplication([])
|
|
main_thread = threading.get_ident()
|
|
original_edges = {
|
|
"is_frozen_install": update_ui.is_frozen_install,
|
|
"frozen_install_root": update_ui.frozen_install_root,
|
|
"download_package": update_ui.download_package,
|
|
"apply_downloaded_update": update_ui.apply_downloaded_update,
|
|
}
|
|
failures: list[dict[str, Any]] = []
|
|
samples: list[list[dict[str, Any]]] = []
|
|
collected_signals = 0
|
|
|
|
try:
|
|
with tempfile.TemporaryDirectory(prefix="zyt-update-commit-") as raw_tmp:
|
|
temp_root = Path(raw_tmp)
|
|
install_root = temp_root / "installed"
|
|
install_root.mkdir()
|
|
(install_root / "DoctorWorkstation.exe").write_bytes(b"MZ")
|
|
update_ui.is_frozen_install = lambda: True
|
|
update_ui.frozen_install_root = lambda: install_root
|
|
|
|
for index in range(iterations):
|
|
run_root = temp_root / f"run-{index}"
|
|
run_root.mkdir()
|
|
events: list[dict[str, Any]] = []
|
|
started = time.perf_counter()
|
|
|
|
def fake_download(
|
|
_url: str,
|
|
destination: Path,
|
|
*,
|
|
_events: list[dict[str, Any]] = events,
|
|
_started: float = started,
|
|
**kwargs: Any,
|
|
) -> Path:
|
|
_record(_events, "download_enter", _started)
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
destination.write_bytes(b"MZ" + b"probe")
|
|
progress = kwargs.get("progress")
|
|
if callable(progress):
|
|
progress(7, 7)
|
|
_record(_events, "download_return", _started)
|
|
return destination
|
|
|
|
def fake_apply(
|
|
_payload: Path,
|
|
*,
|
|
package_type: str,
|
|
_events: list[dict[str, Any]] = events,
|
|
_started: float = started,
|
|
) -> None:
|
|
assert package_type == PACKAGE_TYPE_INNO_SETUP
|
|
_record(_events, "apply_enter", _started)
|
|
_record(_events, "apply_return", _started)
|
|
|
|
update_ui.download_package = fake_download
|
|
update_ui.apply_downloaded_update = fake_apply
|
|
|
|
host = QObject()
|
|
host.config = SimpleNamespace( # type: ignore[attr-defined]
|
|
config_dir=run_root,
|
|
verify_ssl=True,
|
|
)
|
|
|
|
def request_quit(
|
|
_events: list[dict[str, Any]] = events,
|
|
_started: float = started,
|
|
) -> None:
|
|
_record(_events, "request_quit", _started)
|
|
|
|
host.request_quit = request_quit # type: ignore[attr-defined]
|
|
session = update_ui.AppUpdateSession(host)
|
|
offer = UpdateOffer(
|
|
has_update=True,
|
|
force=True,
|
|
enabled=True,
|
|
current_version="1.0.0",
|
|
latest_version=f"1.0.{index + 1}",
|
|
min_version="",
|
|
title="probe",
|
|
notes="probe",
|
|
platform="windows",
|
|
arch="x64",
|
|
package=UpdatePackage(
|
|
url="https://example.invalid/DoctorWorkstation-Setup.exe",
|
|
sha256="a" * 64,
|
|
size=7,
|
|
filename="DoctorWorkstation-Setup.exe",
|
|
type=PACKAGE_TYPE_INNO_SETUP,
|
|
),
|
|
can_install=True,
|
|
)
|
|
dialog = update_ui.AppUpdateDialog(offer)
|
|
session.dialog = dialog
|
|
|
|
original_finish = session._finish_install
|
|
original_finished = session._on_install_finished
|
|
|
|
def finish_probe(
|
|
*args: Any,
|
|
_events: list[dict[str, Any]] = events,
|
|
_started: float = started,
|
|
_original: Any = original_finish,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
_record(_events, "result_slot_enter", _started)
|
|
_original(*args, **kwargs)
|
|
_record(_events, "result_slot_return", _started)
|
|
|
|
def finished_probe(
|
|
*args: Any,
|
|
_events: list[dict[str, Any]] = events,
|
|
_started: float = started,
|
|
_original: Any = original_finished,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
_record(_events, "finished_slot_enter", _started)
|
|
_original(*args, **kwargs)
|
|
_record(_events, "finished_slot_return", _started)
|
|
|
|
session._finish_install = finish_probe # type: ignore[method-assign]
|
|
session._on_install_finished = finished_probe # type: ignore[method-assign]
|
|
session._start_install(dialog, offer)
|
|
signal_ref = weakref.ref(session._active_install_signals)
|
|
|
|
deadline = time.perf_counter() + 3.0
|
|
while time.perf_counter() < deadline:
|
|
app.processEvents()
|
|
if any(item["event"] == "request_quit" for item in events):
|
|
break
|
|
time.sleep(0.001)
|
|
QThreadPool.globalInstance().waitForDone(3000)
|
|
app.processEvents()
|
|
|
|
names = [item["event"] for item in events]
|
|
expected = [
|
|
"download_enter",
|
|
"download_return",
|
|
"result_slot_enter",
|
|
"apply_enter",
|
|
"apply_return",
|
|
"result_slot_return",
|
|
"finished_slot_enter",
|
|
"request_quit",
|
|
"finished_slot_return",
|
|
]
|
|
slot_threads = {
|
|
item["thread_id"]
|
|
for item in events
|
|
if item["event"] in {"result_slot_enter", "finished_slot_enter", "request_quit"}
|
|
}
|
|
if names != expected or slot_threads != {main_thread}:
|
|
failures.append(
|
|
{
|
|
"iteration": index,
|
|
"events": events,
|
|
"main_thread_id": main_thread,
|
|
}
|
|
)
|
|
if index < 3:
|
|
samples.append(events)
|
|
|
|
session.deleteLater()
|
|
dialog.deleteLater()
|
|
del session, dialog, host
|
|
QCoreApplication.sendPostedEvents(None, QEvent.Type.DeferredDelete)
|
|
app.processEvents()
|
|
gc.collect()
|
|
app.processEvents()
|
|
if signal_ref() is None:
|
|
collected_signals += 1
|
|
finally:
|
|
update_ui.is_frozen_install = original_edges["is_frozen_install"]
|
|
update_ui.frozen_install_root = original_edges["frozen_install_root"]
|
|
update_ui.download_package = original_edges["download_package"]
|
|
update_ui.apply_downloaded_update = original_edges["apply_downloaded_update"]
|
|
|
|
return {
|
|
"iterations": iterations,
|
|
"failures": failures,
|
|
"signals_collected_after_iteration": collected_signals,
|
|
"main_thread_id": main_thread,
|
|
"samples": samples,
|
|
}
|
|
|
|
|
|
def exercise_real_popen() -> dict[str, Any]:
|
|
if sys.platform != "win32":
|
|
return {"skipped": f"requires win32, got {sys.platform}"}
|
|
with tempfile.TemporaryDirectory(prefix="zyt-update-helper-") as raw_tmp:
|
|
temp_root = Path(raw_tmp)
|
|
script = temp_root / "probe_helper.ps1"
|
|
script.write_text(
|
|
"\n".join(
|
|
[
|
|
"param(",
|
|
" [int]$TargetPid, [string]$Installer, [string]$RestartExe,",
|
|
" [string]$HelperLogFile, [string]$InstallerLogFile",
|
|
")",
|
|
"Set-Content -LiteralPath $HelperLogFile -Value 'child-started'",
|
|
"Start-Sleep -Milliseconds 1200",
|
|
"Set-Content -LiteralPath $HelperLogFile -Value 'child-complete'",
|
|
]
|
|
),
|
|
encoding="utf-8-sig",
|
|
)
|
|
installer = temp_root / "Setup.exe"
|
|
installer.write_bytes(b"MZ")
|
|
restart = temp_root / "DoctorWorkstation.exe"
|
|
restart.write_bytes(b"MZ")
|
|
helper_log = temp_root / "helper.log"
|
|
fixed_log = temp_root / "fixed.log"
|
|
installer_log = temp_root / "installer.log"
|
|
original_script = script.read_text(encoding="utf-8-sig")
|
|
fixed_log_literal = str(fixed_log).replace("'", "''")
|
|
script.write_text(
|
|
original_script.replace(
|
|
"Set-Content -LiteralPath $HelperLogFile -Value 'child-started'",
|
|
"Set-Content -LiteralPath '"
|
|
+ fixed_log_literal
|
|
+ "' -Value (\"helper=<{0}> args=<{1}>\" -f $HelperLogFile, ($args -join '|'))\n"
|
|
+ "Set-Content -LiteralPath $HelperLogFile -Value 'child-started'",
|
|
),
|
|
encoding="utf-8-sig",
|
|
)
|
|
captured: list[subprocess.Popen[Any]] = []
|
|
captured_calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
|
|
real_popen = update_service.subprocess.Popen
|
|
|
|
def capture_popen(*args: Any, **kwargs: Any) -> subprocess.Popen[Any]:
|
|
captured_calls.append((args, kwargs.copy()))
|
|
process = real_popen(*args, **kwargs)
|
|
captured.append(process)
|
|
return process
|
|
|
|
update_service.subprocess.Popen = capture_popen
|
|
try:
|
|
started = time.perf_counter()
|
|
update_service._spawn_inno_setup_applier(
|
|
script,
|
|
installer=installer,
|
|
restart_exe=restart,
|
|
helper_log_file=helper_log,
|
|
installer_log_file=installer_log,
|
|
)
|
|
returned_ms = round((time.perf_counter() - started) * 1000, 3)
|
|
finally:
|
|
update_service.subprocess.Popen = real_popen
|
|
deadline = time.perf_counter() + 2.5
|
|
child_log = ""
|
|
while time.perf_counter() < deadline:
|
|
if helper_log.exists():
|
|
child_log = helper_log.read_text(encoding="utf-8").strip()
|
|
if child_log == "child-complete":
|
|
break
|
|
time.sleep(0.05)
|
|
return_code = captured[0].poll() if captured else None
|
|
if captured and return_code is None:
|
|
return_code = captured[0].wait(timeout=2.0)
|
|
matrix: dict[str, Any] = {}
|
|
if captured_calls:
|
|
command = list(captured_calls[0][0][0])
|
|
flag_cases = {
|
|
"zero": 0,
|
|
"detached": getattr(subprocess, "DETACHED_PROCESS", 0),
|
|
"new_process_group": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0),
|
|
"no_window": getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
"detached_new_group": getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
| getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0),
|
|
"detached_no_window": getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
| getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
"new_group_no_window": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
|
| getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
"production_all": getattr(subprocess, "DETACHED_PROCESS", 0)
|
|
| getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
|
| getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
|
}
|
|
running: dict[str, tuple[subprocess.Popen[Any], Path]] = {}
|
|
helper_parameter = command.index("-HelperLogFile") + 1
|
|
for name, flags in flag_cases.items():
|
|
case_command = command.copy()
|
|
case_log = temp_root / f"matrix-{name}.log"
|
|
case_command[helper_parameter] = str(case_log)
|
|
process = real_popen(
|
|
case_command,
|
|
close_fds=True,
|
|
creationflags=flags,
|
|
cwd=str(temp_root),
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
running[name] = (process, case_log)
|
|
production_flags = flag_cases["production_all"]
|
|
detached_flags = flag_cases["detached"]
|
|
extra_cases = {
|
|
"production_close_false": (production_flags, False, False),
|
|
"production_all_devnull": (production_flags, True, True),
|
|
"detached_close_false": (detached_flags, False, False),
|
|
"detached_all_devnull": (detached_flags, True, True),
|
|
}
|
|
for name, (flags, close_fds, all_devnull) in extra_cases.items():
|
|
case_command = command.copy()
|
|
case_log = temp_root / f"matrix-{name}.log"
|
|
case_command[helper_parameter] = str(case_log)
|
|
stream_kwargs = (
|
|
{
|
|
"stdin": subprocess.DEVNULL,
|
|
"stdout": subprocess.DEVNULL,
|
|
"stderr": subprocess.DEVNULL,
|
|
}
|
|
if all_devnull
|
|
else {}
|
|
)
|
|
process = real_popen(
|
|
case_command,
|
|
close_fds=close_fds,
|
|
creationflags=flags,
|
|
cwd=str(temp_root),
|
|
**stream_kwargs,
|
|
)
|
|
flag_cases[name] = flags
|
|
running[name] = (process, case_log)
|
|
matrix_deadline = time.perf_counter() + 3.0
|
|
while time.perf_counter() < matrix_deadline:
|
|
if all(case_log.exists() for _, case_log in running.values()):
|
|
break
|
|
time.sleep(0.05)
|
|
for name, (process, case_log) in running.items():
|
|
matrix[name] = {
|
|
"flags": flag_cases[name],
|
|
"log_created": case_log.exists(),
|
|
"return_code": process.poll(),
|
|
}
|
|
control_return_code = None
|
|
control_stdout = ""
|
|
control_stderr = ""
|
|
if not fixed_log.exists() and captured_calls:
|
|
call_args, call_kwargs = captured_calls[0]
|
|
call_kwargs.update(
|
|
creationflags=0,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
control = real_popen(*call_args, **call_kwargs)
|
|
control_stdout, control_stderr = control.communicate(timeout=5.0)
|
|
control_return_code = control.returncode
|
|
return {
|
|
"command": captured_calls[0][0][0] if captured_calls else [],
|
|
"spawn_returned_ms": returned_ms,
|
|
"child_completed": child_log == "child-complete",
|
|
"child_log": child_log,
|
|
"fixed_log": fixed_log.read_text(encoding="utf-8").strip()
|
|
if fixed_log.exists()
|
|
else "",
|
|
"child_return_code": return_code,
|
|
"flag_matrix": matrix,
|
|
"control_return_code": control_return_code,
|
|
"control_stdout": control_stdout,
|
|
"control_stderr": control_stderr,
|
|
}
|
|
|
|
|
|
def exercise_controller_quit() -> dict[str, Any]:
|
|
"""Run the production request_quit method against a real Qt event loop."""
|
|
|
|
from doctor_workstation.app import ApplicationController
|
|
|
|
app = QApplication.instance() or QApplication([])
|
|
events: list[str] = []
|
|
holder = SimpleNamespace(application=app, _shutting_down=False)
|
|
app.aboutToQuit.connect(lambda: events.append("aboutToQuit"))
|
|
QTimer.singleShot(
|
|
0,
|
|
lambda: (
|
|
events.append("request_quit_enter"),
|
|
ApplicationController.request_quit(holder),
|
|
events.append("request_quit_return"),
|
|
),
|
|
)
|
|
watchdog = QTimer()
|
|
watchdog.setSingleShot(True)
|
|
watchdog.timeout.connect(lambda: (events.append("watchdog"), app.quit()))
|
|
watchdog.start(1000)
|
|
started = time.perf_counter()
|
|
return_code = app.exec()
|
|
return {
|
|
"events": events,
|
|
"return_code": return_code,
|
|
"returned_ms": round((time.perf_counter() - started) * 1000, 3),
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
iteration_count = int(sys.argv[1]) if len(sys.argv) > 1 else 25
|
|
output = {
|
|
"session": exercise_session(iteration_count),
|
|
"real_popen": exercise_real_popen(),
|
|
"controller_quit": exercise_controller_quit(),
|
|
}
|
|
print(json.dumps(output, ensure_ascii=False, indent=2))
|