cmshoppe/tests/test_chrome.py

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()