chengma
e325f66eef
- 新增 app/chrome.py 拼装远程调试启动参数并启动 Chrome - 实现 /json/version 端口探测并禁用环境代理 - 新增 chrome 单元测试覆盖参数、Popen 和端口就绪 - 更新任务看板、API 合约、当前状态和进度记录
124 lines
4.3 KiB
Python
124 lines
4.3 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from _helpers import TempDirMixin
|
|
|
|
from app import chrome
|
|
from app import config as account_config
|
|
|
|
|
|
class JsonVersionHandler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path != "/json/version":
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
return
|
|
body = json.dumps({"Browser": "Chrome/Test"}).encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, format, *args):
|
|
return
|
|
|
|
|
|
class ChromeTests(TempDirMixin, unittest.TestCase):
|
|
def start_json_version_server(self):
|
|
server = HTTPServer(("127.0.0.1", 0), JsonVersionHandler)
|
|
thread = threading.Thread(target=server.serve_forever)
|
|
thread.daemon = True
|
|
thread.start()
|
|
self.addCleanup(server.server_close)
|
|
self.addCleanup(server.shutdown)
|
|
self.addCleanup(thread.join, 2)
|
|
return server.server_address[1]
|
|
|
|
def unused_port(self):
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
sock.bind(("127.0.0.1", 0))
|
|
port = sock.getsockname()[1]
|
|
sock.close()
|
|
return port
|
|
|
|
def test_build_launch_args_includes_required_flags(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = {
|
|
"chrome_path": r"C:\Program Files\Google\Chrome\Application\chrome.exe",
|
|
"user_data_root": temp_dir,
|
|
}
|
|
account = {"alias": "alias", "debug_port": "9222"}
|
|
|
|
args = chrome.build_launch_args(account, config=cfg)
|
|
slug = account_config.make_slug("alias")
|
|
user_data_dir = os.path.abspath(os.path.join(temp_dir, slug))
|
|
|
|
self.assertEqual(cfg["chrome_path"], args[0])
|
|
self.assertIn("--remote-debugging-port=9222", args)
|
|
self.assertIn("--remote-allow-origins=*", args)
|
|
self.assertIn(f"--user-data-dir={user_data_dir}", args)
|
|
self.assertTrue(os.path.isdir(user_data_dir))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_build_launch_args_accepts_account_object_user_data_dir(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
user_data_dir = os.path.join(temp_dir, "custom")
|
|
account = SimpleNamespace(debug_port=9333, user_data_dir=user_data_dir)
|
|
args = chrome.build_launch_args(
|
|
account,
|
|
config={"chrome_path": "chrome.exe"},
|
|
)
|
|
|
|
self.assertIn("--remote-debugging-port=9333", args)
|
|
self.assertIn(f"--user-data-dir={os.path.abspath(user_data_dir)}", args)
|
|
self.assertTrue(os.path.isdir(user_data_dir))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_build_launch_args_rejects_invalid_account(self):
|
|
with self.assertRaises(chrome.ChromeLaunchError):
|
|
chrome.build_launch_args(
|
|
{"alias": "alias", "debug_port": 0},
|
|
config={"chrome_path": "chrome.exe"},
|
|
)
|
|
|
|
def test_launch_chrome_uses_subprocess_popen(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
account = {"alias": "alias", "debug_port": 9222}
|
|
cfg = {"chrome_path": "chrome.exe", "user_data_root": temp_dir}
|
|
process = object()
|
|
|
|
with mock.patch("app.chrome.subprocess.Popen", return_value=process) as popen:
|
|
result = chrome.launch_chrome(account, config=cfg)
|
|
|
|
self.assertIs(process, result)
|
|
popen.assert_called_once()
|
|
args = popen.call_args[0][0]
|
|
self.assertEqual("chrome.exe", args[0])
|
|
self.assertIn("--remote-debugging-port=9222", args)
|
|
self.assertIn("--remote-allow-origins=*", args)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_is_running_and_wait_debug_ready(self):
|
|
port = self.start_json_version_server()
|
|
|
|
self.assertTrue(chrome.is_running(port))
|
|
self.assertTrue(chrome.wait_debug_ready(port, timeout=1))
|
|
self.assertFalse(chrome.is_running(self.unused_port()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|