fix(rpa_helpers): require positive evidence in _is_already_logged_in to avoid false success
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -68,35 +68,37 @@ def _account_extra(account: dict) -> dict:
|
|||||||
|
|
||||||
|
|
||||||
def _is_already_logged_in(page, account: dict) -> bool:
|
def _is_already_logged_in(page, account: dict) -> bool:
|
||||||
"""粗略判断是否已登录(可被 extra_json.logged_in_url_pattern 覆盖)。"""
|
"""是否已登录:仅当 extra_json 提供正向证据(URL 正则或可见 selector)时为 True。"""
|
||||||
try:
|
try:
|
||||||
cur = page.url or ""
|
cur = page.url or ""
|
||||||
except Exception:
|
except Exception:
|
||||||
return False
|
return False
|
||||||
|
cur_stripped = cur.strip()
|
||||||
|
if not cur_stripped:
|
||||||
|
return False
|
||||||
|
ul = cur_stripped.lower()
|
||||||
|
if ul == "about:blank" or ul.startswith("about:blank"):
|
||||||
|
return False
|
||||||
|
|
||||||
extra = _account_extra(account)
|
extra = _account_extra(account)
|
||||||
|
|
||||||
pattern = extra.get("logged_in_url_pattern")
|
pattern = extra.get("logged_in_url_pattern")
|
||||||
if isinstance(pattern, str) and pattern:
|
if isinstance(pattern, str) and pattern.strip():
|
||||||
try:
|
try:
|
||||||
return bool(re.search(pattern, cur))
|
if re.search(pattern, cur_stripped):
|
||||||
|
return True
|
||||||
except re.error:
|
except re.error:
|
||||||
pass
|
pass
|
||||||
lc = cur.lower()
|
|
||||||
if any(k in lc for k in ("login", "signin", "sign-in", "auth", "/sso/")):
|
sel = extra.get("logged_in_selector")
|
||||||
return False
|
if isinstance(sel, str) and sel.strip():
|
||||||
try:
|
try:
|
||||||
login_btn = page.locator(
|
if page.locator(sel.strip()).first.is_visible(timeout=500):
|
||||||
"button:has-text('登录'), button:has-text('Sign in'), "
|
return True
|
||||||
"a:has-text('登录'), a:has-text('Sign in')"
|
except Exception:
|
||||||
)
|
pass
|
||||||
if login_btn.count() > 0:
|
|
||||||
try:
|
return False
|
||||||
if login_btn.first.is_visible(timeout=500):
|
|
||||||
return False
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
return True
|
|
||||||
|
|
||||||
|
|
||||||
def _current_machine_fingerprint() -> str:
|
def _current_machine_fingerprint() -> str:
|
||||||
|
|||||||
@@ -102,3 +102,56 @@ def test_device_bound_fingerprint_mismatch(monkeypatch):
|
|||||||
ensure_logged_in(page, account, credentials={"plaintext": "pw"})
|
ensure_logged_in(page, account, credentials={"plaintext": "pw"})
|
||||||
assert ei.value.code == "ERR_DEVICE_FINGERPRINT_MISMATCH"
|
assert ei.value.code == "ERR_DEVICE_FINGERPRINT_MISMATCH"
|
||||||
|
|
||||||
|
|
||||||
|
def test_is_already_logged_in_default_false():
|
||||||
|
from service.rpa_helpers.auth_flows import _is_already_logged_in
|
||||||
|
|
||||||
|
p = MagicMock()
|
||||||
|
p.url = "about:blank"
|
||||||
|
assert _is_already_logged_in(p, {}) is False
|
||||||
|
|
||||||
|
p.url = "https://www.douyin.com"
|
||||||
|
assert _is_already_logged_in(p, {}) is False
|
||||||
|
|
||||||
|
p.url = "https://www.douyin.com/user/abc"
|
||||||
|
assert _is_already_logged_in(
|
||||||
|
p,
|
||||||
|
{"extra_json": {"logged_in_url_pattern": r"/user/"}},
|
||||||
|
) is True
|
||||||
|
|
||||||
|
loc_first = MagicMock()
|
||||||
|
loc_first.is_visible.return_value = True
|
||||||
|
loc = MagicMock()
|
||||||
|
loc.first = loc_first
|
||||||
|
p2 = MagicMock()
|
||||||
|
p2.url = "https://example.com/home"
|
||||||
|
p2.locator.return_value = loc
|
||||||
|
assert _is_already_logged_in(
|
||||||
|
p2,
|
||||||
|
{"extra_json": {"logged_in_selector": "#avatar"}},
|
||||||
|
) is True
|
||||||
|
p2.locator.assert_called_with("#avatar")
|
||||||
|
|
||||||
|
|
||||||
|
def test_qr_code_manual_default_no_hooks():
|
||||||
|
from service.rpa_helpers.auth_flows import ensure_logged_in
|
||||||
|
|
||||||
|
qr_first = MagicMock()
|
||||||
|
qr_first.wait_for.side_effect = TimeoutError("mock qr timeout")
|
||||||
|
qr_loc = MagicMock()
|
||||||
|
qr_loc.first = qr_first
|
||||||
|
page = MagicMock()
|
||||||
|
page.url = "about:blank"
|
||||||
|
page.locator.return_value = qr_loc
|
||||||
|
|
||||||
|
account = {"auth_strategy": "qr_code_manual", "url": "http://login.example"}
|
||||||
|
r = ensure_logged_in(page, account)
|
||||||
|
page.goto.assert_called_once_with(
|
||||||
|
"http://login.example",
|
||||||
|
wait_until="domcontentloaded",
|
||||||
|
timeout=60000,
|
||||||
|
)
|
||||||
|
qr_first.wait_for.assert_called_once()
|
||||||
|
assert r.ok is False
|
||||||
|
assert r.message == "qr code not found"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user