347 lines
27 KiB
Python
347 lines
27 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.uploadError = ''; window.extraction = 'ready'; window.file = {id:'c'.repeat(32),filename:'notes.txt',mime:'text/plain',size:5,extraction_status:'ready'};
|
|
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:[file]};
|
|
if(pendingUpload) await new Promise((resolve,reject)=>options.signal.addEventListener('abort',()=>reject(new DOMException('Stopped','AbortError'))));
|
|
if(uploadError) throw new Error(uploadError); return {attachment:{id:'c'.repeat(32),filename:options.body.name,mime:options.body.type,size:options.body.size,extraction_status:extraction}};
|
|
}
|
|
if(path.startsWith('/api/attachments/')) return {attachment:file};
|
|
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><dialog id="libraryDialog"><div id="libraryContent"></div></dialog><form id="primaryComposer"><textarea id="primaryInput"></textarea><button id="sendMock" type="submit">Send</button><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')
|
|
self.page.evaluate('document.getElementById("primaryComposer").addEventListener("submit",e=>{e.preventDefault();calls.push({path:"/api/conversations/mock/messages",method:"POST",body:{attachment_ids:HopeFeatures.attachmentIDs()}});HopeFeatures.clearAttachmentsAfterSend();})')
|
|
|
|
def tearDown(self):
|
|
self.page.close()
|
|
|
|
def open_panel(self, name):
|
|
self.page.evaluate('document.getElementById("libraryDialog").showModal()')
|
|
self.page.get_by_text(name, exact=True).click()
|
|
|
|
def test_library_upload_queues_without_send(self):
|
|
self.open_panel('Files')
|
|
with self.page.expect_file_chooser() as chooser:
|
|
self.page.get_by_role('button', name='Upload files', exact=True).click()
|
|
chooser.value.set_files({'name':'library.txt','mimeType':'text/plain','buffer':b'Library'})
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.path.includes("/messages")).length'), 0)
|
|
self.assertIn('nothing is sent until you send', self.page.locator('#hopeFeatures').text_content())
|
|
self.assertFalse(self.page.evaluate('document.getElementById("libraryDialog").open'))
|
|
self.assertEqual(self.page.evaluate('document.activeElement.id'), 'primaryInput')
|
|
self.assertTrue(self.page.locator('#featureComposerStatus').is_visible())
|
|
self.assertIn('library.txt', self.page.locator('#featureComposerStatus').inner_text())
|
|
|
|
def test_rejected_files_report_in_current_surface(self):
|
|
self.open_panel('Files')
|
|
with self.page.expect_file_chooser() as chooser:
|
|
self.page.get_by_role('button', name='Upload files', exact=True).click()
|
|
chooser.value.set_files({'name':'unsupported.svg','mimeType':'image/svg+xml','buffer':b'bad'})
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("Unsupported")')
|
|
self.assertTrue(self.page.evaluate('document.getElementById("libraryDialog").open'))
|
|
self.assertIn('unsupported.svg', self.page.locator('.feature-section').filter(has=self.page.get_by_text('Files', exact=True)).locator('.feature-status').inner_text())
|
|
self.page.evaluate('document.getElementById("libraryDialog").close()')
|
|
self.page.evaluate('HopeFeatures.addFiles([new File([],"empty.txt",{type:"text/plain"})])')
|
|
self.assertTrue(self.page.locator('#featureComposerStatus').is_visible())
|
|
self.assertIn('empty.txt', self.page.locator('#featureComposerStatus').inner_text())
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_reuse_queues_then_explicit_send_and_focuses(self):
|
|
self.open_panel('Files')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.assertFalse(self.page.evaluate('document.getElementById("libraryDialog").open'))
|
|
self.assertEqual(self.page.evaluate('document.activeElement.id'), 'primaryInput')
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.method==="POST").length'), 0)
|
|
self.assertIn('notes.txt queued', self.page.locator('#featureStatus').text_content())
|
|
self.page.get_by_role('button', name='Send', exact=True).click()
|
|
self.assertEqual(self.page.evaluate('calls.find(c=>c.path.includes("/messages")).body.attachment_ids'), ['c'*32])
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_reuse_checks_blocked_images_capacity_and_stale_metadata(self):
|
|
self.page.evaluate('file.extraction_status="encrypted"')
|
|
self.open_panel('Files')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("could not be extracted")')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
self.page.evaluate('file.extraction_status="image";file.mime="image/png";capabilities.image_input=false;HopeFeatures.refresh()')
|
|
self.page.get_by_role('button', name='Refresh files').click()
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("Image input")')
|
|
self.page.evaluate('file.extraction_status="ready";file.mime="text/plain";capabilities.image_input=true;capabilities.attachments_per_message=1;HopeFeatures.refresh()')
|
|
self.page.get_by_role('button', name='Refresh files').click()
|
|
self.page.evaluate('HopeFeatures.addFiles([new File(["x"],"pending.txt",{type:"text/plain"})]);pendingUpload=true')
|
|
self.page.wait_for_function('HopeFeatures.attachmentIDs().length===1')
|
|
self.page.evaluate('file.id="d".repeat(32);document.getElementById("libraryDialog").showModal()')
|
|
self.page.get_by_role('button', name='Refresh files').click()
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("at most 1")')
|
|
self.page.evaluate('HopeFeatures.clearAttachmentsAfterSend();file.extraction_status="ready"')
|
|
self.page.get_by_role('button', name='Refresh files').click()
|
|
self.page.evaluate('file.extraction_status="failed"')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("could not be extracted")')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_upload_error_blocked_and_reuse_account_teardown(self):
|
|
self.page.evaluate('uploadError="Account storage is full"')
|
|
self.page.locator('input[type=file]').set_input_files({'name':'full.txt','mimeType':'text/plain','buffer':b'Hello'})
|
|
self.page.get_by_role('button', name='Retry upload').wait_for()
|
|
self.assertTrue(self.page.evaluate('HopeFeatures.hasPendingUploads()'))
|
|
self.page.evaluate('uploadError="";extraction="failed"')
|
|
self.page.get_by_role('button', name='Retry upload').click()
|
|
self.page.wait_for_function('document.getElementById("attachmentPreviews").textContent.includes("blocked")')
|
|
self.assertTrue(self.page.evaluate('HopeFeatures.hasPendingUploads()'))
|
|
self.page.evaluate('changeAccount({id:"owner-b",status:"ready"})')
|
|
self.page.wait_for_function('document.querySelector("[data-attach]").disabled === false')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
self.open_panel('Files')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
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.locator('.feature-editor').filter(has_text='Updated facts').count(), 0)
|
|
|
|
def test_reuse_inflight_account_change_cannot_queue_previous_account(self):
|
|
self.page.evaluate('''() => {const original=HopeAuth.request;HopeAuth.request=async(path,options)=>{if(path.startsWith('/api/attachments/')){window.reuseSignal=options.signal;await new Promise(resolve=>window.resolveReuse=resolve);}return original(path,options);};}''')
|
|
self.open_panel('Files')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('!!window.resolveReuse')
|
|
self.page.evaluate('changeAccount({id:"owner-b",status:"ready"});resolveReuse()')
|
|
self.page.wait_for_function('document.querySelector("[data-attach]").disabled === false')
|
|
self.assertTrue(self.page.evaluate('reuseSignal.aborted'))
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
self.assertNotIn('queued', self.page.locator('#featureStatus').text_content())
|
|
|
|
def test_overlapping_capability_refresh_ignores_old_vision_response(self):
|
|
self.page.evaluate('''() => {const original=HopeAuth.request;window.capabilityResolvers=[];HopeAuth.request=(path,options)=>path==='/api/capabilities'?new Promise(resolve=>capabilityResolvers.push(resolve)):original(path,options);window.oldRefresh=HopeFeatures.refresh();window.newRefresh=HopeFeatures.refresh();}''')
|
|
self.page.evaluate('''() => {capabilityResolvers[1]({features:{...capabilities,image_input:false}});}''')
|
|
self.page.evaluate('() => newRefresh')
|
|
self.assertNotIn('image/png', self.page.locator('input[type=file]').get_attribute('accept'))
|
|
self.page.evaluate('''() => {capabilityResolvers[0]({features:{...capabilities,image_input:true}});}''')
|
|
self.page.evaluate('() => oldRefresh')
|
|
self.assertNotIn('image/png', self.page.locator('input[type=file]').get_attribute('accept'))
|
|
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.assertIn('Image input is not available', self.page.locator('#featureComposerStatus').inner_text())
|
|
|
|
def test_delayed_reuse_discarded_on_thread_change(self):
|
|
self.page.evaluate("window.dispatchEvent(new CustomEvent('hope:conversation-selected',{detail:{conversation_id:'one'}}))")
|
|
self.page.evaluate("""() => {const original=HopeAuth.request;HopeAuth.request=async(path,options)=>{if(path.startsWith('/api/attachments/')) await new Promise(resolve=>window.resolveReuse=resolve); return original(path,options);};}""")
|
|
self.open_panel('Files')
|
|
self.page.get_by_role('button', name='Use in conversation', exact=True).click()
|
|
self.page.wait_for_function('!!window.resolveReuse')
|
|
self.page.evaluate("window.dispatchEvent(new CustomEvent('hope:conversation-selected',{detail:{conversation_id:'two'}}));resolveReuse()")
|
|
self.page.wait_for_function('document.querySelector(".feature-row-actions button").disabled===false')
|
|
self.assertEqual(self.page.evaluate('HopeFeatures.attachmentIDs()'), [])
|
|
|
|
def test_real_memory_conflict_requires_explicit_reviewed_revision(self):
|
|
self.page.evaluate("""() => {window.latestMemory={section:'memory',content:'Saved facts',revision:'m1'};const original=HopeAuth.request;HopeAuth.request=async(path,options)=>{if(path==='/api/memory')return {sections:[{...latestMemory}]};if(path==='/api/memory/memory'){const data=JSON.parse(options.body);calls.push({path,body:data,method:'PUT'});if(data.revision!==latestMemory.revision)throw Object.assign(new Error('Memory revision conflict. Compare latest saved memory before retrying.'),{status:409});latestMemory={...latestMemory,content:data.content,revision:'m3'};return latestMemory;}return original(path,options);};}""")
|
|
self.open_panel('Memory')
|
|
content = self.page.get_by_label('Assistant memory', exact=True)
|
|
content.fill('My retained local draft')
|
|
self.page.evaluate('latestMemory={section:"memory",content:"Concurrent assistant edit",revision:"m2"}')
|
|
self.page.get_by_role('button', name='Save assistant memory', exact=True).click()
|
|
self.page.wait_for_function('document.getElementById("featureStatus").textContent.includes("retained")')
|
|
self.page.get_by_role('button', name='Compare latest saved memory', exact=True).click()
|
|
self.page.get_by_text('Concurrent assistant edit', exact=True).wait_for()
|
|
self.assertEqual(content.input_value(), 'My retained local draft')
|
|
self.page.get_by_role('button', name='Save assistant memory', exact=True).click()
|
|
self.assertEqual(self.page.evaluate('latestMemory.content'), 'Concurrent assistant edit')
|
|
self.page.get_by_role('button', name='Use reviewed revision for next save', exact=True).click()
|
|
self.page.get_by_role('button', name='Save assistant memory', exact=True).click()
|
|
self.page.wait_for_function('latestMemory.content==="My retained local draft"')
|
|
self.assertEqual(self.page.evaluate('calls.filter(c=>c.path==="/api/memory/memory").at(-1).body.revision'), 'm2')
|
|
|
|
def test_reusable_structures_and_task_hints_explicit_paused(self):
|
|
self.open_panel('Scheduled tasks')
|
|
self.page.get_by_role('button', name='New scheduled task').click()
|
|
self.assertTrue(self.page.get_by_label('Create paused', exact=True).is_checked())
|
|
self.assertTrue(self.page.get_by_label('Create paused', exact=True).is_visible())
|
|
self.assertEqual(self.page.locator('.feature-form .feature-field').count(), 5)
|
|
self.assertIn('timezone and next run are not inferred', self.page.locator('.feature-form').inner_text())
|
|
self.page.evaluate('const target=document.createElement("div");target.id="movedMemory";document.body.append(target);HopeFeatures.mountMemory(target);document.getElementById("libraryDialog").close();HopeFeatures.openSection("memory")')
|
|
self.page.get_by_label('About you', exact=True).wait_for()
|
|
self.assertEqual(self.page.locator('#movedMemory .feature-section').count(), 1)
|
|
self.assertIn('For example', self.page.locator('#movedMemory').inner_text())
|
|
|
|
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.locator('summary[aria-label="Delete notes.txt"]').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('Saved drafts')
|
|
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()
|