| """Un-routed streaming transport acceptance for actual overlay removal.""" |
| import contextlib |
| import json |
| import threading |
| import time |
| import unittest |
| import urllib.parse |
| from unittest.mock import patch |
|
|
| from tests.test_reader_performance import StaticHandler, static_server, sync_playwright, PlaywrightError, minimal_pdf |
|
|
|
|
| @unittest.skipIf(sync_playwright is None, 'Playwright is unavailable') |
| class ReaderCloseTrafficTests(unittest.TestCase): |
| def test_close_mid_transfer_stops_bytes_and_next_book_opens(self): |
| original = StaticHandler.do_GET |
| lock = threading.Lock() |
| transfers = [] |
| payload = minimal_pdf() + b'\0' * (8 * 1024 * 1024) |
| source = 'https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/closing.pdf' |
|
|
| def serve(handler): |
| path = urllib.parse.urlsplit(handler.path) |
| if path.path != '/api/reader-content': |
| return original(handler) |
| target = urllib.parse.parse_qs(path.query)['url'][0] |
| if target.endswith('.txt'): |
| body = b'Next book is readable' |
| handler.send_response(200) |
| handler.send_header('Content-Type', 'text/plain') |
| handler.send_header('Content-Length', str(len(body))) |
| handler.end_headers() |
| handler.wfile.write(body) |
| return |
| record = {'sent': 0, 'done': False} |
| with lock: |
| transfers.append(record) |
| try: |
| handler.send_response(200) |
| handler.send_header('Content-Type', 'application/pdf') |
| handler.send_header('Content-Length', str(len(payload))) |
| handler.send_header('Accept-Ranges', 'bytes') |
| handler.end_headers() |
| for offset in range(0, len(payload), 4096): |
| handler.wfile.write(payload[offset:offset+4096]) |
| handler.wfile.flush() |
| with lock: |
| record['sent'] += 4096 |
| time.sleep(.01) |
| except (BrokenPipeError, ConnectionResetError): |
| pass |
| finally: |
| with lock: |
| record['done'] = True |
|
|
| with contextlib.ExitStack() as stack: |
| stack.enter_context(patch.object(StaticHandler, 'do_GET', serve)) |
| origin = stack.enter_context(static_server()) |
| playwright = stack.enter_context(sync_playwright()) |
| try: |
| browser = playwright.chromium.launch(headless=True, args=['--no-sandbox']) |
| except PlaywrightError as error: |
| raise unittest.SkipTest(str(error)) |
| stack.callback(browser.close) |
| context = browser.new_context(service_workers='block') |
| stack.callback(context.close) |
| page = context.new_page() |
| page.set_default_timeout(10000) |
| page.add_init_script('''const nativeFetch=window.fetch; |
| window.fetch=(input,init) => { |
| const url=new URL(input instanceof Request?input.url:input,location.href); |
| return nativeFetch(url.pathname==='/api/reader-content'?location.origin+url.pathname+url.search:input,init); |
| };''') |
| page.goto(origin+'/static/manifest.json') |
| page.add_script_tag(url=origin+'/static/search/reader-navigation.js') |
| page.evaluate('''url => { |
| window.__navigation = VoiceOfMLReaderNavigation.createNavigation('/static/reader.html'); |
| window.__reader = __navigation.mount(new URL(url)); |
| }''', origin+'/static/reader.html?'+urllib.parse.urlencode(dict(url=source, ext='pdf'))) |
| deadline = time.monotonic()+10 |
| while time.monotonic() < deadline: |
| with lock: |
| started = any(row['sent'] > 0 for row in transfers) |
| if started: |
| break |
| page.wait_for_timeout(20) |
| self.assertTrue(started) |
| result = page.evaluate('''() => { |
| let phase; |
| __reader.contentWindow.addEventListener('voice-reader-dispose', () => { |
| phase=__reader.contentDocument.documentElement.dataset.readerPhase; |
| }); |
| __navigation.unmount(); |
| return phase; |
| }''') |
| self.assertEqual(result, 'disposed') |
| with lock: |
| before = sum(row['sent'] for row in transfers) |
| count = len(transfers) |
| page.wait_for_timeout(750) |
| with lock: |
| self.assertTrue(all(row['done'] for row in transfers), transfers) |
| self.assertEqual(len(transfers), count) |
| self.assertLess(sum(row['sent'] for row in transfers)-before, 65536) |
| next_source = source.replace('.pdf', '.txt') |
| page.evaluate('url => __navigation.mount(new URL(url))', origin+'/static/reader.html?'+urllib.parse.urlencode(dict(url=next_source, ext='txt'))) |
| frame = page.frame_locator('.reader-overlay') |
| frame.locator('html[data-reader-phase="ready"]').wait_for() |
| self.assertEqual(frame.locator('.reader-text').inner_text(), 'Next book is readable') |
|
|
| def test_touch_tap_clears_selection_in_text_pdf_and_isolated_html(self): |
| with contextlib.ExitStack() as stack: |
| origin = stack.enter_context(static_server()) |
| playwright = stack.enter_context(sync_playwright()) |
| browser = playwright.chromium.launch(headless=True, args=['--no-sandbox']) |
| stack.callback(browser.close) |
| context = browser.new_context(has_touch=True, is_mobile=True, viewport={'width':390,'height':844}) |
| stack.callback(context.close) |
| page = context.new_page() |
| for extension in ('txt', 'html', 'pdf'): |
| with self.subTest(extension=extension): |
| body = minimal_pdf() if extension == 'pdf' else b'<p>Selectable book text</p><p style="height:1600px">Other book text</p>' |
| page.route('**/api/reader-content?**', lambda route: route.fulfill(body=body, content_type='application/pdf' if extension == 'pdf' else 'text/html')) |
| source = 'https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/touch.'+extension |
| page.goto(origin+'/static/reader.html?'+urllib.parse.urlencode(dict(url=source, ext=extension))) |
| page.locator('html[data-reader-phase="ready"]').wait_for() |
| target = page.frame_locator('.html-frame').locator('p').first if extension == 'html' else page.locator('.reader-pdf-text-run' if extension == 'pdf' else '.reader-text').first |
| target.evaluate('''node => { |
| const doc=node.ownerDocument, range=doc.createRange(); |
| range.selectNodeContents(node); doc.getSelection().removeAllRanges(); doc.getSelection().addRange(range); |
| }''') |
| self.assertTrue(target.evaluate('node => !!node.ownerDocument.getSelection().toString()')) |
| target.dispatch_event('pointerdown', {'pointerType':'touch','pointerId':41,'isPrimary':True,'clientX':100,'clientY':100}) |
| page.wait_for_timeout(350) |
| target.dispatch_event('pointerup', {'pointerType':'touch','pointerId':41,'isPrimary':True,'clientX':100,'clientY':100}) |
| self.assertTrue(target.evaluate('node => !!node.ownerDocument.getSelection().toString()')) |
| preserved = target.evaluate('''node => { |
| const doc=node.ownerDocument, selection=doc.getSelection(), output=[]; |
| const send = (type, values={}) => node.dispatchEvent(new PointerEvent(type, |
| {bubbles:true, composed:true, pointerType:'touch', pointerId:41, isPrimary:true, |
| clientX:100,clientY:100,...values})); |
| for (const action of ['drag','cancel','changed','secondary']) { |
| const range=doc.createRange();range.selectNodeContents(node); |
| selection.removeAllRanges();selection.addRange(range); |
| send('pointerdown', action==='secondary'?{isPrimary:false}:{}); |
| if (action==='drag') send('pointermove',{clientX:120}); |
| if (action==='cancel') send('pointercancel'); |
| if (action==='changed') { |
| const text = node.ownerDocument.createTreeWalker(node, NodeFilter.SHOW_TEXT).nextNode(); |
| selection.setBaseAndExtent(text, 0, text, Math.min(3, text.data.length)); |
| } |
| send('pointerup'); |
| output.push(!!selection.toString()); |
| } |
| const range=doc.createRange();range.selectNodeContents(node); |
| selection.removeAllRanges();selection.addRange(range); |
| return output; |
| }''') |
| self.assertEqual(preserved, [True]*4) |
| box = target.bounding_box() |
| page.touchscreen.tap(min(370, box['x']+box['width']-2), box['y']+min(box['height']-2, 20)) |
| self.assertFalse(target.evaluate('node => !!node.ownerDocument.getSelection().toString()')) |
| page.unroute('**/api/reader-content?**') |
|
|
