137 lines
8.0 KiB
Python
137 lines
8.0 KiB
Python
|
"""Real headless Chromium against local fixtures; no external websites or API keys."""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
from pathlib import Path
|
||
|
|
import queue
|
||
|
|
import subprocess
|
||
|
|
import tempfile
|
||
|
|
import threading
|
||
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
|
|
||
|
|
HELPER = Path(__file__).resolve().parents[1] / 'tools/playwright/browser_helper.mjs'
|
||
|
|
HTML = '''<!doctype html><title>Fixture</title><body style="min-height:2500px">
|
||
|
|
<p id="article">Research evidence: this paragraph is readable independently of the control snapshot.</p>
|
||
|
|
<input id="text"><input type="password" value="must-not-appear"><span id="result"></span>
|
||
|
|
<select id="choice"><option value="a">Alpha</option><option value="b">Beta</option></select>
|
||
|
|
<button id="later" onclick="setTimeout(()=>{document.querySelector('#result').textContent='ready'},200)">Later</button>
|
||
|
|
<a id="popup" target="_blank" href="/popup">Open popup</a><iframe id="child" src="/frame"></iframe>
|
||
|
|
<input id="file" type="file"><a id="download" download="payload.txt" href="/payload">Download</a>
|
||
|
|
<script>document.querySelector('#text').onkeydown=e=>{if(e.key==='Enter')document.querySelector('#result').textContent='entered'};</script>
|
||
|
|
</body>'''
|
||
|
|
|
||
|
|
class Site(BaseHTTPRequestHandler):
|
||
|
|
def do_GET(self):
|
||
|
|
if self.path == '/payload':
|
||
|
|
body, kind = b'fixture download\n', 'application/octet-stream'
|
||
|
|
elif self.path == '/frame':
|
||
|
|
body, kind = b'<body><p>Frame evidence</p><input id="inside"></body>', 'text/html'
|
||
|
|
elif self.path == '/popup':
|
||
|
|
body, kind = b'<title>Popup</title><body>Popup evidence</body>', 'text/html'
|
||
|
|
else:
|
||
|
|
body, kind = HTML.encode(), 'text/html'
|
||
|
|
self.send_response(200)
|
||
|
|
self.send_header('Content-Type', kind)
|
||
|
|
self.send_header('Content-Length', str(len(body)))
|
||
|
|
self.end_headers()
|
||
|
|
self.wfile.write(body)
|
||
|
|
def log_message(self, *_): pass
|
||
|
|
|
||
|
|
class Helper:
|
||
|
|
def __init__(self, root):
|
||
|
|
self.queue = queue.Queue()
|
||
|
|
self.process = subprocess.Popen(['node', str(HELPER)], cwd=root, env={**os.environ,
|
||
|
|
'GROKBOY_BROWSER_HEADED':'0', 'GROKBOY_BROWSER_PROFILE':str(Path(root)/'profile')},
|
||
|
|
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
|
||
|
|
threading.Thread(target=lambda:[self.queue.put(line) for line in self.process.stdout], daemon=True).start()
|
||
|
|
self.errors=[]
|
||
|
|
threading.Thread(target=lambda:[self.errors.append(line) for line in self.process.stderr], daemon=True).start()
|
||
|
|
self.next=0
|
||
|
|
def call(self, op, **args):
|
||
|
|
self.next+=1
|
||
|
|
self.process.stdin.write(json.dumps({'id':str(self.next),'op':op,**args})+'\n')
|
||
|
|
self.process.stdin.flush()
|
||
|
|
try: result=json.loads(self.queue.get(timeout=40))
|
||
|
|
except queue.Empty: raise AssertionError(('helper timeout', self.errors))
|
||
|
|
assert result['id']==str(self.next), result
|
||
|
|
assert result.get('ok'), result
|
||
|
|
return result
|
||
|
|
def close(self):
|
||
|
|
if self.process.poll() is None:
|
||
|
|
try: self.call('close')
|
||
|
|
finally:
|
||
|
|
self.process.stdin.close()
|
||
|
|
self.process.wait(timeout=10)
|
||
|
|
|
||
|
|
def main():
|
||
|
|
with tempfile.TemporaryDirectory(prefix='grokboy-browser-') as root, ThreadingHTTPServer(('127.0.0.1',0), Site) as site:
|
||
|
|
threading.Thread(target=site.serve_forever,daemon=True).start()
|
||
|
|
helper=Helper(root)
|
||
|
|
try:
|
||
|
|
url=f'http://127.0.0.1:{site.server_port}/'
|
||
|
|
helper.call('navigate',url=url)
|
||
|
|
snap=helper.call('snapshot')
|
||
|
|
assert 'must-not-appear' not in json.dumps(snap)
|
||
|
|
assert snap['frames'][0]['selector']=='#child',snap
|
||
|
|
page=helper.call('read_page',limit=3000)
|
||
|
|
assert 'Research evidence' in page['text'] and page['links']
|
||
|
|
assert helper.call('read_page',offset=0,limit=10)['truncated']
|
||
|
|
helper.call('type',selector='#text',text='hello')
|
||
|
|
helper.call('press',selector='#text',key='Enter')
|
||
|
|
assert helper.call('eval',expression="document.querySelector('#result').textContent")['result']=='entered'
|
||
|
|
helper.call('select',selector='#choice',values=['b'])
|
||
|
|
assert helper.call('eval',expression="document.querySelector('#choice').value")['result']=='b'
|
||
|
|
helper.call('click',selector='#later')
|
||
|
|
helper.call('wait',text='ready',timeout_ms=2000)
|
||
|
|
print('PASS body, links, segmentation, password omission, keyboard, select, dynamic wait')
|
||
|
|
helper.call('type',frame='#child',selector='#inside',text='in frame')
|
||
|
|
assert helper.call('eval',frame='#child',expression="document.querySelector('#inside').value")['result']=='in frame'
|
||
|
|
helper.call('click',selector='#popup')
|
||
|
|
tabs=helper.call('tabs')['tabs']
|
||
|
|
popup=next(t for t in tabs if t['url'].endswith('/popup'))
|
||
|
|
helper.call('tabs',action='switch',tab_id=popup['tab_id'])
|
||
|
|
assert 'Popup evidence' in helper.call('read_page')['text']
|
||
|
|
helper.call('tabs',action='close',tab_id=popup['tab_id'])
|
||
|
|
helper.call('scroll',delta_y=700)
|
||
|
|
assert helper.call('eval',expression='window.scrollY')['result']>0
|
||
|
|
print('PASS iframe, popup, tab switching/closing, scrolling')
|
||
|
|
Path(root,'upload.txt').write_text('upload content')
|
||
|
|
helper.call('upload',selector='#file',path='upload.txt')
|
||
|
|
assert helper.call('eval',expression="document.querySelector('#file').files[0].name")['result']=='upload.txt'
|
||
|
|
helper.call('download',selector='#download',path='downloads/file.txt')
|
||
|
|
assert Path(root,'downloads/file.txt').read_text()=='fixture download\n'
|
||
|
|
print('PASS workspace upload and actual download')
|
||
|
|
helper.call('eval',expression="localStorage.setItem('proof','saved'); document.cookie='login=yes; path=/'")
|
||
|
|
helper.close()
|
||
|
|
helper=Helper(root)
|
||
|
|
helper.call('navigate',url=url)
|
||
|
|
assert helper.call('eval',expression="localStorage.getItem('proof')")['result']=='saved'
|
||
|
|
assert 'login=yes' in helper.call('eval',expression='document.cookie')['result']
|
||
|
|
print('PASS persistent profile after helper restart')
|
||
|
|
if '--handoff' in __import__('sys').argv:
|
||
|
|
helper.call('eval',expression="sessionStorage.setItem('oauth_state','keep-me')")
|
||
|
|
original=helper.call('tabs')['tabs'][0]['tab_id']
|
||
|
|
helper.call('tabs',action='new',url=url+'popup')
|
||
|
|
helper.call('tabs',action='switch',tab_id=original)
|
||
|
|
assert helper.call('eval',expression="sessionStorage.getItem('oauth_state')")['result']=='keep-me'
|
||
|
|
prepared=helper.call('handoff_prepare')
|
||
|
|
assert prepared['headed'] and prepared['relaunched'],prepared
|
||
|
|
assert prepared['url']==url,prepared
|
||
|
|
state=helper.call('eval',expression="sessionStorage.getItem('oauth_state')")
|
||
|
|
assert state['result']=='keep-me',(state,prepared)
|
||
|
|
assert 'login=yes' in helper.call('eval',expression='document.cookie')['result']
|
||
|
|
assert helper.call('eval',expression="localStorage.getItem('proof')")['result']=='saved'
|
||
|
|
assert len(helper.call('tabs')['tabs'])>=2
|
||
|
|
assert helper.call('handoff_prepare')['relaunched'] is False
|
||
|
|
helper.call('eval',expression="sessionStorage.setItem('oauth_state','new-value')")
|
||
|
|
helper.call('navigate',url=url)
|
||
|
|
assert helper.call('eval',expression="sessionStorage.getItem('oauth_state')")['result']=='new-value'
|
||
|
|
helper.call('eval',expression="document.cookie='manual_login=done; path=/'")
|
||
|
|
assert 'manual_login=done' in helper.call('eval',expression='document.cookie')['result']
|
||
|
|
print('PASS visible handoff preserves active tab, cookies, localStorage, sessionStorage and subsequent browser operations')
|
||
|
|
|
||
|
|
finally:
|
||
|
|
helper.close()
|
||
|
|
site.shutdown()
|
||
|
|
|
||
|
|
if __name__=='__main__': main()
|