122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
# 合同浏览器适配：profile 在仓库外、调试口稳定、端口/DevTools 用轮询。
|
||
|
||
from __future__ import annotations
|
||
|
||
import socket
|
||
import threading
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from apa_browser_adapter import (
|
||
ApaBrowserAdapter,
|
||
REPO_ROOT,
|
||
chrome_debug_port,
|
||
resolve_chrome_profile_base,
|
||
resolve_chrome_profile_dir,
|
||
wait_debug_port,
|
||
wait_devtools_port,
|
||
)
|
||
|
||
|
||
# 合同 ApaBrowserAdapter 就是 rpa_browser 里的那一份。
|
||
def test_adapter_class_is_shared_rpa_browser():
|
||
from rpa_browser.adapter import ApaBrowserAdapter as Shared
|
||
|
||
assert ApaBrowserAdapter is Shared
|
||
|
||
|
||
# 未配环境变量时 Chrome 用户目录落在仓库外 rpa-chrome-profile。
|
||
def test_default_chrome_profile_is_outside_repo():
|
||
profile = resolve_chrome_profile_base(requested="")
|
||
assert profile == (REPO_ROOT.parent / "rpa-chrome-profile").resolve()
|
||
assert REPO_ROOT.resolve() not in profile.parents
|
||
assert profile != REPO_ROOT.resolve()
|
||
|
||
|
||
# RPA_CHROME_PROFILE 指到仓库内时改用仓库外默认目录。
|
||
def test_env_path_inside_repo_is_rejected(tmp_path, monkeypatch):
|
||
inside = REPO_ROOT / "rpa_contract" / "screenshots" / "chrome-profile"
|
||
monkeypatch.delenv("RPA_CHROME_PROFILE", raising=False)
|
||
profile = resolve_chrome_profile_base(requested=str(inside))
|
||
assert profile == (REPO_ROOT.parent / "rpa-chrome-profile").resolve()
|
||
assert not _is_relative_to(profile, REPO_ROOT)
|
||
|
||
|
||
# RPA_CHROME_PROFILE 在仓库外时按原路径使用。
|
||
def test_env_path_outside_repo_is_kept(tmp_path):
|
||
outside = tmp_path / "custom-chrome"
|
||
outside.mkdir()
|
||
profile = resolve_chrome_profile_base(requested=str(outside))
|
||
assert profile == outside.resolve()
|
||
|
||
|
||
# 同一 flow 的 profile 目录稳定，且不落在仓库内。
|
||
def test_chrome_profile_dir_is_stable_per_flow():
|
||
first = resolve_chrome_profile_dir("CONTRACT_CREATE")
|
||
second = resolve_chrome_profile_dir("CONTRACT_CREATE")
|
||
assert first == second
|
||
assert first == (REPO_ROOT.parent / "rpa-chrome-profile" / "CONTRACT_CREATE").resolve()
|
||
assert not _is_relative_to(first, REPO_ROOT)
|
||
|
||
|
||
# CONTRACT_CREATE 默认调试口 9333，同一 flow 两次相同。
|
||
def test_chrome_debug_port_is_stable_per_flow(monkeypatch):
|
||
monkeypatch.delenv("RPA_CHROME_DEBUG_PORT", raising=False)
|
||
assert chrome_debug_port("CONTRACT_CREATE") == 9333
|
||
assert chrome_debug_port("CONTRACT_CREATE") == chrome_debug_port("CONTRACT_CREATE")
|
||
|
||
|
||
# 端口已开时 wait_debug_port 立刻返回 True。
|
||
def test_wait_debug_port_true_when_open():
|
||
srv = socket.socket()
|
||
srv.bind(("127.0.0.1", 0))
|
||
srv.listen(1)
|
||
port = srv.getsockname()[1]
|
||
try:
|
||
assert wait_debug_port(port, timeout=1.0) is True
|
||
finally:
|
||
srv.close()
|
||
|
||
|
||
# 端口一直关着时短超时轮询后返回 False，整段远小于无条件 sleep(1)。
|
||
def test_wait_debug_port_false_on_timeout():
|
||
port = _closed_port()
|
||
started = time.monotonic()
|
||
assert wait_debug_port(port, timeout=0.25) is False
|
||
assert time.monotonic() - started < 0.8
|
||
|
||
|
||
# profile 里已有 DevToolsActivePort 时立刻读出端口。
|
||
def test_wait_devtools_port_reads_marker(tmp_path):
|
||
(tmp_path / "DevToolsActivePort").write_text("9333\n/devtools/browser/x\n", encoding="utf-8")
|
||
assert wait_devtools_port(tmp_path, timeout=0.5) == 9333
|
||
|
||
|
||
# DevToolsActivePort 稍后出现时轮询仍能读到端口。
|
||
def test_wait_devtools_port_picks_up_late_marker(tmp_path):
|
||
# 稍后写入 DevToolsActivePort，用来证明 wait 是轮询不是一次读。
|
||
def _write() -> None:
|
||
time.sleep(0.15)
|
||
(tmp_path / "DevToolsActivePort").write_text("9335\n", encoding="utf-8")
|
||
|
||
threading.Thread(target=_write, daemon=True).start()
|
||
assert wait_devtools_port(tmp_path, timeout=1.0) == 9335
|
||
|
||
|
||
# 借一个刚释放的本机端口，调用时应当是关着的。
|
||
def _closed_port() -> int:
|
||
srv = socket.socket()
|
||
srv.bind(("127.0.0.1", 0))
|
||
port = srv.getsockname()[1]
|
||
srv.close()
|
||
return port
|
||
|
||
|
||
# path 是否落在 root 目录树内。
|
||
def _is_relative_to(path: Path, root: Path) -> bool:
|
||
try:
|
||
path.resolve().relative_to(root.resolve())
|
||
except ValueError:
|
||
return False
|
||
return True
|