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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259"""Browser discovery through the real kernel, without starting a GUI app."""
import unittestimport http.serverimport jsonimport osimport signalimport threadingimport shutilimport timefrom pathlib import Path
from harness import Albedo, Provider, exclusive, python, text
@exclusiveclass BrowserTests(unittest.TestCase): def test_headless_shell_preserves_overrides_headed_launches_and_fallback(self): code = ""
def script(request): if request["messages"][-1].get("role") == "user": return python(code) return text("done")
def prepare(app): nonlocal code bin_dir = app.workspace / "bin" bin_dir.mkdir() chrome = bin_dir / "chromium" shell = bin_dir / "headless-shell" shell_link = app.home / "browsers/chrome-headless-shell" shell_link.parent.mkdir() shell_link.symlink_to(shell) for executable in (chrome, shell): executable.write_text( '#!/bin/sh\necho "test browser exited" >&2\nexit 23\n' ) executable.chmod(0o755) for variable in ( "ALBEDO_BROWSER_EXECUTABLE", "PRIME_BROWSER_EXECUTABLE", "CHROME_PATH", ): app.daemon.env[variable] = ""
code = f"""from pathlib import Pathfrom albedo_plugins.browser import LaunchErrorchrome = {str(chrome)!r}shell = {str(shell)!r}os.environ['PATH'] = {str(bin_dir)!r}async def check(expected, **options): try: b = await browser.spawn(sandbox=False, timeout=3, **options) except LaunchError as error: assert error.details['executable'] == expected, error.to_dict() assert error.details['reason'] == 'process_exited', error.to_dict() assert 'test browser exited' in error.details['stderr_tail'], error.to_dict() assert not Path(error.details['profile']).exists() else: await b.close() raise AssertionError('test browser should exit during startup')await check(shell)path_shell = Path({str(bin_dir / "chrome-headless-shell")!r})path_shell.symlink_to(chrome)await check(chrome)path_shell.unlink()await check(chrome, headless=False)await check(chrome, executable=chrome)os.environ['ALBEDO_BROWSER_EXECUTABLE'] = chromeawait check(chrome)del os.environ['ALBEDO_BROWSER_EXECUTABLE']Path({str(shell_link)!r}).unlink()await check(chrome)try: await browser.spawn(executable={str(shell_link)!r}, sandbox=False)except LaunchError as error: assert error.details['reason'] == 'executable_not_found', error.to_dict()else: raise AssertionError('missing explicit browser must not fall back')Path({str(app.workspace / "passed")!r}).write_text('passed')"""
provider = Provider(script) self.addCleanup(provider.close) with Albedo(provider, prepare=prepare) as app: session = app.session() app.prompt(session, "try the installed browsers").close() app.idle(session) self.assertTrue((app.workspace / "passed").exists(), app.history(session))
def require_chrome(test): if not ( shutil.which("chromium") or shutil.which("google-chrome") or shutil.which("chrome-headless-shell") or Path("/Applications/Google Chrome.app/Contents/MacOS/Google Chrome").exists() or Path("/Applications/Chromium.app/Contents/MacOS/Chromium").exists() ): test.skipTest("No Chrome or Chromium installed on this system")
class BrowserLifetimeTests(unittest.TestCase): def test_a_kernel_killed_outright_takes_its_browser_with_it(self): require_chrome(self) found = None
def reply(request): if request["messages"][-1].get("role") == "user": return python( f"""import jsonfrom albedo_plugins.browser import launchb = await browser.spawn()owner = next(iter(launch._LIVE))Path({str(found)!r}).write_text(json.dumps( {{'kernel': os.getpid(), 'chrome': owner.process.pid, 'profile': str(owner.profile)}}))""" ) return text("done")
provider = Provider(reply) self.addCleanup(provider.close) with Albedo(provider) as app: found = app.workspace / "browser.json" session = app.session() app.prompt(session, "start a browser").close() app.idle(session) spawned = json.loads(found.read_text()) # SIGKILL skips the kernel's own cleanup; only the watchdog is left. os.kill(spawned["kernel"], signal.SIGKILL) deadline = time.monotonic() + 15 # The watchdog removes the profile once Chrome's group has ended. while time.monotonic() < deadline and ( running(spawned["chrome"]) or Path(spawned["profile"]).exists() ): time.sleep(0.1) self.assertFalse(running(spawned["chrome"]), "Chrome outlived its kernel") self.assertFalse(Path(spawned["profile"]).exists())
def running(pid): try: os.kill(pid, 0) except ProcessLookupError: return False return True
class BrowserGestureTests(unittest.TestCase): def test_gestures_and_statement_bodies_in_a_real_browser(self): require_chrome(self)
class Handler(http.server.BaseHTTPRequestHandler): def do_GET(self): body = b"<!doctype html><title>CSP test</title>" self.send_response(200) self.send_header("Content-Type", "text/html") self.send_header("Content-Length", str(len(body))) self.send_header("Content-Security-Policy", "script-src 'self'") self.end_headers() self.wfile.write(body)
def log_message(self, format, *args): pass
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() self.addCleanup(server.server_close) self.addCleanup(thread.join) self.addCleanup(server.shutdown) code = """from albedo_plugins.browser import SelectionError, JavaScriptErrorb = await browser.spawn()try: page = await b.new_page() await page.goto('data:text/html,<button style="position:absolute;left:10px;top:10px;width:90px;height:40px" onclick="window.clicks=(window.clicks||0)+1">Drag handle</button><a href="#">Other link</a><button>Other button</button>') assert await page.evaluate('1 + 2') == 3 assert await page.evaluate('() => 4') == 4 assert await page.evaluate('window.runs = 0; return 5') == 5 try: await page.evaluate("window.runs++; throw new SyntaxError('runtime')") except JavaScriptError: pass else: raise AssertionError('runtime error was swallowed') assert await page.evaluate('window.runs') == 1 try: await page.evaluate("(() => { window.runs++; throw new SyntaxError('runtime'); })()") except JavaScriptError: pass else: raise AssertionError('expression runtime error was swallowed') assert await page.evaluate('window.runs') == 2 await page.click((30, 30)) await page.click_text('Drag handle') assert await page.evaluate('window.clicks') == 2 for label in ('missing', 'Other'): try: await page.click_text(label) except SelectionError as error: assert ('found 0' if label == 'missing' else 'found 2') in str(error) else: raise AssertionError('ambiguous or missing text was clicked') await page.evaluate(''' window.moves = []; window.releases = 0; document.addEventListener('pointermove', e => {if(e.buttons === 1) window.moves.push([e.clientX,e.clientY]);}); document.addEventListener('pointerup', () => window.releases++); return true; ''') snapshot = await page.observe() node = browser.one(snapshot, role='button', name='Drag handle') await page.drag(node, (200, 150), steps=5) assert await page.evaluate('window.moves.length') == 5 assert await page.evaluate('window.moves.at(-1)') == [200, 150] assert await page.evaluate('window.releases') == 1 await page.drag((200, 150), node, steps=3) assert await page.evaluate('window.moves.length') == 8 assert await page.evaluate('window.releases') == 2 print('document drag: 8 held-button moves, 2 releases; clicks: 2') await page.goto(CSP_URL) try: await page.evaluate("new Function('return 1')()") except JavaScriptError as error: assert 'unsafe-eval' in str(error.details), error else: raise AssertionError('CSP must block Function compilation') assert await page.evaluate('1 + 2') == 3 assert await page.evaluate('localStorage.clear(); return 5') == 5 print('CSP expression and statement body passed')finally: await b.close()"""
def reply(request): if request["messages"][-1].get("role") == "user": return python( code.replace( "CSP_URL", repr(f"http://127.0.0.1:{server.server_port}/") ) ) return text("done")
provider = Provider(reply) self.addCleanup(provider.close) with Albedo(provider) as app: session = app.session() app.prompt(session, "exercise browser gestures").close() app.idle(session) history = str(app.history(session)) self.assertIn("CSP expression and statement body passed", history) self.assertIn( "document drag: 8 held-button moves, 2 releases; clicks: 2", history )