test(rpa_helpers): add mock-based tests for browser/human/auth_flows

Co-authored-by: Cursor <cursoragent@cursor.com>

tests/test_rpa_helpers_auth_flows.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def test_ensure_logged_in_dispatches_password_auto(monkeypatch):
+    from service.rpa_helpers import auth_flows
+
+    monkeypatch.setattr(auth_flows, "_is_already_logged_in", lambda p, a: False)
+    monkeypatch.setattr(auth_flows, "_poll_until", lambda *a, **k: True)
+
+    loc = MagicMock()
+    page = MagicMock()
+    page.url = "http://example.com"
+    page.locator.return_value.first = loc
+
+    account = {"auth_strategy": "password_auto", "login_id": "u", "url": "http://x"}
+    r = auth_flows.ensure_logged_in(page, account, credentials={"plaintext": "pw"})
+    assert r.ok is True
+    page.goto.assert_called()
+    assert loc.type.call_count >= 2
+    loc.click.assert_called()
+
+
+def test_ensure_logged_in_dispatches_qr_code_manual(monkeypatch):
+    from service.rpa_helpers import auth_flows
+
+    monkeypatch.setattr(auth_flows, "_is_already_logged_in", lambda p, a: True)
+
+    page = MagicMock()
+    account = {"auth_strategy": "qr_code_manual", "url": "http://x"}
+    r = auth_flows.ensure_logged_in(page, account)
+    assert r.ok is True
+
+
+def test_ensure_logged_in_client_certificate_raises():
+    from service.rpa_helpers.auth_flows import AuthStrategyError, ensure_logged_in
+
+    page = MagicMock()
+    account = {"auth_strategy": "client_certificate"}
+    with pytest.raises(AuthStrategyError) as ei:
+        ensure_logged_in(page, account)
+    assert ei.value.code == "ERR_AUTH_STRATEGY_NOT_AUTOMATABLE"
+
+
+def test_ensure_logged_in_api_token_no_op():
+    from service.rpa_helpers.auth_flows import ensure_logged_in
+
+    page = MagicMock()
+    account = {"auth_strategy": "api_token"}
+    r = ensure_logged_in(page, account)
+    assert r.ok is True
+    page.goto.assert_not_called()
+
+
+def test_ensure_logged_in_missing_credentials():
+    from service.rpa_helpers.auth_flows import AuthStrategyError, ensure_logged_in
+
+    page = MagicMock()
+    account = {"auth_strategy": "password_auto", "login_id": "u"}
+    with pytest.raises(AuthStrategyError) as ei:
+        ensure_logged_in(page, account, credentials=None)
+    assert ei.value.code == "ERR_CREDENTIAL_MISSING"
+
+
+def test_ensure_logged_in_unknown_strategy():
+    from service.rpa_helpers.auth_flows import AuthStrategyError, ensure_logged_in
+
+    page = MagicMock()
+    account = {"auth_strategy": "bogus"}
+    with pytest.raises(AuthStrategyError) as ei:
+        ensure_logged_in(page, account)
+    assert ei.value.code == "ERR_AUTH_STRATEGY_NOT_SUPPORTED"
+
+
+def test_ensure_logged_in_no_strategy():
+    from service.rpa_helpers.auth_flows import AuthStrategyError, ensure_logged_in
+
+    page = MagicMock()
+    account = {"auth_strategy": ""}
+    with pytest.raises(AuthStrategyError) as ei:
+        ensure_logged_in(page, account)
+    assert ei.value.code == "ERR_AUTH_STRATEGY_NOT_SUPPORTED"
+
+
+def test_device_bound_fingerprint_mismatch(monkeypatch):
+    from service.rpa_helpers.auth_flows import AuthStrategyError, ensure_logged_in
+
+    monkeypatch.setattr(
+        "service.rpa_helpers.auth_flows._current_machine_fingerprint",
+        lambda: "right_fp",
+    )
+    page = MagicMock()
+    account = {
+        "auth_strategy": "device_bound",
+        "device_fingerprint": "wrong_fp",
+        "login_id": "u",
+        "url": "http://x",
+    }
+    with pytest.raises(AuthStrategyError) as ei:
+        ensure_logged_in(page, account, credentials={"plaintext": "pw"})
+    assert ei.value.code == "ERR_DEVICE_FINGERPRINT_MISMATCH"
+

tests/test_rpa_helpers_browser.py

@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import MagicMock
+
+import pytest
+
+
+@pytest.fixture
+def mock_playwright(monkeypatch):
+    pw = MagicMock()
+    ctx = MagicMock()
+    page = MagicMock()
+    ctx.pages = [page]
+    pw.chromium.launch_persistent_context.return_value = ctx
+
+    sp = MagicMock()
+    sp.start.return_value = pw
+    monkeypatch.setattr("service.rpa_helpers.browser.sync_playwright", lambda: sp)
+
+    return {"pw": pw, "ctx": ctx, "page": page, "sp": sp}
+
+
+def test_launch_browser_uses_chrome_channel_default(mock_playwright, monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENCLAW_PLAYWRIGHT_STEALTH", "1")
+    monkeypatch.setattr(
+        "service.rpa_helpers.browser.resolve_chromium_channel",
+        lambda: "chrome",
+    )
+    prof = tmp_path / "p1"
+    from service.rpa_helpers.browser import launch_browser_with_profile
+
+    ctx, page = launch_browser_with_profile(str(prof))
+    pw = mock_playwright["pw"]
+    kw = pw.chromium.launch_persistent_context.call_args.kwargs
+    assert kw["channel"] == "chrome"
+    assert "--disable-blink-features=AutomationControlled" in kw["args"]
+    assert ctx is mock_playwright["ctx"]
+    assert page is mock_playwright["page"]
+
+
+def test_launch_browser_respects_explicit_channel(mock_playwright, monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENCLAW_PLAYWRIGHT_STEALTH", "1")
+    prof = tmp_path / "p2"
+    from service.rpa_helpers.browser import launch_browser_with_profile
+
+    launch_browser_with_profile(str(prof), channel="msedge")
+    kw = mock_playwright["pw"].chromium.launch_persistent_context.call_args.kwargs
+    assert kw["channel"] == "msedge"
+
+
+def test_launch_browser_creates_profile_dir(mock_playwright, monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENCLAW_PLAYWRIGHT_STEALTH", "1")
+    monkeypatch.setattr(
+        "service.rpa_helpers.browser.resolve_chromium_channel",
+        lambda: "chrome",
+    )
+    prof = tmp_path / "new_prof"
+    assert not prof.is_dir()
+    from service.rpa_helpers.browser import launch_browser_with_profile
+
+    launch_browser_with_profile(str(prof))
+    assert prof.is_dir()

tests/test_rpa_helpers_human.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def test_pause_respects_range(monkeypatch):
+    sleeps = []
+
+    def _rec(s):
+        sleeps.append(s)
+
+    monkeypatch.setattr("service.rpa_helpers.human.time.sleep", _rec)
+    from service.rpa_helpers.human import pause
+
+    for _ in range(10):
+        pause(50, 100)
+    assert len(sleeps) == 10
+    for s in sleeps:
+        assert 0.04 <= s <= 0.12
+
+
+def test_human_type_calls_locator_methods():
+    from service.rpa_helpers.human import human_type
+
+    loc = MagicMock()
+    human_type(loc, "hello")
+    loc.scroll_into_view_if_needed.assert_called()
+    loc.click.assert_called()
+    loc.type.assert_called_once()
+    args, kwargs = loc.type.call_args
+    assert args[0] == "hello"
+    assert "delay" in kwargs
+    assert 50 <= kwargs["delay"] <= 200
+
+
+def test_human_click_basic():
+    from service.rpa_helpers.human import human_click
+
+    loc = MagicMock()
+    human_click(loc)
+    loc.scroll_into_view_if_needed.assert_called()
+    loc.click.assert_called()
+
+
+def test_human_select_basic():
+    from service.rpa_helpers.human import human_select
+
+    loc = MagicMock()
+    human_select(loc, "opt1")
+    loc.scroll_into_view_if_needed.assert_called()
+    loc.click.assert_called()
+    assert loc.select_option.call_count >= 1
+