245 lines
9.4 KiB
Python
245 lines
9.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Browser launch helper checks (no real browser started).
|
|
"""
|
|
import json
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
|
|
class TestPlaywrightLaunchHelpers:
|
|
"""Persistent context launch parts must match RPA contract."""
|
|
|
|
def test_persistent_context_includes_maximized(self):
|
|
from util.playwright_stealth import persistent_context_launch_parts
|
|
|
|
args, _ignore = persistent_context_launch_parts()
|
|
assert "--start-maximized" in args
|
|
|
|
|
|
class TestEnsureSessionRestoreDoesNotTouchProtectedPref:
|
|
def test_leaves_restore_on_startup_unchanged(self, tmp_path, fake_env):
|
|
from pathlib import Path
|
|
|
|
from service.chrome_profile_prefs import ensure_session_restore, preferences_path
|
|
from util.logging_config import setup_skill_logging
|
|
|
|
setup_skill_logging("account-manager", "openclaw.skill.account_manager_test")
|
|
|
|
profile = tmp_path / "prof"
|
|
prefs = Path(preferences_path(str(profile)))
|
|
prefs.parent.mkdir(parents=True, exist_ok=True)
|
|
prefs.write_text(
|
|
json.dumps(
|
|
{
|
|
"session": {"restore_on_startup": 5},
|
|
"profile": {"exit_type": "Crashed", "exited_cleanly": False},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = ensure_session_restore(str(profile))
|
|
assert result["ok"] is True
|
|
assert result["prefs_restore_on_startup_touched"] is False
|
|
data = json.loads(prefs.read_text(encoding="utf-8"))
|
|
assert data["session"]["restore_on_startup"] == 5
|
|
assert data["profile"]["exit_type"] == "Normal"
|
|
assert data["profile"]["exited_cleanly"] is True
|
|
|
|
|
|
class TestProfileHasRestorableSession:
|
|
def test_empty_profile_false(self, tmp_path):
|
|
from service.chrome_profile_prefs import profile_has_restorable_session
|
|
|
|
assert profile_has_restorable_session(str(tmp_path / "prof")) is False
|
|
|
|
def test_legacy_current_session(self, tmp_path):
|
|
from service.chrome_profile_prefs import profile_has_restorable_session
|
|
|
|
default = tmp_path / "prof" / "Default"
|
|
default.mkdir(parents=True)
|
|
(default / "Current Session").write_bytes(b"x")
|
|
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
|
|
|
|
def test_modern_sessions_dir(self, tmp_path):
|
|
from service.chrome_profile_prefs import profile_has_restorable_session
|
|
|
|
sessions = tmp_path / "prof" / "Default" / "Sessions"
|
|
sessions.mkdir(parents=True)
|
|
(sessions / "Session_13434010951759956").write_bytes(b"sess")
|
|
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
|
|
|
|
def test_modern_tabs_only(self, tmp_path):
|
|
from service.chrome_profile_prefs import profile_has_restorable_session
|
|
|
|
sessions = tmp_path / "prof" / "Default" / "Sessions"
|
|
sessions.mkdir(parents=True)
|
|
(sessions / "Tabs_13434010952144072").write_bytes(b"tabs")
|
|
assert profile_has_restorable_session(str(tmp_path / "prof")) is True
|
|
|
|
|
|
class TestCmdOpenDirectBrowser:
|
|
"""account open = same-process Chrome/Edge launch (like reveal-profile)."""
|
|
|
|
def _seed_account(self, clean_db, tmp_path, profile):
|
|
from db.accounts_repo import insert_account
|
|
from util.platforms import seed_platforms
|
|
|
|
seed_platforms(clean_db)
|
|
aid = insert_account(
|
|
clean_db,
|
|
platform_key="sohu",
|
|
account_label="open-test",
|
|
login_id="1",
|
|
login_id_type="phone",
|
|
tenant_id=None,
|
|
environment="production",
|
|
role="default",
|
|
provider_code="sohu",
|
|
profile_dir=str(profile),
|
|
url="https://example.com",
|
|
extra_json=None,
|
|
now=1_700_000_000,
|
|
auth_strategy="password_auto",
|
|
)
|
|
clean_db.commit()
|
|
return aid
|
|
|
|
def _patch_launch(self, monkeypatch, browser_service, tmp_path):
|
|
fake_exe = str(tmp_path / "chrome.exe")
|
|
monkeypatch.setattr(
|
|
browser_service,
|
|
"resolve_chromium_executable",
|
|
lambda: fake_exe,
|
|
)
|
|
|
|
class FakeProc:
|
|
pid = 5252
|
|
|
|
spawned = {}
|
|
|
|
def fake_popen(cmd, **kwargs):
|
|
spawned["cmd"] = list(cmd)
|
|
spawned["kwargs"] = kwargs
|
|
return FakeProc()
|
|
|
|
monkeypatch.setattr(browser_service.subprocess, "Popen", fake_popen)
|
|
return fake_exe, spawned
|
|
|
|
def test_cmd_open_launches_browser_exe(self, clean_db, tmp_path, capsys, monkeypatch):
|
|
from service import browser_service
|
|
|
|
profile = tmp_path / "prof"
|
|
aid = self._seed_account(clean_db, tmp_path, profile)
|
|
fake_exe, spawned = self._patch_launch(monkeypatch, browser_service, tmp_path)
|
|
|
|
browser_service.cmd_open(str(aid))
|
|
out = capsys.readouterr().out.strip()
|
|
payload = json.loads(out.splitlines()[-1])
|
|
|
|
assert payload["success"] is True
|
|
assert payload["message"] == ""
|
|
assert payload["type"] == "result"
|
|
assert payload["status"] == "success"
|
|
assert payload["data"]["pid"] == 5252
|
|
assert payload["data"]["browser_exe"] == fake_exe
|
|
assert spawned["cmd"][0] == fake_exe
|
|
assert spawned["cmd"][1].startswith("--user-data-dir=")
|
|
assert "--start-maximized" in spawned["cmd"]
|
|
assert "--no-first-run" in spawned["cmd"]
|
|
assert "--no-default-browser-check" in spawned["cmd"]
|
|
assert "--disable-search-engine-choice-screen" in spawned["cmd"]
|
|
assert spawned["cmd"][-1] == "https://example.com"
|
|
assert "--enable-automation" not in spawned["cmd"]
|
|
# must NOT spawn python / open_child_runner
|
|
joined = " ".join(spawned["cmd"])
|
|
assert "open_child_runner" not in joined
|
|
assert "python" not in joined.lower()
|
|
|
|
def test_cmd_open_skips_url_when_modern_sessions_exist(
|
|
self, clean_db, tmp_path, capsys, monkeypatch
|
|
):
|
|
"""Modern Chrome Sessions/ must restore via --restore-last-session, no homepage URL."""
|
|
from service import browser_service
|
|
|
|
profile = tmp_path / "prof"
|
|
sessions = profile / "Default" / "Sessions"
|
|
sessions.mkdir(parents=True)
|
|
(sessions / "Session_13434010951759956").write_bytes(b"sess")
|
|
aid = self._seed_account(clean_db, tmp_path, profile)
|
|
_fake_exe, spawned = self._patch_launch(monkeypatch, browser_service, tmp_path)
|
|
|
|
browser_service.cmd_open(str(aid))
|
|
out = capsys.readouterr().out.strip()
|
|
payload = json.loads(out.splitlines()[-1])
|
|
|
|
assert payload["success"] is True
|
|
assert payload["data"]["has_session"] is True
|
|
assert payload["data"]["restore_last_session"] is True
|
|
assert "https://example.com" not in spawned["cmd"]
|
|
assert "--restore-last-session" in spawned["cmd"]
|
|
assert spawned["cmd"][-1] == "--restore-last-session" or spawned["cmd"][-1].startswith("--")
|
|
|
|
|
|
class TestHumanChromeLaunchArgs:
|
|
def test_flags_then_url(self, tmp_path):
|
|
from service.browser_service import HUMAN_CHROME_FLAGS, human_chrome_launch_args
|
|
|
|
profile = str(tmp_path / "prof")
|
|
args = human_chrome_launch_args(profile, "https://chat.deepseek.com")
|
|
assert args[0].startswith("--user-data-dir=")
|
|
for flag in HUMAN_CHROME_FLAGS:
|
|
assert flag in args
|
|
assert "--restore-last-session" not in args
|
|
assert any(a.startswith("--remote-debugging-port=") for a in args)
|
|
assert args[-1] == "https://chat.deepseek.com"
|
|
assert "--enable-automation" not in args
|
|
|
|
def test_restore_session_flag_without_url(self, tmp_path):
|
|
from service.browser_service import human_chrome_launch_args
|
|
|
|
args = human_chrome_launch_args(str(tmp_path / "prof"), None, restore_session=True)
|
|
assert "--restore-last-session" in args
|
|
assert any(a.startswith("--remote-debugging-port=") for a in args)
|
|
assert not any(a.startswith("http") for a in args)
|
|
|
|
def test_shortcut_arguments_quote_profile_and_include_url(self, tmp_path):
|
|
from service.browser_service import human_chrome_shortcut_arguments
|
|
|
|
profile = tmp_path / "my prof"
|
|
profile.mkdir()
|
|
text = human_chrome_shortcut_arguments(str(profile), "https://example.com")
|
|
assert '--user-data-dir="' in text
|
|
assert "--start-maximized" in text
|
|
assert "--no-first-run" in text
|
|
assert "--restore-last-session" not in text
|
|
assert "--remote-debugging-port=" in text
|
|
assert text.endswith("https://example.com")
|
|
|
|
def test_shortcut_restore_session(self, tmp_path):
|
|
from service.browser_service import human_chrome_shortcut_arguments
|
|
|
|
text = human_chrome_shortcut_arguments(str(tmp_path / "p"), None, restore_session=True)
|
|
assert "--restore-last-session" in text
|
|
assert "--remote-debugging-port=" in text
|
|
assert "http" not in text
|
|
|
|
def test_cdp_port_stable_for_profile(self, tmp_path):
|
|
from service.browser_service import cdp_port_for_profile
|
|
|
|
p = str(tmp_path / "prof")
|
|
assert cdp_port_for_profile(p) == cdp_port_for_profile(p)
|
|
assert 9330 <= cdp_port_for_profile(p) < 9390
|
|
|
|
|
|
class TestResolveChromiumChannel:
|
|
"""resolve_chromium_channel returns a channel string when browsers exist."""
|
|
|
|
def test_returns_string_or_none(self):
|
|
from service.browser_service import resolve_chromium_channel
|
|
|
|
ch = resolve_chromium_channel()
|
|
assert ch is None or ch in ("chrome", "msedge")
|