1065 lines
52 KiB
Python
1065 lines
52 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
import os
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from browser_worker import (
|
|
BrowserWorker, BrowserRuntimeWorker, ElementMetadata, WorkerError, BrowserOperationError,
|
|
RealBrowserBackend, MAX_LINE_BYTES, MAX_READ_TEXT, process_request, run_stream,
|
|
validate_cdp_url, main,
|
|
)
|
|
|
|
|
|
class StartupDiagnosticsTests(unittest.TestCase):
|
|
def test_missing_approved_tab_reports_origins_without_private_url_components(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend.allowed_origins = {"https://fixture.test"}
|
|
backend._helpers = Mock()
|
|
backend._helpers.list_tabs.return_value = [
|
|
{"url": "https://login.example.test/private-path?token=private-token#private-fragment"},
|
|
{"url": "chrome://welcome"},
|
|
{"url": "https://user:private-password@login.example.test/"},
|
|
]
|
|
with self.assertRaises(WorkerError) as caught:
|
|
backend._attach_approved_tab()
|
|
message = str(caught.exception)
|
|
self.assertIn("found 0", message)
|
|
self.assertIn('open web origins: ["https://login.example.test"]', message)
|
|
self.assertNotIn("private-", message)
|
|
backend._helpers.switch_tab.assert_not_called()
|
|
|
|
def test_startup_errors_use_safe_ipc_and_remain_closable(self):
|
|
for error, expected in [
|
|
(WorkerError("expected exactly one existing approved-origin browser tab; found 0"), "found 0"),
|
|
(RuntimeError("private-startup-value"), "browser worker initialization failed"),
|
|
]:
|
|
with self.subTest(error=type(error).__name__):
|
|
reader = io.BytesIO(b'{"id":1,"op":"observe"}\n{"id":2,"op":"close"}\n')
|
|
writer = io.BytesIO()
|
|
with patch("browser_worker.RealBrowserBackend", side_effect=error), \
|
|
patch("browser_worker.signal.signal"), \
|
|
patch("browser_worker.run_stream", side_effect=lambda worker: run_stream(worker, reader, writer)):
|
|
self.assertEqual(main(["--cdp-url", "http://127.0.0.1:9222", "--allowed-origins", '["http://fixture.test"]']), 0)
|
|
messages = [json.loads(line) for line in writer.getvalue().splitlines()]
|
|
self.assertEqual(messages[0]["id"], 1)
|
|
self.assertFalse(messages[0]["ok"])
|
|
self.assertIn(expected, messages[0]["error"])
|
|
self.assertTrue(messages[1]["ok"])
|
|
self.assertNotIn(b"private-startup-value", writer.getvalue())
|
|
|
|
|
|
class FakeBackend:
|
|
def __init__(self):
|
|
self.url = "http://fixture.test/form"
|
|
self.target = "tab-1"
|
|
self.frame = "frame-1"
|
|
self.loader = "loader-1"
|
|
self.elements = {
|
|
1: ElementMetadata(1, "input", "text", "textbox", "Email", True, False),
|
|
2: ElementMetadata(2, "input", "password", "textbox", "Password", True, False),
|
|
3: ElementMetadata(3, "button", "", "button", "Submit", True, False),
|
|
4: ElementMetadata(4, "input", "text", "textbox", "Disabled", True, True),
|
|
}
|
|
self.values = {1: ""}
|
|
self.nodes = [
|
|
{"backendDOMNodeId": 1, "role": {"value": "textbox"}, "name": {"value": "Email"}},
|
|
{"backendDOMNodeId": 3, "role": {"value": "button"}, "name": {"value": "Submit"}},
|
|
{"backendDOMNodeId": 4, "role": {"value": "textbox"}, "name": {"value": "Disabled"}},
|
|
# Password controls are supported; their values remain private.
|
|
{"backendDOMNodeId": 2, "role": {"value": "textbox"}, "name": {"value": "Password"}},
|
|
]
|
|
self.clicks = 0
|
|
|
|
def current_tab(self):
|
|
return {"targetId": self.target, "url": self.url, "title": "Fixture"}
|
|
|
|
def document_identity(self):
|
|
return self.frame, self.loader
|
|
|
|
def accessibility_nodes(self):
|
|
return self.nodes
|
|
|
|
def metadata(self, backend_id):
|
|
return self.elements[backend_id]
|
|
|
|
def click(self, backend_id):
|
|
self.clicks += 1
|
|
if backend_id == 3:
|
|
self.url = "http://fixture.test/done"
|
|
|
|
def fill(self, backend_id, text):
|
|
self.values[backend_id] = text
|
|
|
|
def value_matches(self, backend_id, text):
|
|
return self.values.get(backend_id) == text
|
|
|
|
def visible_text(self, text):
|
|
return text == "Finished"
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class BatchFakeBackend(FakeBackend):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.metadata_many_calls = []
|
|
self.metadata_calls = []
|
|
self.batch_result = None
|
|
|
|
def metadata(self, backend_id):
|
|
self.metadata_calls.append(backend_id)
|
|
return super().metadata(backend_id)
|
|
|
|
def metadata_many(self, backend_ids):
|
|
ids = list(backend_ids)
|
|
self.metadata_many_calls.append(ids)
|
|
if self.batch_result is not None:
|
|
return self.batch_result
|
|
return {backend_id: self.elements[backend_id] for backend_id in ids if backend_id in self.elements}
|
|
|
|
|
|
class PopupFakeBackend(FakeBackend):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.status = "Scenario fixture ready"
|
|
|
|
def click(self, backend_id):
|
|
super().click(backend_id)
|
|
if backend_id == 3:
|
|
self.status = "Popup opened"
|
|
|
|
def visible_text(self, text):
|
|
return text == self.status
|
|
|
|
|
|
class RuntimeFakeBackend(FakeBackend):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.tabs = [
|
|
{"targetId": "tab-a", "url": "http://fixture.test/form", "title": "Fixture"},
|
|
{"targetId": "tab-b", "url": "http://other.test/second", "title": "Second"},
|
|
]
|
|
self.target = "tab-a"
|
|
self.text_content = "Fixture complete Привет"
|
|
self.events = []
|
|
self._session_id = "fake-session"
|
|
self._session_counter = 0
|
|
self.switches = []
|
|
self.closed_tabs = []
|
|
|
|
def list_tabs(self):
|
|
return [dict(tab) for tab in self.tabs]
|
|
|
|
def switch_tab(self, target_id):
|
|
if not any(tab["targetId"] == target_id for tab in self.tabs):
|
|
raise RuntimeError("missing target")
|
|
self.target = target_id
|
|
self._session_counter += 1
|
|
self._session_id = f"fake-session-{self._session_counter}"
|
|
self.switches.append(target_id)
|
|
|
|
def current_tab(self):
|
|
return next(tab for tab in self.tabs if tab["targetId"] == self.target)
|
|
|
|
def visible_text_content(self):
|
|
return self.text_content
|
|
|
|
def navigate(self, url):
|
|
self.current_tab()["url"] = url
|
|
self.loader = self.loader + "-nav"
|
|
|
|
def reload(self):
|
|
self.loader = self.loader + "-reload"
|
|
|
|
def wait_ready(self, timeout):
|
|
return True
|
|
|
|
def close_tab(self, target_id):
|
|
self.closed_tabs.append(target_id)
|
|
self.tabs[:] = [tab for tab in self.tabs if tab["targetId"] != target_id]
|
|
|
|
def drain_events(self):
|
|
events, self.events = self.events, []
|
|
return events
|
|
|
|
def screenshot_bytes(self):
|
|
return b"\\x89PNG" + (b"x" * 32)
|
|
|
|
|
|
class RuntimeTests(unittest.TestCase):
|
|
def make_runtime(self, backend=None, origins="*"):
|
|
backend = backend or RuntimeFakeBackend()
|
|
return BrowserRuntimeWorker(backend, origins, session_id="runtime-test")
|
|
|
|
def test_list_tabs_does_not_fail_or_select_ambiguous_pages(self):
|
|
backend = RuntimeFakeBackend()
|
|
runtime = self.make_runtime(backend)
|
|
tabs = runtime.list_tabs()
|
|
self.assertEqual(len(tabs), 2)
|
|
self.assertFalse(any(tab["selected"] for tab in tabs))
|
|
with self.assertRaisesRegex(BrowserOperationError, "select an approved tab"):
|
|
runtime.read({})
|
|
chosen = tabs[0]["tabId"]
|
|
selected = runtime.select_tab({"tabId": chosen})
|
|
self.assertTrue(selected["tab"]["selected"])
|
|
self.assertEqual(backend.target, "tab-a" if tabs[0]["origin"] == "http://fixture.test" else "tab-b")
|
|
|
|
def test_read_is_bounded_unicode_and_has_no_form_or_hidden_values(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
backend.text_content = "Visible Привет\n" + ("é" * (MAX_READ_TEXT + 20))
|
|
runtime = self.make_runtime(backend)
|
|
result = runtime.read({})
|
|
self.assertTrue(result["visibleTextTruncated"])
|
|
self.assertEqual(len(result["visibleText"]), MAX_READ_TEXT)
|
|
self.assertIn("Привет", result["visibleText"])
|
|
self.assertNotIn("password-value", result["visibleText"])
|
|
self.assertNotIn("<input", result["visibleText"])
|
|
|
|
def test_assertion_polling_is_read_only_and_reports_timeout(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
result = runtime.assert_page({
|
|
"assertion": {"kind": "visibleText", "match": "contains", "value": "missing"},
|
|
"wait": "none",
|
|
})
|
|
self.assertEqual(result["status"], "not_satisfied")
|
|
result = runtime.assert_page({
|
|
"assertion": {"kind": "title", "match": "contains", "value": "missing"},
|
|
"wait": "bounded",
|
|
})
|
|
self.assertEqual(result["status"], "timed_out")
|
|
self.assertEqual(backend.clicks, 0)
|
|
|
|
def test_post_dispatch_failures_are_unknown_and_never_retryable(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
observed = runtime.observe()
|
|
def mutate_then_raise(_backend_id):
|
|
backend.clicks += 1
|
|
raise RuntimeError("browser evaluation failed")
|
|
backend.click = mutate_then_raise
|
|
with self.assertRaises(BrowserOperationError) as caught:
|
|
runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
|
|
self.assertEqual(caught.exception.details["mutationState"], "unknown")
|
|
self.assertEqual(caught.exception.details["retry"], "never")
|
|
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
observed = runtime.observe()
|
|
checks = [0]
|
|
def verify_then_raise(_text):
|
|
checks[0] += 1
|
|
if checks[0] == 1:
|
|
return False
|
|
raise RuntimeError("postcondition evaluation failed")
|
|
backend.visible_text = verify_then_raise
|
|
with self.assertRaises(BrowserOperationError) as caught:
|
|
runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "text", "value": "done"}})
|
|
self.assertEqual(caught.exception.details["mutationState"], "unknown")
|
|
self.assertEqual(caught.exception.details["retry"], "never")
|
|
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
observed = runtime.observe()
|
|
def cancel_after_dispatch(_backend_id):
|
|
class Cancelled(BaseException):
|
|
pass
|
|
raise Cancelled()
|
|
backend.click = cancel_after_dispatch
|
|
with self.assertRaises(BrowserOperationError) as caught:
|
|
runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
|
|
self.assertEqual(caught.exception.details["code"], "aborted")
|
|
self.assertEqual(caught.exception.details["mutationState"], "unknown")
|
|
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
def navigate_then_raise(url):
|
|
backend.current_tab()["url"] = url
|
|
raise RuntimeError("navigation acknowledgement failed")
|
|
backend.navigate = navigate_then_raise
|
|
with self.assertRaises(BrowserOperationError) as caught:
|
|
runtime.navigate({"url": "http://fixture.test/next"})
|
|
self.assertEqual(caught.exception.details["mutationState"], "unknown")
|
|
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
tab_id = runtime.list_tabs()[0]["tabId"]
|
|
original_close = backend.close_tab
|
|
def close_then_raise(target):
|
|
original_close(target)
|
|
raise RuntimeError("close acknowledgement failed")
|
|
backend.close_tab = close_then_raise
|
|
with self.assertRaises(BrowserOperationError) as caught:
|
|
runtime.close_tab({"tabId": tab_id})
|
|
self.assertEqual(caught.exception.details["mutationState"], "unknown")
|
|
|
|
def test_navigation_reload_change_document_without_retrying_mutation(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend, ["http://fixture.test"])
|
|
first = runtime.read({})["documentId"]
|
|
navigated = runtime.navigate({"url": "http://fixture.test/next"})
|
|
self.assertEqual(navigated["mutationState"], "verified")
|
|
self.assertNotEqual(first, navigated["documentId"])
|
|
reloaded = runtime.reload({})
|
|
self.assertNotEqual(navigated["documentId"], reloaded["documentId"])
|
|
with self.assertRaisesRegex(BrowserOperationError, "outside"):
|
|
runtime.navigate({"url": "https://not-approved.test/"})
|
|
|
|
def test_narrow_policy_is_per_dispatch_and_cannot_broaden_base(self):
|
|
backend = RuntimeFakeBackend()
|
|
runtime = self.make_runtime(backend)
|
|
narrowed = runtime.narrow_origin_policy({"allowedOrigins": ["http://fixture.test"]})
|
|
self.assertEqual(narrowed, {"allowedOrigins": ["http://fixture.test"]})
|
|
runtime.apply_context(narrowed)
|
|
self.assertEqual([tab["origin"] for tab in runtime.list_tabs()], ["http://fixture.test"])
|
|
runtime.apply_context(None)
|
|
self.assertEqual(len(runtime.list_tabs()), 2)
|
|
restricted = self.make_runtime(RuntimeFakeBackend(), ["http://fixture.test"])
|
|
with self.assertRaisesRegex(BrowserOperationError, "broaden"):
|
|
restricted.apply_context({"allowedOrigins": ["https://not-in-base.test"]})
|
|
|
|
def test_popup_handle_is_owned_but_wait_does_not_switch(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
opener = runtime.list_tabs()[0]["tabId"]
|
|
backend.tabs.append({"targetId": "popup", "url": "http://fixture.test/popup", "title": "Popup", "openerTargetId": "tab-a"})
|
|
popup = runtime.wait_popup({"openerTabId": opener})
|
|
self.assertEqual(popup["status"], "found")
|
|
self.assertFalse(popup["autoSwitched"])
|
|
self.assertEqual(backend.target, "tab-a")
|
|
self.assertEqual(popup["tab"]["kind"], "popup")
|
|
|
|
def test_immediate_popup_is_found_through_typed_dispatcher_sequence(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
original_click = backend.click
|
|
def click_and_open(backend_id):
|
|
original_click(backend_id)
|
|
backend.visible_text = lambda text: text == "POPUP_OPEN_REQUESTED"
|
|
backend.text_content = "POPUP_OPEN_REQUESTED"
|
|
backend.tabs.append({"targetId": "immediate-popup", "url": "http://fixture.test/popup",
|
|
"title": "Popup", "openerTargetId": "tab-a"})
|
|
backend.click = click_and_open
|
|
runtime = self.make_runtime(backend)
|
|
observed = process_request(runtime, {"id": 1, "operation": {"op": "observe"}})["result"]["result"]
|
|
acted = process_request(runtime, {"id": 2, "operation": {"op": "act", "request": {
|
|
"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "text", "value": "POPUP_OPEN_REQUESTED"},
|
|
}}})
|
|
self.assertTrue(acted["result"]["result"]["verified"])
|
|
opener = runtime.list_tabs()[0]["tabId"]
|
|
waited = process_request(runtime, {"id": 3, "operation": {"op": "waitPopup", "request": {
|
|
"openerTabId": opener,
|
|
}}})
|
|
popup = waited["result"]["result"]
|
|
self.assertEqual(popup["status"], "found")
|
|
self.assertFalse(popup["autoSwitched"])
|
|
self.assertEqual(popup["tab"]["kind"], "popup")
|
|
self.assertEqual(backend.target, "tab-a")
|
|
|
|
def test_popup_wait_excludes_baseline_unrelated_and_claimed_targets(self):
|
|
from unittest.mock import patch
|
|
with patch("browser_worker.POPUP_WAIT_SECONDS", 0.01):
|
|
backend = RuntimeFakeBackend()
|
|
baseline_popup = {"targetId": "baseline-popup", "url": "http://fixture.test/old-popup",
|
|
"title": "Old", "openerTargetId": "tab-a"}
|
|
# Deliberately put the popup before its opener in the CDP inventory.
|
|
backend.tabs = [baseline_popup, backend.tabs[0], backend.tabs[1]]
|
|
runtime = self.make_runtime(backend)
|
|
opener = next(tab for tab in runtime.list_tabs() if tab["url"].endswith("/form"))["tabId"]
|
|
self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")
|
|
|
|
backend.tabs.append({"targetId": "unrelated-popup", "url": "http://fixture.test/unrelated",
|
|
"title": "Unrelated", "openerTargetId": "tab-b"})
|
|
self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")
|
|
|
|
backend.tabs.append({"targetId": "claimed-popup", "url": "http://fixture.test/claimed",
|
|
"title": "Claimed", "openerTargetId": "tab-a"})
|
|
found = runtime.wait_popup({"openerTabId": opener})
|
|
self.assertEqual(found["status"], "found")
|
|
self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")
|
|
|
|
def test_closed_tab_handle_is_invalidated(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
tab_id = runtime.list_tabs()[0]["tabId"]
|
|
self.assertEqual(runtime.close_tab({"tabId": tab_id})["mutationState"], "verified")
|
|
with self.assertRaisesRegex(BrowserOperationError, "stale|unavailable"):
|
|
runtime.select_tab({"tabId": tab_id})
|
|
|
|
def test_transport_wrapper_context_restores_base_and_uses_exact_result_shape(self):
|
|
backend = RuntimeFakeBackend()
|
|
runtime = self.make_runtime(backend)
|
|
requests = (
|
|
b'{"id":1,"operation":{"op":"narrowOriginPolicy","request":{"allowedOrigins":["http://fixture.test"]}}}\n'
|
|
b'{"id":2,"operation":{"op":"listTabs"},"context":{"allowedOrigins":["http://fixture.test"]}}\n'
|
|
b'{"id":3,"operation":{"op":"listTabs"}}\n'
|
|
)
|
|
output = io.BytesIO()
|
|
run_stream(runtime, io.BytesIO(requests), output)
|
|
rows = [json.loads(line) for line in output.getvalue().splitlines()]
|
|
self.assertEqual(rows[0]["result"]["op"], "narrowOriginPolicy")
|
|
self.assertEqual(rows[0]["result"]["result"], {"allowedOrigins": ["http://fixture.test"]})
|
|
self.assertEqual(len(rows[1]["result"]["result"]), 1)
|
|
self.assertEqual(len(rows[2]["result"]["result"]), 2)
|
|
|
|
def test_real_screenshot_temp_is_private_bounded_and_cleaned_on_failure(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
# Use an isolated directory rather than any user temp location.
|
|
import tempfile
|
|
root = Path(tempfile.mkdtemp(prefix="fb-shot-test-"))
|
|
shot_dir = root / "shots"
|
|
shot_dir.mkdir(mode=0o700)
|
|
backend._screenshot_dir = shot_dir
|
|
backend._helpers = Mock()
|
|
observed = []
|
|
def capture(path, **_kwargs):
|
|
candidate = Path(path)
|
|
observed.append((candidate, candidate.stat().st_mode & 0o777))
|
|
candidate.write_bytes(b"png")
|
|
backend._helpers.capture_screenshot.side_effect = capture
|
|
self.assertEqual(backend.screenshot_bytes(), b"png")
|
|
self.assertEqual(observed[0][0].parent, shot_dir)
|
|
self.assertEqual(observed[0][1], 0o600)
|
|
self.assertEqual(list(shot_dir.iterdir()), [])
|
|
backend._helpers.capture_screenshot.side_effect = RuntimeError("capture failed")
|
|
with self.assertRaises(RuntimeError):
|
|
backend.screenshot_bytes()
|
|
self.assertEqual(list(shot_dir.iterdir()), [])
|
|
import shutil
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
|
|
def test_screenshot_transport_is_capped_and_marked_truncated(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
backend.screenshot_bytes = lambda: b"x" * (8 * 1024 * 1024 + 1)
|
|
runtime = self.make_runtime(backend)
|
|
evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
|
|
self.assertTrue(evidence["screenshot"]["truncated"])
|
|
self.assertEqual(evidence["screenshot"]["byteLength"], 8 * 1024 * 1024)
|
|
|
|
def test_evidence_filters_third_party_network_failures_by_effective_policy(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend, ["http://fixture.test"])
|
|
backend.events = [
|
|
{"method": "Runtime.consoleAPICalled", "session_id": backend._session_id,
|
|
"params": {"type": "error", "args": [{"value": "https://third-party.test/console?token=bad"}]}},
|
|
{"method": "Network.loadingFailed", "session_id": backend._session_id,
|
|
"params": {"requestId": "third", "url": "https://third-party.test/fail?token=bad"}},
|
|
{"method": "Network.loadingFailed", "session_id": backend._session_id,
|
|
"params": {"requestId": "local", "url": "http://fixture.test/fail?token=bad"}},
|
|
]
|
|
evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
|
|
self.assertEqual([item["origin"] for item in evidence["failedRequests"]], ["http://fixture.test"])
|
|
self.assertNotIn("third-party.test", json.dumps(evidence["console"]))
|
|
|
|
def test_evidence_does_not_replay_buffered_history_after_tab_switches(self):
|
|
backend = RuntimeFakeBackend()
|
|
runtime = self.make_runtime(backend)
|
|
tabs = runtime.list_tabs()
|
|
runtime.select_tab({"tabId": tabs[0]["tabId"]})
|
|
session_a = backend._session_id
|
|
backend.events = [{"method": "Network.loadingFailed", "session_id": session_a,
|
|
"params": {"requestId": "a", "url": "http://fixture.test/a"}}]
|
|
first = runtime.capture_evidence({"runId": "run-a", "reason": "requested"})
|
|
self.assertEqual([item["origin"] for item in first["failedRequests"]], ["http://fixture.test"])
|
|
|
|
runtime.select_tab({"tabId": tabs[1]["tabId"]})
|
|
second = runtime.capture_evidence({"runId": "run-b", "reason": "requested"})
|
|
self.assertEqual(second["failedRequests"], [])
|
|
|
|
# Returning to the target creates a new CDP session; old A history is
|
|
# still retained privately but must not be invented as current evidence.
|
|
runtime.select_tab({"tabId": tabs[0]["tabId"]})
|
|
third = runtime.capture_evidence({"runId": "run-a2", "reason": "requested"})
|
|
self.assertEqual(third["failedRequests"], [])
|
|
|
|
def test_evidence_drops_events_from_prior_tab_sessions(self):
|
|
backend = RuntimeFakeBackend()
|
|
runtime = self.make_runtime(backend)
|
|
tabs = runtime.list_tabs()
|
|
runtime.select_tab({"tabId": tabs[0]["tabId"]})
|
|
prior_session = backend._session_id
|
|
runtime.select_tab({"tabId": tabs[1]["tabId"]})
|
|
current_session = backend._session_id
|
|
backend.events = [
|
|
{"method": "Network.loadingFailed", "session_id": prior_session,
|
|
"params": {"requestId": "old", "url": "http://fixture.test/old?secret=bad"}},
|
|
{"method": "Network.loadingFailed", "session_id": current_session,
|
|
"params": {"requestId": "new", "url": "http://other.test/new?secret=bad"}},
|
|
{"method": "Network.loadingFailed", "params": {"requestId": "unknown", "url": "http://other.test/unknown"}},
|
|
]
|
|
evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
|
|
self.assertEqual([item["origin"] for item in evidence["failedRequests"]], ["http://other.test"])
|
|
|
|
def test_protocol_dispatch_returns_structured_safe_errors(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0], backend.tabs[1]]
|
|
runtime = self.make_runtime(backend)
|
|
output = io.BytesIO()
|
|
run_stream(runtime, io.BytesIO(b'{"id":9,"operation":{"op":"read","request":{}}}\n'), output)
|
|
response = json.loads(output.getvalue())
|
|
self.assertFalse(response["ok"])
|
|
self.assertEqual(response["details"]["code"], "ambiguous_tab")
|
|
self.assertEqual(response["details"]["operation"], "read")
|
|
self.assertEqual(response["details"]["mutationState"], "not_attempted")
|
|
self.assertNotIn("targetId", json.dumps(response))
|
|
|
|
def test_evidence_bounds_redacts_and_returns_no_caller_path(self):
|
|
backend = RuntimeFakeBackend()
|
|
backend.tabs = [backend.tabs[0]]
|
|
runtime = self.make_runtime(backend)
|
|
backend.events = [
|
|
{"method": "Runtime.consoleAPICalled", "session_id": backend._session_id, "params": {"type": "error", "args": [{"value": "token=private https://fixture.test/x?secret=bad"}]}},
|
|
{"method": "Network.loadingFailed", "session_id": backend._session_id, "params": {"requestId": "r", "url": "http://fixture.test/fail?token=bad"}},
|
|
]
|
|
evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
|
|
self.assertEqual(set(evidence["screenshot"]), {"kind", "contentType", "encoding", "dataBase64", "byteLength", "truncated"})
|
|
self.assertEqual(set(evidence["diagnostics"]), {"kind", "contentType", "encoding", "dataBase64", "byteLength", "truncated"})
|
|
self.assertEqual(evidence["screenshot"]["encoding"], "base64")
|
|
self.assertEqual(evidence["diagnostics"]["encoding"], "base64")
|
|
self.assertFalse(evidence["screenshot"]["truncated"])
|
|
self.assertNotIn("path", evidence)
|
|
self.assertNotIn("private", json.dumps(evidence))
|
|
self.assertNotIn("?token=bad", json.dumps(evidence))
|
|
self.assertEqual(evidence["failedRequests"][0]["origin"], "http://fixture.test")
|
|
|
|
|
|
class WorkerTests(unittest.TestCase):
|
|
def test_unrestricted_policy_accepts_any_http_origin_but_not_internal_urls(self):
|
|
backend = FakeBackend()
|
|
backend.url = "https://other.example/path"
|
|
worker = BrowserWorker(backend, "*")
|
|
observation = worker.observe()
|
|
self.assertEqual(observation["url"], "https://other.example/path")
|
|
for url in ["file:///tmp/private", "chrome://settings", "about:blank",
|
|
"https://user:private-password@example.test/", "https://*.example.test/"]:
|
|
with self.subTest(url=url):
|
|
backend.url = url
|
|
# Leaving policy stops a worker permanently; use a fresh one
|
|
# to exercise each unsupported URL rather than its stopped fuse.
|
|
worker = BrowserWorker(backend, "*")
|
|
with self.assertRaisesRegex(WorkerError, "outside"):
|
|
worker.observe()
|
|
|
|
def test_observe_bounds_candidates_and_does_not_return_values(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
self.assertEqual(observation["url"], "http://fixture.test/form")
|
|
self.assertEqual([c["name"] for c in observation["candidates"]], ["Email", "Submit", "Password"])
|
|
self.assertNotIn("values", json.dumps(observation))
|
|
self.assertNotIn("Disabled", json.dumps(observation))
|
|
|
|
def test_observe_batches_dom_metadata_but_keeps_full_ax_roles_and_names(self):
|
|
backend = BatchFakeBackend()
|
|
backend.elements[1] = replace(backend.elements[1], name="DOM Email", ax_role="textbox", ax_name="AX Email")
|
|
backend.elements[3] = replace(backend.elements[3], name="DOM Submit", ax_role="button", ax_name="AX Submit")
|
|
backend.nodes[0]["name"] = {"value": "AX Email old-test-secret"}
|
|
backend.nodes[1]["name"] = {"value": "AX Submit"}
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
|
|
observation = worker.observe()
|
|
|
|
self.assertEqual(backend.metadata_many_calls, [[1, 3, 4, 2]])
|
|
self.assertEqual(backend.metadata_calls, [])
|
|
self.assertEqual([candidate["name"] for candidate in observation["candidates"]],
|
|
["DOM Email", "AX Submit", "Password"])
|
|
self.assertEqual([candidate["role"] for candidate in observation["candidates"]],
|
|
["textbox", "button", "textbox"])
|
|
self.assertNotIn("old-test-secret", json.dumps(observation))
|
|
|
|
result = worker.act({
|
|
"observationId": observation["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"},
|
|
})
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(backend.metadata_calls, [3])
|
|
|
|
def test_observe_falls_back_for_missing_batch_metadata(self):
|
|
backend = BatchFakeBackend()
|
|
backend.batch_result = {1: backend.elements[1]}
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
|
|
observation = worker.observe()
|
|
|
|
self.assertEqual([candidate["name"] for candidate in observation["candidates"]],
|
|
["Email", "Submit", "Password"])
|
|
self.assertEqual(backend.metadata_calls, [3, 4, 2])
|
|
|
|
def test_observe_batches_in_bounded_chunks_and_preserves_order_and_deduplication(self):
|
|
backend = BatchFakeBackend()
|
|
backend.elements = {
|
|
index: ElementMetadata(index, "button", "", "button", f"Button {index}", True, False,
|
|
ax_role="button", ax_name=f"Button {index}")
|
|
for index in range(1, 34)
|
|
}
|
|
backend.nodes = [
|
|
{"backendDOMNodeId": index, "role": {"value": "button"}, "name": {"value": f"AX {index}"}}
|
|
for index in range(1, 34)
|
|
]
|
|
backend.nodes.insert(10, dict(backend.nodes[9]))
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
|
|
observation = worker.observe()
|
|
|
|
self.assertEqual([len(chunk) for chunk in backend.metadata_many_calls], [16, 16, 1])
|
|
self.assertEqual([candidate["name"] for candidate in observation["candidates"]],
|
|
[f"AX {index}" for index in range(1, 34)])
|
|
self.assertEqual(len(observation["candidates"]), 33)
|
|
|
|
def test_observe_skips_detached_nodes_and_keeps_candidate_limit(self):
|
|
backend = BatchFakeBackend()
|
|
backend.elements = {
|
|
index: ElementMetadata(index, "button", "", "button", f"Button {index}", True, False,
|
|
connected=index != 10, ax_role="button", ax_name=f"Button {index}")
|
|
for index in range(1, 63)
|
|
}
|
|
backend.nodes = [
|
|
{"backendDOMNodeId": index, "role": {"value": "button"}, "name": {"value": f"AX {index}"}}
|
|
for index in range(1, 63)
|
|
]
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
|
|
observation = worker.observe()
|
|
|
|
self.assertEqual(len(observation["candidates"]), 60)
|
|
self.assertNotIn("AX 10", json.dumps(observation))
|
|
self.assertTrue(all(len(chunk) <= 16 for chunk in backend.metadata_many_calls))
|
|
|
|
def test_password_fill_uses_literal_without_exposing_value_in_metadata_or_result(self):
|
|
backend = FakeBackend()
|
|
backend.values[2] = "old-test-secret"
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
password = next(c for c in observation["candidates"] if c["name"] == "Password")
|
|
self.assertEqual(password["actions"], ["fill"])
|
|
self.assertNotIn("old-test-secret", json.dumps(observation))
|
|
result = worker.act({"observationId": observation["observationId"], "candidateId": password["id"],
|
|
"action": "fill", "text": "new-test-secret"})
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(backend.values[2], "new-test-secret")
|
|
self.assertNotIn("new-test-secret", json.dumps(result))
|
|
self.assertNotIn("new-test-secret", json.dumps(worker.observe()))
|
|
|
|
def test_fill_verifies_literal_and_consumes_observation(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
result = worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-0",
|
|
"action": "fill",
|
|
"text": "literal@example.test",
|
|
})
|
|
self.assertEqual(result["verified"], True)
|
|
self.assertEqual(backend.values[1], "literal@example.test")
|
|
with self.assertRaisesRegex(WorkerError, "missing|consumed"):
|
|
worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-0",
|
|
"action": "fill",
|
|
"text": "retry",
|
|
})
|
|
|
|
def test_mutation_invalidates_all_previous_observations(self):
|
|
worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
|
|
old = worker.observe()
|
|
new = worker.observe()
|
|
worker.act({"observationId": new["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada"})
|
|
with self.assertRaisesRegex(WorkerError, "missing|consumed"):
|
|
worker.act({"observationId": old["observationId"], "candidateId": "c-0", "action": "fill", "text": "replay"})
|
|
|
|
def test_click_rejects_postcondition_that_was_already_satisfied(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
with self.assertRaisesRegex(WorkerError, "already satisfied"):
|
|
worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-1",
|
|
"action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/form"},
|
|
})
|
|
self.assertEqual(backend.clicks, 0)
|
|
|
|
def test_popup_fixture_postcondition_is_false_before_click_and_true_after(self):
|
|
backend = PopupFakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
self.assertFalse(backend.visible_text("Popup opened"))
|
|
result = worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-1",
|
|
"action": "click",
|
|
"postcondition": {"kind": "text", "value": "Popup opened"},
|
|
})
|
|
self.assertTrue(result["verified"])
|
|
self.assertTrue(backend.visible_text("Popup opened"))
|
|
self.assertEqual(backend.clicks, 1)
|
|
|
|
def test_click_requires_postcondition_and_does_not_retry(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
result = worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-1",
|
|
"action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"},
|
|
})
|
|
self.assertTrue(result["verified"])
|
|
self.assertEqual(backend.clicks, 1)
|
|
|
|
def test_stale_document_is_rejected_before_mutation(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
backend.loader = "new-loader"
|
|
with self.assertRaisesRegex(WorkerError, "stale"):
|
|
worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-1",
|
|
"action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"},
|
|
})
|
|
self.assertEqual(backend.clicks, 0)
|
|
|
|
def test_outside_redirect_blocks_worker(self):
|
|
backend = FakeBackend()
|
|
original_click = backend.click
|
|
|
|
def redirect(backend_id):
|
|
original_click(backend_id)
|
|
backend.url = "https://not-approved.test/"
|
|
|
|
backend.click = redirect
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
with self.assertRaisesRegex(WorkerError, "outside"):
|
|
worker.act({
|
|
"observationId": observation["observationId"],
|
|
"candidateId": "c-1",
|
|
"action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"},
|
|
})
|
|
with self.assertRaisesRegex(WorkerError, "stopped"):
|
|
worker.observe()
|
|
|
|
def test_jsonl_protocol_returns_bounded_errors(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
source = io.BytesIO(b'{"id":1,"op":"observe"}\nnot-json\n')
|
|
output = io.BytesIO()
|
|
run_stream(worker, source, output)
|
|
rows = [json.loads(line) for line in output.getvalue().splitlines()]
|
|
self.assertEqual(rows[0]["ok"], True)
|
|
self.assertEqual(rows[1]["ok"], False)
|
|
self.assertEqual(rows[1]["error"], "browser operation failed; no action was retried")
|
|
|
|
def test_process_request_rejects_extra_protocol_fields(self):
|
|
worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
|
|
with self.assertRaisesRegex(WorkerError, "malformed"):
|
|
process_request(worker, {"id": 1, "op": "observe", "extra": True})
|
|
|
|
|
|
class BridgeRegressionTests(unittest.TestCase):
|
|
def test_loopback_endpoint_contract(self):
|
|
for url in ["http://127.0.0.1:9222", "http://localhost:9222", "http://[::1]:9222"]:
|
|
self.assertEqual(validate_cdp_url(url), url)
|
|
for url in ["https://localhost:9222", "http://example.com:9222", "http://localhost:9222/path", "http://localhost:9222?key=x"]:
|
|
with self.assertRaises(ValueError):
|
|
validate_cdp_url(url)
|
|
|
|
def test_changed_candidate_role_or_name_fails_before_click(self):
|
|
for changes in [{"role": "link"}, {"name": "Delete"}, {"visible": False}, {"main_frame": False}]:
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
obs = worker.observe()
|
|
backend.elements[3] = replace(backend.elements[3], **changes)
|
|
with self.assertRaises(WorkerError):
|
|
worker.act({"observationId": obs["observationId"], "candidateId": "c-1", "action": "click",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
|
|
self.assertEqual(backend.clicks, 0)
|
|
|
|
def test_fill_optional_condition_and_post_fill_navigation_guard(self):
|
|
backend = FakeBackend()
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
obs = worker.observe()
|
|
result = worker.act({"observationId": obs["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada",
|
|
"postcondition": {"kind": "text", "value": "Finished"}})
|
|
self.assertTrue(result["verified"])
|
|
obs = worker.observe()
|
|
original = backend.fill
|
|
def redirected(node, text):
|
|
original(node, text)
|
|
backend.url = "https://outside.test"
|
|
backend.fill = redirected
|
|
with self.assertRaisesRegex(WorkerError, "outside"):
|
|
worker.act({"observationId": obs["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada"})
|
|
|
|
def test_fill_url_postcondition_reuses_post_action_tab_state(self):
|
|
backend = FakeBackend()
|
|
original_fill = backend.fill
|
|
def fill_and_navigate(node, text):
|
|
original_fill(node, text)
|
|
backend.url = "http://fixture.test/done"
|
|
backend.fill = fill_and_navigate
|
|
current_tab = backend.current_tab
|
|
backend.current_tab = Mock(side_effect=current_tab)
|
|
worker = BrowserWorker(backend, ["http://fixture.test"])
|
|
observation = worker.observe()
|
|
result = worker.act({
|
|
"observationId": observation["observationId"], "candidateId": "c-0",
|
|
"action": "fill", "text": "Ada",
|
|
"postcondition": {"kind": "url", "value": "http://fixture.test/done"},
|
|
})
|
|
self.assertTrue(result["verified"])
|
|
# observe, fresh pre-mutation revalidation, and one post-action read.
|
|
self.assertEqual(backend.current_tab.call_count, 3)
|
|
|
|
def test_oversized_input_and_sensitive_error_are_not_echoed(self):
|
|
worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
|
|
out = io.BytesIO()
|
|
self.assertEqual(run_stream(worker, io.BytesIO(b"x" * (MAX_LINE_BYTES + 2)), out), 1)
|
|
self.assertLess(len(out.getvalue()), 200)
|
|
worker.backend.current_tab = Mock(side_effect=RuntimeError("private-password"))
|
|
out = io.BytesIO()
|
|
run_stream(worker, io.BytesIO(b'{"id":1,"op":"observe"}\n'), out)
|
|
self.assertNotIn(b"private-password", out.getvalue())
|
|
|
|
def test_version_guard_precedes_import_or_daemon_spawn(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
with patch("importlib.metadata.version", return_value="0.1.8"), patch("browser_worker.subprocess.Popen") as spawn:
|
|
with self.assertRaisesRegex(WorkerError, "0.1.13 is required"):
|
|
backend._load_harness()
|
|
spawn.assert_not_called()
|
|
|
|
def test_real_backend_list_tabs_uses_one_inventory_with_helper_filtering_and_openers(self):
|
|
helper = Mock()
|
|
helper.cdp.return_value = {"targetInfos": [
|
|
{"type": "page", "targetId": "page", "url": "http://fixture.test/form", "title": "Fixture",
|
|
"openerId": "opener"},
|
|
{"type": "page", "targetId": "internal", "url": "chrome://settings", "title": "Settings"},
|
|
{"type": "page", "targetId": "devtools", "url": "devtools://devtools", "title": "DevTools"},
|
|
{"type": "page", "targetId": "placeholder", "url": "about:blank#startup", "title": "Starting agent 1"},
|
|
{"type": "service_worker", "targetId": "worker", "url": "http://fixture.test/sw", "title": "Worker"},
|
|
{"type": "page", "targetId": "plain", "url": "https://other.test/", "title": "Other", "openerId": ""},
|
|
{"type": "page", "targetId": "malformed-opener", "url": "http://fixture.test/other", "title": "Other",
|
|
"openerId": 42},
|
|
]}
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = helper
|
|
|
|
tabs = backend.list_tabs()
|
|
|
|
self.assertEqual([tab["targetId"] for tab in tabs], ["page", "plain", "malformed-opener"])
|
|
self.assertEqual(tabs[0]["openerTargetId"], "opener")
|
|
self.assertIsNone(tabs[1]["openerTargetId"])
|
|
self.assertIsNone(tabs[2]["openerTargetId"])
|
|
helper.cdp.assert_called_once_with("Target.getTargets")
|
|
helper.list_tabs.assert_not_called()
|
|
|
|
def test_real_backend_list_tabs_propagates_inventory_errors_without_fallback(self):
|
|
helper = Mock()
|
|
helper.cdp.side_effect = RuntimeError("inventory unavailable")
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = helper
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "inventory unavailable"):
|
|
backend.list_tabs()
|
|
helper.cdp.assert_called_once_with("Target.getTargets")
|
|
helper.list_tabs.assert_not_called()
|
|
|
|
def test_real_backend_switch_tab_does_not_reenable_harness_domains(self):
|
|
helper = Mock()
|
|
helper.switch_tab.return_value = "session-1"
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = helper
|
|
|
|
backend.switch_tab("target-1")
|
|
|
|
self.assertEqual(backend._session_id, "session-1")
|
|
self.assertEqual(backend._attached_target, "target-1")
|
|
helper.switch_tab.assert_called_once_with("target-1", activate=False)
|
|
helper.cdp.assert_not_called()
|
|
|
|
def test_real_backend_caches_select_all_modifier_for_backend_lifetime(self):
|
|
helper = Mock()
|
|
calls = []
|
|
def cdp(method, **_kwargs):
|
|
calls.append(method)
|
|
return {"userAgent": "Macintosh"} if method == "Browser.getVersion" else {}
|
|
helper.cdp.side_effect = cdp
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = helper
|
|
backend._call = Mock(return_value=True)
|
|
|
|
backend.fill(1, "first")
|
|
backend.fill(1, "second")
|
|
|
|
self.assertEqual(calls.count("Browser.getVersion"), 1)
|
|
keydowns = [call for call in helper.cdp.call_args_list
|
|
if call.args[0] == "Input.dispatchKeyEvent" and call.kwargs["type"] == "rawKeyDown"]
|
|
self.assertEqual([call.kwargs["modifiers"] for call in keydowns], [4, 4])
|
|
|
|
def test_real_backend_unrestricted_policy_still_requires_one_web_tab(self):
|
|
helper = Mock()
|
|
helper.list_tabs.return_value = [
|
|
{"targetId": "blank", "url": "about:blank"},
|
|
{"targetId": "web", "url": "https://other.example/form"},
|
|
]
|
|
child = Mock()
|
|
child.poll.return_value = None
|
|
child.wait.return_value = 0
|
|
with patch.dict(os.environ, dict(os.environ)), \
|
|
patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
|
|
patch("browser_worker.subprocess.Popen", return_value=child):
|
|
backend = RealBrowserBackend("http://127.0.0.1:9222", "*")
|
|
helper.switch_tab.assert_called_once_with("web", activate=False)
|
|
backend.close()
|
|
|
|
def test_real_backend_owns_one_daemon_and_only_attaches_approved_tab(self):
|
|
helper = Mock()
|
|
helper.list_tabs.return_value = [
|
|
{"targetId": "blank-owned-by-harness", "url": "about:blank"},
|
|
{"targetId": "approved", "url": "http://fixture.test/form"},
|
|
{"targetId": "unrelated", "url": "https://other.test/"},
|
|
]
|
|
child = Mock()
|
|
child.poll.return_value = None
|
|
child.wait.return_value = 0
|
|
with patch.dict(os.environ, dict(os.environ)), \
|
|
patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
|
|
patch("browser_worker.subprocess.Popen", return_value=child) as spawn:
|
|
backend = RealBrowserBackend("http://127.0.0.1:9222", ["http://fixture.test"])
|
|
root = backend._sandbox
|
|
spawn.assert_called_once()
|
|
self.assertTrue(spawn.call_args.kwargs["start_new_session"])
|
|
helper.switch_tab.assert_called_once_with("approved", activate=False)
|
|
self.assertEqual(os.environ["BU_AUTOSPAWN"], "")
|
|
self.assertEqual(os.environ["BROWSER_USE_API_KEY"], "")
|
|
backend.close()
|
|
backend.close()
|
|
self.assertFalse(root.exists())
|
|
helper._send.assert_called_once_with({"meta": "shutdown"}, response_timeout=0.3)
|
|
child.wait.assert_called_once()
|
|
|
|
def test_ambiguous_tabs_and_startup_errors_cleanup(self):
|
|
helper = Mock()
|
|
helper.list_tabs.return_value = [
|
|
{"targetId": str(n), "url": "http://fixture.test/form"} for n in range(2)
|
|
]
|
|
child = Mock()
|
|
child.poll.return_value = None
|
|
child.wait.return_value = 0
|
|
with patch.dict(os.environ, dict(os.environ)), \
|
|
patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
|
|
patch("browser_worker.subprocess.Popen", return_value=child):
|
|
backend = RealBrowserBackend("http://127.0.0.1:9222", ["http://fixture.test"])
|
|
root = Path(os.environ["BH_HOME"]).parent
|
|
self.assertTrue(root.exists())
|
|
helper.switch_tab.assert_not_called()
|
|
backend.close()
|
|
self.assertFalse(root.exists())
|
|
helper._send.assert_called_once()
|
|
|
|
def test_hit_test_blocks_covered_click_and_bulk_fill_uses_insert_text(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = Mock()
|
|
backend._helpers.cdp.return_value = {"model": {"content": [0, 0, 20, 0, 20, 20, 0, 20]}}
|
|
backend._call = Mock(return_value=False)
|
|
with self.assertRaisesRegex(WorkerError, "covered"):
|
|
backend.click(1)
|
|
self.assertEqual(backend._helpers.cdp.call_count, 1)
|
|
backend._call.return_value = True
|
|
backend._helpers.cdp.reset_mock()
|
|
backend._helpers.cdp.return_value = {"userAgent": "Macintosh"}
|
|
backend.fill(1, "hello\nworld")
|
|
calls = backend._helpers.cdp.call_args_list
|
|
insert = [c for c in calls if c.args[0] == "Input.insertText"]
|
|
self.assertEqual(len(insert), 1)
|
|
self.assertEqual(insert[0].kwargs["text"], "hello\nworld")
|
|
keydowns = [c for c in calls if c.args[0] == "Input.dispatchKeyEvent" and c.kwargs["type"] == "rawKeyDown"]
|
|
self.assertEqual(keydowns[0].kwargs["modifiers"], 4)
|
|
backend._helpers.press_key.assert_called_once_with("Backspace")
|
|
|
|
def test_metadata_many_uses_one_call_and_releases_temporary_object_group(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
helper = Mock()
|
|
backend._helpers = helper
|
|
|
|
def cdp(method, **kwargs):
|
|
if method == "DOM.resolveNode":
|
|
return {"object": {"objectId": f"object-{kwargs['backendNodeId']}"}}
|
|
if method == "Runtime.callFunctionOn":
|
|
return {"result": {"value": [
|
|
{"tag": "input", "type": "text", "role": "", "name": "Email",
|
|
"connected": True, "mainFrame": True, "contenteditable": False,
|
|
"visible": True, "disabled": False},
|
|
{"tag": "button", "type": "", "role": "button", "name": "Submit",
|
|
"connected": True, "mainFrame": True, "contenteditable": False,
|
|
"visible": True, "disabled": False},
|
|
]}}
|
|
return {}
|
|
helper.cdp.side_effect = cdp
|
|
|
|
result = backend.metadata_many([1, 2])
|
|
|
|
self.assertEqual([metadata.name for metadata in result.values()], ["Email", "Submit"])
|
|
calls = [call.args[0] for call in helper.cdp.call_args_list]
|
|
self.assertEqual(calls.count("DOM.resolveNode"), 2)
|
|
self.assertEqual(calls.count("Runtime.callFunctionOn"), 1)
|
|
self.assertEqual(calls.count("Runtime.releaseObjectGroup"), 1)
|
|
call = next(call for call in helper.cdp.call_args_list if call.args[0] == "Runtime.callFunctionOn")
|
|
self.assertEqual([argument.get("objectId") for argument in call.kwargs["arguments"]],
|
|
["object-1", "object-2"])
|
|
|
|
def test_metadata_many_releases_object_group_when_batch_call_fails(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
helper = Mock()
|
|
backend._helpers = helper
|
|
|
|
def cdp(method, **kwargs):
|
|
if method == "DOM.resolveNode":
|
|
return {"object": {"objectId": f"object-{kwargs['backendNodeId']}"}}
|
|
if method == "Runtime.callFunctionOn":
|
|
raise RuntimeError("detached")
|
|
return {}
|
|
helper.cdp.side_effect = cdp
|
|
|
|
self.assertEqual(backend.metadata_many([1, 2]), {})
|
|
self.assertEqual(
|
|
[call.args[0] for call in helper.cdp.call_args_list].count("Runtime.releaseObjectGroup"), 1
|
|
)
|
|
|
|
def test_call_releases_remote_object_on_browser_exception(self):
|
|
backend = RealBrowserBackend.__new__(RealBrowserBackend)
|
|
backend._helpers = Mock()
|
|
backend._helpers.cdp.side_effect = [{"object": {"objectId": "obj"}}, {"exceptionDetails": {"text": "private"}}, {}]
|
|
with self.assertRaisesRegex(WorkerError, "evaluation failed"):
|
|
backend._call(1, "function () {}")
|
|
self.assertEqual(backend._helpers.cdp.call_args.args[0], "Runtime.releaseObject")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|