"""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

',kind:'markdown',revision:2,content:''}; 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(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') return options.method==='POST'?{artifact:{...artifact,...JSON.parse(options.body),revision:1}}:{artifacts:[artifact]}; 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('') 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_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('

changed') self.assertEqual(self.page.locator('pre').inner_text(), '

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()