325 lines
13 KiB
Python
325 lines
13 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 appconfig, 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 make_executable(self, root, *parts):
|
|
path = os.path.join(root, *parts)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
fh.write("chrome")
|
|
return path
|
|
|
|
def test_registry_candidates_prioritize_current_user_and_all_views(self):
|
|
class FakeKey:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
class FakeWinreg:
|
|
HKEY_CURRENT_USER = "HKCU"
|
|
HKEY_LOCAL_MACHINE = "HKLM"
|
|
KEY_READ = 1
|
|
KEY_WOW64_64KEY = 16
|
|
KEY_WOW64_32KEY = 32
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
self.values = {
|
|
(self.HKEY_CURRENT_USER, self.KEY_READ | self.KEY_WOW64_32KEY): "user-chrome.exe",
|
|
(self.HKEY_LOCAL_MACHINE, self.KEY_READ): "machine-chrome.exe",
|
|
}
|
|
|
|
def OpenKey(self, hive, sub_key, _reserved, access):
|
|
self.calls.append((hive, sub_key, access))
|
|
try:
|
|
return FakeKey(self.values[(hive, access)])
|
|
except KeyError as exc:
|
|
raise FileNotFoundError() from exc
|
|
|
|
@staticmethod
|
|
def QueryValueEx(key, name):
|
|
self.assertEqual("", name)
|
|
return key.value, 1
|
|
|
|
fake = FakeWinreg()
|
|
with mock.patch("app.chrome._load_winreg", return_value=fake):
|
|
candidates = list(chrome._registry_chrome_candidates())
|
|
|
|
self.assertEqual(["user-chrome.exe", "machine-chrome.exe"], candidates)
|
|
self.assertEqual("HKCU", fake.calls[0][0])
|
|
self.assertEqual(chrome.CHROME_APP_PATHS_KEY, fake.calls[0][1])
|
|
self.assertIn(("HKCU", chrome.CHROME_APP_PATHS_KEY, 1 | 16), fake.calls)
|
|
self.assertIn(("HKCU", chrome.CHROME_APP_PATHS_KEY, 1 | 32), fake.calls)
|
|
self.assertIn(("HKLM", chrome.CHROME_APP_PATHS_KEY, 1), fake.calls)
|
|
|
|
def test_detect_chrome_path_normalizes_registry_value(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
executable = self.make_executable(temp_dir, "Chrome", "chrome.exe")
|
|
environment_name = "CMSHOPEE_TEST_CHROME"
|
|
registry_value = f'"%{environment_name}%"'
|
|
|
|
with mock.patch.dict(os.environ, {environment_name: executable}, clear=False), \
|
|
mock.patch(
|
|
"app.chrome._registry_chrome_candidates",
|
|
return_value=[registry_value],
|
|
), \
|
|
mock.patch("app.chrome._standard_chrome_candidates", return_value=[]), \
|
|
mock.patch("app.chrome.shutil.which", return_value=None):
|
|
detected = chrome.detect_chrome_path()
|
|
|
|
self.assertEqual(os.path.abspath(executable), detected)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_detect_chrome_path_uses_local_app_data_without_registry(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
executable = self.make_executable(
|
|
temp_dir,
|
|
"Google",
|
|
"Chrome",
|
|
"Application",
|
|
"chrome.exe",
|
|
)
|
|
with mock.patch.dict(os.environ, {"LOCALAPPDATA": temp_dir}, clear=True), \
|
|
mock.patch("app.chrome._registry_chrome_candidates", return_value=[]), \
|
|
mock.patch("app.chrome.shutil.which", return_value=None):
|
|
detected = chrome.detect_chrome_path()
|
|
|
|
self.assertEqual(os.path.abspath(executable), detected)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_detect_chrome_path_never_uses_edge_and_handles_missing_registry(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
self.make_executable(
|
|
temp_dir,
|
|
"Microsoft",
|
|
"Edge",
|
|
"Application",
|
|
"msedge.exe",
|
|
)
|
|
with mock.patch("app.chrome._load_winreg", return_value=None), \
|
|
mock.patch.dict(os.environ, {"LOCALAPPDATA": temp_dir}, clear=True), \
|
|
mock.patch("app.chrome.shutil.which", return_value=None):
|
|
detected = chrome.detect_chrome_path()
|
|
|
|
self.assertEqual("", detected)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_ensure_configured_chrome_path_persists_only_when_invalid(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
config_path = os.path.join(temp_dir, "config.json")
|
|
detected = self.make_executable(temp_dir, "Chrome", "chrome.exe")
|
|
config = appconfig.load_config(config_path)
|
|
|
|
with mock.patch("app.chrome.detect_chrome_path", return_value=detected):
|
|
first = chrome.ensure_configured_chrome_path(config)
|
|
|
|
self.assertTrue(first["changed"])
|
|
self.assertEqual(os.path.abspath(detected), first["config"]["chrome_path"])
|
|
self.assertIn("已自动定位 Chrome", first["message"])
|
|
self.assertEqual(
|
|
os.path.abspath(detected),
|
|
appconfig.load_config(config_path)["chrome_path"],
|
|
)
|
|
|
|
missing = os.path.join(temp_dir, "missing", "chrome.exe")
|
|
invalid = appconfig.save_config(
|
|
{"chrome_path": missing},
|
|
path=config_path,
|
|
)
|
|
with mock.patch("app.chrome.detect_chrome_path", return_value=detected):
|
|
repaired = chrome.ensure_configured_chrome_path(invalid)
|
|
self.assertTrue(repaired["changed"])
|
|
self.assertEqual(os.path.abspath(detected), repaired["config"]["chrome_path"])
|
|
|
|
custom = self.make_executable(temp_dir, "Portable", "chrome.exe")
|
|
valid = appconfig.save_config(
|
|
{"chrome_path": custom},
|
|
path=config_path,
|
|
)
|
|
with mock.patch("app.chrome.detect_chrome_path") as detect:
|
|
unchanged = chrome.ensure_configured_chrome_path(valid)
|
|
self.assertFalse(unchanged["changed"])
|
|
self.assertEqual(custom, unchanged["config"]["chrome_path"])
|
|
detect.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_ensure_configured_chrome_path_preserves_path_command_and_no_match(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
config_path = os.path.join(temp_dir, "config.json")
|
|
config = appconfig.save_config(
|
|
{"chrome_path": "chrome.exe"},
|
|
path=config_path,
|
|
)
|
|
resolved_path = self.make_executable(temp_dir, "PATH", "chrome.exe")
|
|
|
|
with mock.patch("app.chrome.shutil.which", return_value=resolved_path), \
|
|
mock.patch("app.chrome.detect_chrome_path") as detect:
|
|
unchanged = chrome.ensure_configured_chrome_path(config)
|
|
self.assertFalse(unchanged["changed"])
|
|
detect.assert_not_called()
|
|
|
|
empty = appconfig.save_config({"chrome_path": ""}, path=config_path)
|
|
with mock.patch("app.chrome.detect_chrome_path", return_value=""), \
|
|
mock.patch("app.chrome.appconfig.save_config") as save:
|
|
no_match = chrome.ensure_configured_chrome_path(empty)
|
|
self.assertFalse(no_match["changed"])
|
|
self.assertEqual("", no_match["message"])
|
|
save.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
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_create_shortcut_writes_lnk_with_required_chrome_arguments(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
chrome_path = os.path.join(temp_dir, "Chrome App", "chrome.exe")
|
|
shortcut_path = os.path.join(temp_dir, "Desktop", "主店.lnk")
|
|
account = {
|
|
"alias": "alias",
|
|
"debug_port": 9222,
|
|
"user_data_dir": os.path.join(temp_dir, "profile with space"),
|
|
}
|
|
cfg = {"chrome_path": chrome_path}
|
|
|
|
with mock.patch("app.chrome.subprocess.run") as run:
|
|
result = chrome.create_shortcut(
|
|
account,
|
|
shortcut_path=shortcut_path,
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertEqual(os.path.abspath(shortcut_path), result)
|
|
run.assert_called_once()
|
|
command = run.call_args[0][0]
|
|
self.assertEqual("powershell", command[0])
|
|
script = command[-1]
|
|
self.assertIn("$shortcut.TargetPath", script)
|
|
self.assertIn(chrome_path, script)
|
|
self.assertIn("--remote-debugging-port=9222", script)
|
|
self.assertIn("--remote-allow-origins=*", script)
|
|
self.assertIn("--user-data-dir=", script)
|
|
self.assertIn("profile with space", script)
|
|
|
|
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()
|