198 lines
14 KiB
Python
198 lines
14 KiB
Python
"""Offline Chromium qualification for advanced chat controls; no service calls."""
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
MOCK = r'''() => {
|
|
window.calls = []; window.listeners = []; window.pendingUpload = false; window.conflict = false;
|
|
window.capabilities = {attachments:true, artifacts:true, scheduled_tasks:true, memory_controls:true, image_input:true,
|
|
attachment_mimes:['text/plain','text/markdown','application/pdf','image/png'], upload_bytes:8388608};
|
|
window.artifact = {id:'a'.repeat(32),title:'Draft <img src=x onerror=alert(1)>',kind:'markdown',revision:2,content:'<script>window.injected=true</script>'};
|
|
window.task = {id:'b'.repeat(12),name:'Daily note',schedule:'0 9 * * *',prompt:'Review notes',enabled:true};
|
|
window.HopeAuth = {user:{id:'owner-a',status:'ready'},onChange(fn){listeners.push(fn);fn(this.user,true);},
|
|
async request(path,options={}) {
|
|
calls.push({path,method:options.method||'GET',body:typeof options.body==='string'?JSON.parse(options.body):null,
|
|
binary:options.body instanceof File,headers:options.headers});
|
|
if (path==='/api/capabilities') return {features:capabilities};
|
|
if (path==='/api/attachments') {
|
|
if(options.method==='GET') return {attachments:[{id:'c'.repeat(32),filename:'notes.txt',size:5,extraction_status:'ready'}]};
|
|
if(pendingUpload) await new Promise((resolve,reject)=>options.signal.addEventListener('abort',()=>reject(new DOMException('Stopped','AbortError'))));
|
|
return {attachment:{id:'c'.repeat(32),filename:options.body.name,extraction_status:'ready'}};
|
|
}
|
|
if(path==='/api/artifacts') {const {content,...metadata}=artifact;return options.method==='POST'?{artifact:{...artifact,...JSON.parse(options.body),revision:1}}:{artifacts:[metadata]};}
|
|
if(path.startsWith('/api/artifacts/')) {
|
|
if(options.method==='PATCH') {if(conflict) throw new Error('Artifact revision conflict');artifact={...artifact,...JSON.parse(options.body),revision:artifact.revision+1};}
|
|
return {artifact};
|
|
}
|
|
if(path==='/api/tasks') return options.method==='POST'?{task}:{tasks:[task]};
|
|
if(path.startsWith('/api/tasks/')) return {task};
|
|
if(path==='/api/memory') return {sections:[{section:'memory',content:'Saved facts',revision:'m1'},{section:'user',content:'Profile',revision:'u1'}]};
|
|
if(path.startsWith('/api/memory/')) {if(conflict) throw new Error('Memory revision conflict');return {...JSON.parse(options.body),revision:'m2'};}
|
|
return {ok:true};
|
|
}};
|
|
window.changeAccount = user => {HopeAuth.user=user;listeners.forEach(fn=>fn(user,true));};
|
|
window.createdURLs=[];window.revokedURLs=[];
|
|
URL.createObjectURL=()=>{const url='blob:preview-'+createdURLs.length;createdURLs.push(url);return url;};
|
|
URL.revokeObjectURL=url=>revokedURLs.push(url);
|
|
}'''
|
|
|
|
|
|
class ChatFeaturesWebTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
try:
|
|
from playwright.sync_api import sync_playwright
|
|
except ImportError:
|
|
raise unittest.SkipTest('Playwright required for offline browser qualification')
|
|
cls.playwright = sync_playwright().start()
|
|
cls.browser = cls.playwright.chromium.launch(headless=True)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.browser.close()
|
|
cls.playwright.stop()
|
|
|
|
def setUp(self):
|
|
self.page = self.browser.new_page()
|
|
self.page.set_content('<html><body><aside id="chatSidebar"></aside><form id="primaryComposer"><div id="featureActions"></div><div id="attachmentPreviews"></div></form></body></html>')
|
|
self.page.evaluate(MOCK)
|
|
self.page.add_script_tag(path=str(ROOT / 'web/features.js'))
|
|
self.page.wait_for_function('document.querySelector("[data-attach]").disabled === false')
|
|
|
|
def tearDown(self):
|
|
self.page.close()
|
|
|
|
def open_panel(self, name):
|
|
self.page.get_by_text(name, exact=True).click()
|
|
|
|
def test_upload_raw_bytes_and_ack_only_clear(self):
|
|
self.page.locator('input[type=file]').set_input_files({'name':'notes.txt','mimeType':'text/plain','buffer':b'Hello'})
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.assertFalse(self.page.evaluate('HopeFeatures.hasPendingUploads()'))
|
|
call = self.page.evaluate('calls.find(c=>c.path==="/api/attachments")')
|
|
self.assertTrue(call['binary'])
|
|
self.assertEqual(call['headers'], {'Content-Type':'text/plain','X-Filename':'notes.txt'})
|
|
self.page.evaluate('window.dispatchEvent(new CustomEvent("hope:conversation-selected",{detail:{conversation_id:"first"}}))')
|
|
self.assertEqual(len(self.page.evaluate('HopeFeatures.attachmentIDs()')), 1)
|
|
self.page.evaluate('HopeFeatures.clearAttachmentsAfterSend([])')
|
|
self.assertEqual(len(self.page.evaluate('HopeFeatures.attachmentIDs()')), 1)
|
|
self.page.evaluate('HopeFeatures.clearAttachmentsAfterSend()')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_retained_upload_management_delete_is_explicit(self):
|
|
self.open_panel('Files')
|
|
self.page.get_by_text('Delete notes.txt',exact=True).click()
|
|
self.assertIn("old chat attachment references won't work", self.page.locator('#hopeFeatures').inner_text())
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.method==="DELETE").length'), 0)
|
|
self.assertTrue(self.page.get_by_role('link',name='Download notes.txt').get_attribute('href').endswith('/content'))
|
|
self.page.get_by_role('button',name='Confirm delete notes.txt').click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.startsWith("/api/attachments/")&&c.method==="DELETE")')
|
|
|
|
def test_cancel_retry_and_conversation_change(self):
|
|
self.page.evaluate('pendingUpload=true')
|
|
self.page.locator('input[type=file]').set_input_files({'name':'notes.txt','mimeType':'text/plain','buffer':b'Hello'})
|
|
self.page.get_by_role('button',name='Cancel upload').click()
|
|
self.page.get_by_role('button',name='Retry upload').wait_for()
|
|
self.assertTrue(self.page.evaluate('HopeFeatures.hasPendingUploads()'))
|
|
self.page.evaluate('pendingUpload=false')
|
|
self.page.get_by_role('button',name='Retry upload').click()
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.page.evaluate('window.dispatchEvent(new CustomEvent("hope:conversation-selected",{detail:{conversation_id:"one"}}));window.dispatchEvent(new CustomEvent("hope:conversation-selected",{detail:{conversation_id:"two"}}))')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_capabilities_validation_and_logout_cleanup(self):
|
|
self.page.evaluate('capabilities.image_input=false;HopeFeatures.refresh()')
|
|
self.page.wait_for_function('document.getElementById("featureAvailability").textContent.includes("Image input")')
|
|
self.page.evaluate('HopeFeatures.addFiles([new File(["x"],"image.png",{type:"image/png"})])')
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.path==="/api/attachments").length'), 0)
|
|
self.page.evaluate('capabilities.image_input=true;HopeFeatures.refresh()')
|
|
self.page.wait_for_function('document.querySelector("input[type=file]").accept.includes("image/png")')
|
|
self.page.locator('input[type=file]').set_input_files({'name':'image.png','mimeType':'image/png','buffer':b'fake image for mocked endpoint'})
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.page.evaluate('changeAccount(null)')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
self.assertEqual(self.page.evaluate('createdURLs'), self.page.evaluate('revokedURLs'))
|
|
self.assertTrue(self.page.locator('#hopeFeatures').is_hidden())
|
|
|
|
def test_artifact_safe_preview_versions_conflict_and_export(self):
|
|
self.open_panel('Artifacts')
|
|
self.page.get_by_role('button',name='Edit Draft',exact=False).click()
|
|
self.page.get_by_label('Content',exact=True).fill('<img src=x onerror=alert(1)> changed')
|
|
self.assertEqual(self.page.locator('pre').inner_text(), '<img src=x onerror=alert(1)> changed')
|
|
self.assertEqual(self.page.locator('#hopeFeatures script, #hopeFeatures img').count(), 0)
|
|
self.page.evaluate('conflict=true')
|
|
self.page.get_by_role('button',name='Save new version').click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("retained")')
|
|
self.assertIn('changed', self.page.get_by_label('Content',exact=True).input_value())
|
|
self.page.evaluate('conflict=false')
|
|
self.page.get_by_role('button',name='Save new version').click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("version 3")')
|
|
self.page.get_by_label('Saved version').select_option('1')
|
|
self.page.get_by_role('button',name='View saved version').click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.endsWith("?revision=1"))')
|
|
self.assertIn('/export', self.page.get_by_role('link',name='Export latest saved version').get_attribute('href'))
|
|
self.assertTrue(self.page.get_by_role('link',name='Export selected saved version').get_attribute('href').endswith('/export?revision=1'))
|
|
patch = self.page.evaluate('calls.find(c=>c.method==="PATCH")')
|
|
self.assertEqual(patch['body']['revision'], 2)
|
|
|
|
def test_task_crud_action_shapes(self):
|
|
self.open_panel('Scheduled tasks')
|
|
self.page.get_by_role('button',name='Pause',exact=True).click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.endsWith("/pause"))')
|
|
self.page.get_by_role('button',name='Run now').click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.endsWith("/run"))')
|
|
self.page.get_by_role('button',name='New scheduled task').click()
|
|
self.page.get_by_label('Task name').fill('Reminder')
|
|
self.page.get_by_label('Schedule',exact=True).fill('0 9 * * *')
|
|
self.page.get_by_label('Prompt',exact=True).fill('Summarize notes')
|
|
self.page.get_by_role('button',name='Create task',exact=True).click()
|
|
self.page.wait_for_function('calls.some(c=>c.path==="/api/tasks"&&c.method==="POST")')
|
|
data = self.page.evaluate('calls.find(c=>c.path==="/api/tasks"&&c.method==="POST").body')
|
|
self.assertEqual(data, {'name':'Reminder','schedule':'0 9 * * *','prompt':'Summarize notes','paused':True})
|
|
|
|
def test_drag_drop_validation_and_disabled_feature_panels(self):
|
|
self.page.evaluate('''() => {const transfer=new DataTransfer();transfer.items.add(new File(['notes'],'drop.txt',{type:'text/plain'}));document.getElementById('primaryComposer').dispatchEvent(new DragEvent('drop',{dataTransfer:transfer,bubbles:true,cancelable:true}));}''')
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.page.evaluate('HopeFeatures.addFiles([new File(["x"],"bad.svg",{type:"image/svg+xml"}),new File([],"empty.txt",{type:"text/plain"})])')
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.path==="/api/attachments").length'), 1)
|
|
self.page.evaluate('capabilities.scheduled_tasks=false;capabilities.memory_controls=false;HopeFeatures.refresh()')
|
|
self.page.wait_for_function('document.getElementById("featureAvailability").textContent.includes("Scheduled tasks")')
|
|
self.assertTrue(self.page.get_by_text('Scheduled tasks',exact=True).is_hidden())
|
|
self.assertTrue(self.page.get_by_text('Memory',exact=True).is_hidden())
|
|
|
|
def test_task_edit_resume_delete(self):
|
|
self.page.evaluate('task.enabled=false')
|
|
self.open_panel('Scheduled tasks')
|
|
self.page.get_by_role('button',name='Resume',exact=True).click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.endsWith("/resume"))')
|
|
self.page.get_by_role('button',name='Edit task',exact=True).click()
|
|
self.page.get_by_label('Task name').fill('Renamed task')
|
|
self.page.get_by_role('button',name='Save task',exact=True).click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.startsWith("/api/tasks/")&&c.method==="PATCH")')
|
|
self.assertEqual(self.page.evaluate('calls.find(c=>c.method==="PATCH").body.name'), 'Renamed task')
|
|
self.page.get_by_text('Delete task',exact=True).click()
|
|
self.page.get_by_role('button',name='Confirm delete task').click()
|
|
self.page.wait_for_function('calls.some(c=>c.path.startsWith("/api/tasks/")&&c.method==="DELETE")')
|
|
|
|
def test_memory_revision_conflict_retains_edits_and_explicit_clear(self):
|
|
self.open_panel('Memory')
|
|
content = self.page.get_by_label('Assistant memory',exact=True)
|
|
content.fill('Updated facts')
|
|
self.page.evaluate('conflict=true')
|
|
self.page.get_by_role('button',name='Save assistant memory').click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("retained")')
|
|
self.assertEqual(content.input_value(), 'Updated facts')
|
|
self.page.evaluate('conflict=false')
|
|
self.page.get_by_role('button',name='Save assistant memory').click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("saved")')
|
|
self.page.get_by_text('Clear assistant memory',exact=True).click()
|
|
self.page.get_by_role('button',name='Confirm clear assistant memory').click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("cleared")')
|
|
last = self.page.evaluate('calls.filter(c=>c.path==="/api/memory/memory").at(-1).body')
|
|
self.assertEqual(last, {'content':'','revision':'m2'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|