""" Attaching to a Chrome the bot did not start (BROWSER_CDP_URL). python tests\\test_cdp_browser.py The operator starts Chrome with `start_ecw_chrome.bat` (profile on D:, port 9222 on 127.0.0.1) and adds the Claude extension to it, so a run can be watched live. The bot then attaches over CDP instead of launching its own Chromium. What has to hold, each checked against a real Chromium started the same way the .bat starts Chrome: * **Only a loopback URL is accepted.** Anything that can reach the port can drive a browser logged in to eCW. * **eCW's cookies and storage go at the end of a run, and at the start** (a run killed half-way leaves them behind). Only eCW's: claude.ai's cookie is what keeps the extension signed in. * **close() leaves Chrome running** and the operator's own tab open — it closes the bot's tabs and disconnects. * **Dialogs and downloads still work** on Chrome's own context, which the bot cannot create with its usual options. """ import io import os import sys import contextlib import time import socket import pathlib import tempfile import subprocess ROOT = pathlib.Path(__file__).resolve().parent.parent sys.path.insert(0, str(ROOT / "src")) from ecw_bot import settings as cfg # noqa: E402 from ecw_bot.core import browser # noqa: E402 ECW = "https://app1.ecwcloud.com" def test_cdp_url_must_be_loopback(): for ok in ("http://127.0.0.1:9222", "http://localhost:9222", "http://127.0.0.1:9222/"): assert cfg.cdp_url_problem(ok) is None, ok for bad in ("http://10.100.0.23:9222", "http://0.0.0.0:9222", "http://example.com:9222", "127.0.0.1:9222", "http://127.0.0.1", "ws://127.0.0.1:9222"): assert cfg.cdp_url_problem(bad), bad def test_ecw_sites_are_the_registrable_domain(): assert browser.ecw_sites(["https://abc.ecwcloud.com/mobiledoc/x", "https://abc.ecwcloud.com", ""]) == {"ecwcloud.com"} assert browser.is_ecw_host("abc.ecwcloud.com", {"ecwcloud.com"}) assert browser.is_ecw_host(".ecwcloud.com", {"ecwcloud.com"}) assert not browser.is_ecw_host("claude.ai", {"ecwcloud.com"}) assert not browser.is_ecw_host("notecwcloud.com", {"ecwcloud.com"}) def test_start_script_command(): sys.path.insert(0, str(ROOT / "tools")) import start_ecw_chrome as s assert s.chrome_command is browser.chrome_command, "one command, two users" cmd = s.chrome_command("chrome.exe", 9222, r"D:\EcwChrome", {"server": "http://10.0.0.1:8091"}) assert "--remote-debugging-port=9222" in cmd, cmd assert r"--user-data-dir=D:\EcwChrome" in cmd, cmd assert "--proxy-server=http://10.0.0.1:8091" in cmd, cmd assert any(a.startswith("--proxy-bypass-list=") and "claude.ai" in a for a in cmd), cmd assert all(a in cmd for a in browser.CHROME_ARGS), cmd assert not any(a.startswith("--proxy") for a in s.chrome_command("c", 9222, "p", None)) def test_a_chrome_with_the_proxy_passes_and_one_without_is_refused(): saved = (browser._cmdline_on_port, cfg.PROXY_IP, cfg.PROXY_PORT) try: cfg.PROXY_IP, cfg.PROXY_PORT = "10.0.0.1", "8091" browser._cmdline_on_port = lambda port: ( "chrome.exe --remote-debugging-port=9222 " "--proxy-server=http://10.0.0.1:8091") assert browser.chrome_proxy_problem("http://127.0.0.1:9222") is None browser._cmdline_on_port = lambda port: ( "chrome.exe --remote-debugging-port=9222") assert "without the eCW proxy" in browser.chrome_proxy_problem( "http://127.0.0.1:9222") browser._cmdline_on_port = lambda port: None # cannot tell with contextlib.redirect_stdout(io.StringIO()): assert browser.chrome_proxy_problem("http://127.0.0.1:9222") \ is None cfg.PROXY_IP = "" # no proxy at all assert browser.chrome_proxy_problem("http://127.0.0.1:9222") is None finally: browser._cmdline_on_port, cfg.PROXY_IP, cfg.PROXY_PORT = saved def test_cdp_ready_false_when_nothing_listens(): assert browser.cdp_ready(f"http://127.0.0.1:{_free_port()}") is False # ── Against a real Chromium on a debug port ─────────────────────────────────── def _free_port() -> int: with socket.socket() as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def _routes(context): """Serve the fake eCW and claude.ai origins without a network.""" def handle(route): route.fulfill(status=200, content_type="text/html", body="ok") context.route(f"{ECW}/**", handle) context.route("https://claude.ai/**", handle) def _seed(pw, url): """Leave eCW and claude.ai data in the profile, as a run would.""" b = pw.chromium.connect_over_cdp(url) ctx = b.contexts[0] _routes(ctx) p = ctx.new_page() p.goto(f"{ECW}/mobiledoc/") p.evaluate("localStorage.setItem('ecw', '1')") ctx.add_cookies([ {"name": "JSESSIONID", "value": "x", "domain": "app1.ecwcloud.com", "path": "/", "secure": True}, {"name": "dev", "value": "y", "domain": ".ecwcloud.com", "path": "/", "secure": True}, {"name": "sessionKey", "value": "z", "domain": "claude.ai", "path": "/", "secure": True}, ]) p.close() b.close() def _cookie_names(ctx): return sorted(c["name"] for c in ctx.cookies()) def _ecw_local_storage(ctx): _routes(ctx) p = ctx.new_page() p.goto(f"{ECW}/mobiledoc/") val = p.evaluate("localStorage.getItem('ecw')") p.close() return val def case_attach_clears_leftover_ecw_data(pw, url): _seed(pw, url) b, ctx, page = browser.launch(pw) try: assert _cookie_names(ctx) == ["sessionKey"], _cookie_names(ctx) assert page.context is ctx # The operator's Chrome: its blank tab is theirs, never closed. assert len(ctx.pages) >= 2 and any( p.url == "about:blank" for p in ctx.pages if p is not page), \ [p.url for p in ctx.pages] finally: b.close() def case_close_clears_ecw_only_and_keeps_chrome(pw, url): b, ctx, page = browser.launch(pw) _routes(ctx) page.goto(f"{ECW}/mobiledoc/") page.evaluate("localStorage.setItem('ecw', '1')") ctx.add_cookies([ {"name": "JSESSIONID", "value": "x", "domain": "app1.ecwcloud.com", "path": "/", "secure": True}, {"name": "sessionKey", "value": "z", "domain": "claude.ai", "path": "/", "secure": True}, ]) popup = ctx.new_page() # a window eCW opened b.close() assert popup.is_closed() and page.is_closed() assert browser.cdp_ready(url), "Chrome went away with the bot" look = pw.chromium.connect_over_cdp(url) try: ctx2 = look.contexts[0] assert _cookie_names(ctx2) == ["sessionKey"], _cookie_names(ctx2) assert _ecw_local_storage(ctx2) is None assert len(ctx2.pages) >= 1, "the operator's own tab was closed" finally: look.close() def case_dialog_handler_is_wired(pw, url): # The same handler as a launched browser gets, on the bot's tab and on # every tab eCW opens. A stand-in records it: what the real handler does # with a dialog is page_utils' business, not this file's. seen = [] def record(dialog): seen.append(dialog.type) dialog.accept() # Kept in place until the end: a new tab looks the handler up when it # opens, not when the bot attached. real, browser.auto_dialog = browser.auto_dialog, record b = None try: b, ctx, page = browser.launch(pw) page.set_content("

x

") page.evaluate("alert('one')") popup = ctx.new_page() popup.set_content("

y

") popup.evaluate("alert('two')") assert seen.count("alert") >= 2, seen finally: if b: b.close() browser.auto_dialog = real def case_download_over_cdp(pw, url): b, ctx, page = browser.launch(pw) try: page.set_content('get') with page.expect_download(timeout=15_000) as info: page.click("#d") out = pathlib.Path(tempfile.mkdtemp()) / "r.csv" info.value.save_as(out) assert out.read_text() == "a,b\n1,2", out.read_text() finally: b.close() CASES = [ case_attach_clears_leftover_ecw_data, case_close_clears_ecw_only_and_keeps_chrome, case_dialog_handler_is_wired, case_download_over_cdp, ] def _start_chrome(exe, port, profile): proc = subprocess.Popen( [exe, f"--remote-debugging-port={port}", f"--user-data-dir={profile}", "--headless=new", "--no-first-run", "--no-default-browser-check", "about:blank"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) url = f"http://127.0.0.1:{port}" for _ in range(60): if browser.cdp_ready(url): return proc, url time.sleep(0.25) proc.kill() raise RuntimeError("Chromium did not open its debug port") def _run_live() -> int: from playwright.sync_api import sync_playwright failures = 0 base = "D:/claude_tmp" if os.path.isdir("D:/claude_tmp") else None profile = tempfile.mkdtemp(prefix="cdp_profile_", dir=base) saved = (cfg.BROWSER_CDP_URL, cfg.ECW_URL, cfg.ECW_LANDING_URL) with sync_playwright() as pw: proc, url = _start_chrome(pw.chromium.executable_path, _free_port(), profile) cfg.BROWSER_CDP_URL, cfg.ECW_URL, cfg.ECW_LANDING_URL = ( url, f"{ECW}/mobiledoc/jsp/webemr/login/newLogin.jsp", ECW) try: for fn in CASES: name = fn.__name__.replace("case_", "").replace("_", " ") try: fn(pw, url) print(f" ok {name}") except AssertionError as exc: print(f" FAIL {name}: {exc}") failures += 1 except Exception as exc: print(f" ERROR {name}: {type(exc).__name__}: {exc}") failures += 1 finally: cfg.BROWSER_CDP_URL, cfg.ECW_URL, cfg.ECW_LANDING_URL = saved proc.kill() proc.wait() return failures def _pid_on_port(port: int): """(pid, parent pid) of the process listening on `port`.""" ps = (f"$c = Get-NetTCPConnection -LocalPort {port} -State Listen " f"-ErrorAction SilentlyContinue | Select-Object -First 1; " f"if ($c) {{ $p = Get-CimInstance Win32_Process -Filter " f"\"ProcessId=$($c.OwningProcess)\"; " f"\"$($p.ProcessId) $($p.ParentProcessId)\" }}") out = subprocess.run(["powershell", "-NoProfile", "-Command", ps], capture_output=True, text=True).stdout.split() return (int(out[0]), int(out[1])) if len(out) == 2 else (None, None) def _gone(url, seconds=15): end = time.monotonic() + seconds while time.monotonic() < end: if not browser.cdp_ready(url): return True time.sleep(0.25) return False def _run_bot_starts_chrome() -> int: """Start pressed, no Chrome on the port: the bot opens it, outside its own process tree (a Stop's taskkill /T must not take it), attaches, and closes it at the end — only the one it opened.""" from playwright.sync_api import sync_playwright failures = 0 base = "D:/claude_tmp" if os.path.isdir("D:/claude_tmp") else None saved = (cfg.BROWSER_CDP_URL, cfg.CHROME_PATH, cfg.CDP_PROFILE_DIR, cfg.PROXY_IP, cfg.CDP_CLOSE_AFTER_RUN, cfg.ECW_URL, cfg.ECW_LANDING_URL, cfg.PROXY_PORT) with sync_playwright() as pw: try: cfg.CHROME_PATH = pw.chromium.executable_path cfg.PROXY_IP = "" cfg.ECW_URL = cfg.ECW_LANDING_URL = ECW for close_after in (True, False): name = ("the bot opens Chrome, outside its tree, and " + ("closes it after the run" if close_after else "leaves it with CDP_CLOSE_AFTER_RUN=0")) port = _free_port() url = f"http://127.0.0.1:{port}" cfg.BROWSER_CDP_URL = url cfg.CDP_PROFILE_DIR = tempfile.mkdtemp(prefix="cdp_bot_", dir=base) cfg.CDP_CLOSE_AFTER_RUN = close_after pid = None try: assert browser.start_chrome(url, ["--headless=new"]) assert browser.cdp_ready(url) pid, parent = _pid_on_port(port) assert pid and parent != os.getpid(), (pid, parent) # Opened with no proxy: fine while none is configured, # refused once the eCW proxy is (the 2026-10-22 case). assert browser.chrome_proxy_problem(url) is None cfg.PROXY_IP, cfg.PROXY_PORT = "10.0.0.1", "8091" problem = browser.chrome_proxy_problem(url) cfg.PROXY_IP, cfg.PROXY_PORT = "", "" assert problem and "without the eCW proxy" in problem, \ problem b, ctx, page = browser.launch(pw) # The Chrome the run opened shows eCW only: its own # empty first tab is closed (operator, 2026-10-23). assert ctx.pages == [page], [p.url for p in ctx.pages] page.set_content("

x

") b.close() if close_after: assert _gone(url), "Chrome still answers" else: time.sleep(1) assert browser.cdp_ready(url), "Chrome was closed" print(f" ok {name}") except Exception as exc: print(f" FAIL {name}: {type(exc).__name__}: {exc}") failures += 1 finally: if browser.cdp_ready(url): pid, _ = _pid_on_port(port) if pid: subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True) finally: (cfg.BROWSER_CDP_URL, cfg.CHROME_PATH, cfg.CDP_PROFILE_DIR, cfg.PROXY_IP, cfg.CDP_CLOSE_AFTER_RUN, cfg.ECW_URL, cfg.ECW_LANDING_URL, cfg.PROXY_PORT) = saved return failures def main() -> int: plain = [test_cdp_url_must_be_loopback, test_ecw_sites_are_the_registrable_domain, test_start_script_command, test_a_chrome_with_the_proxy_passes_and_one_without_is_refused, test_cdp_ready_false_when_nothing_listens] failures = 0 for fn in plain: name = fn.__name__.replace("test_", "").replace("_", " ") try: fn() print(f" ok {name}") except AssertionError as exc: print(f" FAIL {name}: {exc}") failures += 1 failures += _run_live() failures += _run_bot_starts_chrome() total = len(plain) + len(CASES) + 2 print(f"\n{total - failures}/{total} passed") return 1 if failures else 0 if __name__ == "__main__": sys.exit(main())