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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273"""Optional Chrome scenarios exercise the CDP client against a real browser."""
from __future__ import annotations
from functools import partialfrom http.server import SimpleHTTPRequestHandler, ThreadingHTTPServerfrom pathlib import Pathfrom unittest import mockimport contextlibimport osimport shutilimport sysimport tempfileimport threadingimport timeimport unittest
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "priv" / "python"))
from albedo_plugins.browser import spawn, render, one # noqa: E402from albedo_plugins.browser.errors import NotActionable # noqa: E402from albedo_plugins.browser import hooks, paths # noqa: E402from albedo_plugins.browser.launch import OwnedChrome, _command # noqa: E402
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>" "<label for='pick'>Pick</label>" "<select id='pick' onchange='window.picked=this.value'>" "<option value='a'>Alpha</option>" "<option value='b'>Beta</option>" "<option value='c' disabled>Gamma</option>" "</select>" "</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"))
# Select takes an option's value or its label picker = one(obs, role="combobox", name="Pick") self.assertIsNotNone(picker) await page.select(picker, "b") self.assertEqual( await page.evaluate( "[window.picked, document.querySelector('#pick').value]" ), ["b", "b"], ) await page.select(picker, "Alpha") self.assertEqual( await page.evaluate("document.querySelector('#pick').value"), "a" ) with self.assertRaises(NotActionable) as refused: await page.select(picker, "Gamma") self.assertEqual(refused.exception.details["found"], False)
# A screenshot nobody gave a path goes to the cache, not the working tree with ( tempfile.TemporaryDirectory() as home, mock.patch.dict(os.environ, {"ALBEDO_HOME": home}), ): shot = await page.screenshot() self.assertEqual( shot.parent, (Path(home) / "cache" / "browser-screenshots").resolve(), ) self.assertTrue(shot.read_bytes().startswith(b"\x89PNG"))
# Close page await page.close() finally: await b.close()
async def test_prerender_takes_the_opt_in_and_the_page_follows_activation(self): if not self.has_chrome: self.skipTest("No Chrome or Chromium installed on this system")
root = tempfile.TemporaryDirectory() self.addCleanup(root.cleanup) (Path(root.name) / "index.html").write_text( "<!doctype html><title>Start</title>" '<script type="speculationrules">' '{"prerender": [{"source": "list", "urls": ["/next.html"]}]}' "</script>" '<a href="/next.html">Next</a>' ) (Path(root.name) / "next.html").write_text( "<!doctype html><title>Next</title><h1>Prerendered</h1>" ) handler = partial(SimpleHTTPRequestHandler, directory=root.name) server = ThreadingHTTPServer(("127.0.0.1", 0), handler) threading.Thread(target=server.serve_forever, daemon=True).start() self.addCleanup(server.server_close) self.addCleanup(server.shutdown) start = f"http://127.0.0.1:{server.server_port}/index.html"
b = await spawn(headless=True) try: outcomes = {} for prerender in (False, True): page = await b.new_page(prerender=prerender) await page.cdp.send("Preload.enable") async with page.events( "Preload.prerenderStatusUpdated", enable=False ) as updates: await page.goto(start) update = await updates.next( where=lambda e: e["status"] in {"Ready", "Failure"} ) outcomes[prerender] = (update["status"], update.get("prerenderStatus")) if prerender: before = page.target_id await page.click( one(await page.observe(), role="link", name="Next") ) await page.wait_for(role="heading", name="Prerendered") self.assertNotEqual(page.target_id, before) self.assertEqual( await page.evaluate( "[document.title, performance.getEntriesByType('navigation')[0].activationStart > 0]" ), ["Next", True], ) await page.close() self.assertEqual( outcomes, { False: ("Failure", "PrerenderingDisabledByDevTools"), True: ("Ready", None), }, ) finally: await b.close()
PROFILE = Path("/tmp/profile")
class LaunchCommandTest(unittest.TestCase): """The argv albedo hands Chrome, read without starting one."""
def test_macos_launches_stay_out_of_the_login_keychain(self): # Chrome raises a blocking "Keychain Not Found" panel when it cannot # resolve a default keychain for its own process. command = _command( "/opt/chrome", PROFILE, headless=False, sandbox=True, args=[] ) self.assertEqual("--use-mock-keychain" in command, sys.platform == "darwin")
def test_the_owned_options_are_the_ones_asked_for(self): command = _command( "/opt/chrome", PROFILE, headless=False, sandbox=False, args=["--lang=en"] ) self.assertEqual(command[0], "/opt/chrome") self.assertIn("--user-data-dir=/tmp/profile", command) self.assertIn("--remote-debugging-port=0", command) self.assertNotIn("--headless=new", command) self.assertIn("--no-sandbox", command) self.assertEqual(command[-2:], ["--lang=en", "about:blank"]) headless = _command( "/opt/chrome", PROFILE, headless=True, sandbox=True, args=[] ) self.assertIn("--headless=new", headless)
class ScreenshotCacheTest(unittest.TestCase): """Default screenshots live under the albedo home and are not kept forever."""
def test_they_go_to_the_cache_under_the_albedo_home(self): with ( tempfile.TemporaryDirectory() as home, mock.patch.dict(os.environ, {"ALBEDO_HOME": home}), ): self.assertEqual( paths.screenshots(), Path(home) / "cache" / "browser-screenshots" )
def test_only_old_screenshots_are_pruned(self): with tempfile.TemporaryDirectory() as directory: folder = Path(directory) old, fresh, other = ( folder / "old.png", folder / "fresh.png", folder / "n.txt", ) for file in (old, fresh, other): file.write_bytes(b"x") week_ago = time.time() - paths.SCREENSHOT_TTL_SECONDS - 60 for file in (old, other): os.utime(file, (week_ago, week_ago)) paths.prune(folder) self.assertEqual( sorted(p.name for p in folder.iterdir()), ["fresh.png", "n.txt"] ) paths.prune(folder / "missing")
class BookkeepingTest(unittest.TestCase): """The browser's own files stay out of a cell's file-change trace."""
def test_profile_setup_runs_unobserved(self): entered: list[bool] = [] original = hooks.unobserved
@contextlib.contextmanager def unobserved(): entered.append(True) yield
hooks.unobserved = unobserved try: chrome = OwnedChrome(None) try: self.assertTrue(entered) self.assertTrue((chrome.profile / ".albedo-browser-profile").exists()) finally: chrome._release() finally: hooks.unobserved = original
if __name__ == "__main__": unittest.main()