File size: 3,675 Bytes
2ecdf7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | import contextlib
import json
import unittest
import urllib.parse
from tests.test_reader_performance import (
PlaywrightError,
static_server,
sync_playwright,
)
SOURCE = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/timeout.txt"
TIMEOUT_SCRIPT = """
const nativeTimeout = window.setTimeout.bind(window);
window.__requestTimeouts = [];
window.setTimeout = (callback, delay, ...args) =>
nativeTimeout(callback, delay === 120000 ? 50 : delay, ...args);
const nativeFetch = window.fetch.bind(window);
window.fetch = (input, init = {}) => {
const url = new URL(String(input), location.href);
if (!url.pathname.startsWith('/api/reader-content') &&
!url.pathname.startsWith('/api/reader-resolve') &&
!url.pathname.endsWith('/timeout.txt')) return nativeFetch(input, init);
const signal = init.signal;
signal.addEventListener('abort', () => {
window.__requestTimeouts.push(signal.reason.name);
}, { once: true });
if (window.__timeoutScenario === 'body') {
return Promise.resolve(new Response(new ReadableStream({
start(controller) {
signal.addEventListener('abort', () => controller.error(signal.reason),
{ once: true });
}
}), { headers: { 'Content-Type': 'text/plain' } }));
}
return new Promise((resolve, reject) => {
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
};
"""
@unittest.skipIf(sync_playwright is None, "install requirements-test.txt to run Reader timeout tests")
class ReaderTimeoutTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
resources = contextlib.ExitStack()
cls.addClassCleanup(resources.close)
cls.origin = resources.enter_context(static_server())
playwright = resources.enter_context(sync_playwright())
try:
cls.browser = playwright.chromium.launch(headless=True, args=["--no-sandbox"])
except PlaywrightError as error:
raise unittest.SkipTest(f"Chromium is unavailable: {error}")
cls.addClassCleanup(cls.browser.close)
def assert_timeout_ui(self, scenario):
with self.browser.new_context(service_workers="block") as context:
page = context.new_page()
page_errors = []
page.on("pageerror", lambda error: page_errors.append(str(error)))
page.add_init_script(
f"window.__timeoutScenario = {json.dumps(scenario)};\n" + TIMEOUT_SCRIPT
)
query = {"ext": "txt", "title": "Timeout"}
query.update({"id": "timeout-id"} if scenario == "resolve" else {"url": SOURCE})
page.goto(
f"{self.origin}/static/reader.html?{urllib.parse.urlencode(query)}",
wait_until="domcontentloaded",
)
page.locator(".reader-error").wait_for(state="visible", timeout=10000)
self.assertEqual(page.locator("html").get_attribute("data-reader-phase"), "failed")
self.assertEqual(page.locator("#content").get_attribute("data-error-code"), "READER_NETWORK")
self.assertEqual(page.locator(".reader-loading-indicator").count(), 0)
names = page.evaluate("window.__requestTimeouts")
self.assertTrue(names)
self.assertEqual(set(names), {"TimeoutError"})
self.assertEqual(page_errors, [])
def test_txt_timeout_before_headers_shows_failure(self):
self.assert_timeout_ui("headers")
def test_txt_timeout_after_headers_shows_failure(self):
self.assert_timeout_ui("body")
def test_id_resolution_timeout_shows_failure(self):
self.assert_timeout_ui("resolve")
|