Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146"""WebSocket masking needs a controlled peer that daemon E2E cannot expose.
Optional Chrome scenarios exercise the same CDP client against a real browser."""
from __future__ import annotations
import asyncioimport base64import hashlibfrom pathlib import Pathimport shutilimport sysimport unittest
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "priv" / "python"))
from albedo_plugins.browser.transport import SimpleWebSocketClient # noqa: E402from albedo_plugins.browser import spawn, render, one # noqa: E402
class WebSocketClientUnitTest(unittest.IsolatedAsyncioTestCase): """Test the pure-Python RFC 6455 client against an asyncio mock server."""
async def test_handshake_and_echo(self): received_frames: list[str] = []
async def handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter): key = "" while True: line = await reader.readline() if line in (b"\r\n", b"\n", b""): break if line.lower().startswith(b"sec-websocket-key:"): key = line.split(b":", 1)[1].strip().decode("ascii")
accept = base64.b64encode( hashlib.sha1( (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii") ).digest() ).decode("ascii") resp = ( f"HTTP/1.1 101 Switching Protocols\r\n" f"Upgrade: websocket\r\n" f"Connection: Upgrade\r\n" f"Sec-WebSocket-Accept: {accept}\r\n\r\n" ).encode("ascii") writer.write(resp) await writer.drain()
# Read frame from client (client masks frames) head = await reader.readexactly(2) b2 = head[1] length = b2 & 0x7F mask = await reader.readexactly(4) payload = await reader.readexactly(length) unmasked = bytearray(payload) for i in range(len(unmasked)): unmasked[i] ^= mask[i % 4] received_frames.append(unmasked.decode("utf-8"))
# Send back echo frame unmasked msg = b"echo: " + bytes(unmasked) frame = bytearray([0x81, len(msg)]) + msg writer.write(frame) await writer.drain()
writer.close() await writer.wait_closed()
server = await asyncio.start_server(handler, "127.0.0.1", 0) port = server.sockets[0].getsockname()[1] async with server: client = await SimpleWebSocketClient.connect(f"ws://127.0.0.1:{port}/") await client.send("hello albedo") reply = await client.recv_message() self.assertEqual(reply, "echo: hello albedo") self.assertEqual(received_frames, ["hello albedo"]) await client.close()
class BrowserIntegrationTest(unittest.IsolatedAsyncioTestCase): """End-to-end integration tests using local Chromium if present."""
async def asyncSetUp(self): self.has_chrome = ( shutil.which("chromium") is not None or shutil.which("google-chrome") is not None or Path("/Applications/Chromium.app/Contents/MacOS/Chromium").exists() or Path( "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" ).exists() )
async def test_spawn_navigate_observe_interact_close(self): if not self.has_chrome: self.skipTest("No Chrome or Chromium installed on this system")
b = await spawn(headless=True) try: page = await b.new_page() html = ( "data:text/html," "<!doctype html>" "<html><head><title>Albedo Test</title></head>" "<body>" "<h1>Header</h1>" "<label for='name'>Name</label><input id='name' value='initial'>" "<button id='save' onclick='window.clicked=true'>Save</button>" "</body></html>" ) await page.goto(html)
# Observe obs = await page.observe() self.assertEqual(obs["title"], "Albedo Test") text = render(obs) self.assertIn("Header", text) self.assertIn("Save", text)
# Find and One btn = one(obs, role="button", name="Save") self.assertIsNotNone(btn) field = one(obs, role="textbox", name="Name") self.assertIsNotNone(field)
# Fill and Click await page.fill(field, "new value") self.assertEqual( await page.evaluate("document.querySelector('#name').value"), "new value", )
await page.click(btn) self.assertTrue(await page.evaluate("window.clicked"))
# Close page await page.close() finally: await b.close()
if __name__ == "__main__": unittest.main()