examples_async.py
December 14, 2024
import asyncio from pathlib import Path from typing import List
from torequests.dummy import Requests
from ichrome import AsyncChromeDaemon, ChromeEngine from ichrome.async_utils import AsyncChrome, AsyncTab, Tag, logger
logger.setLevel('DEBUG')
AsyncTab._log_all_recv = True
headless = False
headless = True
async def test_chrome(chrome: AsyncChrome): assert str(chrome) == "<Chrome(connected): http://127.0.0.1:9222>" assert chrome.server == "http://127.0.0.1:9222" version = await chrome.version assert isinstance(version, dict) and "Browser" in version ok = await chrome.check() assert ok is True, ok ok = await chrome.ok assert ok is True, ok resp = await chrome.get_server("json") assert isinstance(resp.json(), list) await chrome.new_tab() tabs1: List[AsyncTab] = await chrome.get_tabs() tabs2: List[AsyncTab] = await chrome.tabs assert tabs1 == tabs2 and tabs1 and tabs2, (tabs1, tabs2) tab0: AsyncTab = await chrome.get_tab(0) tab0_by_getitem = await chrome[0] assert tab0 == tab0_by_getitem assert tabs1[0] == tab0, (tabs1, tab0) tab1: AsyncTab = await chrome.new_tab() assert isinstance(tab1, AsyncTab) await asyncio.sleep(0.2) await chrome.activate_tab(tab0) # test batch connect multiple tabs async with chrome.connect_tabs([tab0, tab1]): assert tab0.status == "connected", (tab0.status, tab0) assert tab1.status == "connected", (tab1.status, tab1) # watch the tabs switch await tab1.activate_tab() await asyncio.sleep(0.2) await tab0.activate_tab() await asyncio.sleep(0.2) await tab1.activate_tab() # test connect single tab async with chrome.connect_tabs(tab0): assert tab0.status == "connected" await chrome.close_tab(tab1)
async def test_tab_ws(tab: AsyncTab): # test msg_id auto increase assert tab.msg_id == tab.msg_id - 1 assert tab.status == "disconnected", tab.status assert tab.ws is None or (tab.flatten and not tab._session_id) async with tab(): assert tab.ws assert tab.status == "connected" # assert no connection out of async context. assert tab.status == "disconnected" assert tab.ws is None or (tab.flatten and not tab._session_id)
async def test_send_msg(tab: AsyncTab): # test send msg assert ( tab.get_data_value( await tab.send("Network.enable"), value_path="value", default={} ) == {} ) # disable Network await tab.disable("Network")
async def test_tab_cookies(tab: AsyncTab): await tab.clear_browser_cookies() assert len(await tab.get_cookies(urls="http://httpbin.org/get")) == 0 assert await tab.set_cookie("test", "test_value", url="https://github.com/") assert await tab.set_cookie("test2", "test_value", url="https://github.com/") assert await tab.set_cookies( [{"name": "a", "value": "b", "url": "http://httpbin.org/get"}] ) assert len(await tab.get_cookies(urls="https://github.com/")) == 2 assert len(await tab.get_cookies(urls="http://httpbin.org/get")) == 1 assert await tab.delete_cookies("test", url="https://github.com/") assert len(await tab.get_cookies(urls="https://github.com/")) == 1 # get all Browser cookies assert len(await tab.get_all_cookies()) > 0
async def test_browser_context( tab1: AsyncTab, chrome: AsyncChrome, chromed: AsyncChromeDaemon ): # test chrome.create_context assert len(await tab1.get_all_cookies()) > 0 async with chrome.create_context(proxyServer=None) as context: async with context.new_tab() as tab: assert len(await tab.get_all_cookies()) == 0 await tab.set_cookie("a", "1", "http://httpbin.org") assert len(await tab.get_all_cookies()) > 0
# test chrome_daemon.create_context
assert len(await tab1.get_all_cookies()) > 0
async with chromed.create_context(proxyServer=None) as context:
async with context.new_tab() as tab:
assert len(await tab.get_all_cookies()) == 0
await tab.set_cookie("a", "1", "http://httpbin.org")
assert len(await tab.get_all_cookies()) > 0
# test chrome.incognito_tab
assert len(await tab1.get_all_cookies()) > 0
async with chrome.incognito_tab(proxyServer=None) as tab:
assert len(await tab.get_all_cookies()) == 0
await tab.set_cookie("a", "1", "http://httpbin.org")
assert len(await tab.get_all_cookies()) > 0
# test chrome_daemon.incognito_tab
assert len(await tab1.get_all_cookies()) > 0
async with chromed.incognito_tab(proxyServer=None) as tab:
assert len(await tab.get_all_cookies()) == 0
await tab.set_cookie("a", "1", "http://httpbin.org")
assert len(await tab.get_all_cookies()) > 0
async def test_tab_set_url(tab: AsyncTab): # set new url for this tab, timeout will stop loading for timeout_stop_loading defaults to True assert not (await tab.set_url("https://postman-echo.com/delay/5", timeout=1)) assert await tab.set_url("https://postman-echo.com/delay/1", timeout=5) assert await tab.set_url("https://postman-echo.com/ip", timeout=5) mhtml_len = len(await tab.snapshot_mhtml()) assert mhtml_len in range(1, 1000), f"test snapshot failed {mhtml_len}" await tab.goto("https://bing.com?setlang=en&cc=US") assert await tab.wait_tag( "#sb_form_q", max_wait_time=10 ), "wait_tag failed for #sb_form_q"
async def test_tab_js(tab: AsyncTab): await tab.goto("https://staticfile.org/about_en.html", timeout=5) # test js update title await tab.js("document.title = 'abc'") # test js_code assert (await tab.js_code("return document.title")) == "abc" # test findall await tab.js("document.title = 123456789") assert (await tab.findall("<title>(.?)</title>")) == ["123456789"] assert (await tab.findall("<title>.?</title>")) == ["<title>123456789</title>"] assert (await tab.findall("<title>(1)(2).?</title>")) == [["1", "2"]] assert (await tab.findall("<title>(1)(2).?</title>", "body")) == [] new_title = await tab.current_title # test refresh_tab_info for tab meta info assert (await tab.title) == new_title assert tab._title != new_title assert await tab.refresh_tab_info() assert tab._title == new_title # inject JS timeout return None if not headless: temp = await tab.js("alert()") assert temp is None # close the alert dialog assert await tab.handle_dialog(accept=True) # inject js url: vue.js # get window.Vue variable before injecting vue_obj = await tab.js("window.Vue", "result.result.type") # {'id': 22, 'result': {'result': {'type': 'undefined'}}} assert vue_obj == "undefined" assert await tab.inject_js_url( "https://cdn.staticfile.org/vue/2.6.10/vue.min.js", timeout=3 ) vue_obj = await tab.js("window.Vue", value_path=None) # {'id': 23, 'result': {'result': {'type': 'function', 'className': 'Function', 'description': 'function wn(e){this._init(e)}', 'objectId': '{"injectedScriptId":1,"id":1}'}}} assert "Function" in str(vue_obj) tag = await tab.querySelector("#not-exist") assert not tag # querySelectorAll with JS, return list of Tag object tags = await tab.querySelectorAll("#logo") assert tags, f"{[tags, type(tags)]}" assert isinstance(tags[0], Tag), f"{[tags[0], type(tags[0])]}" assert tags[0].tagName in {"a"}, f"{[tags[0], tags[0].tagName]}" # querySelectorAll with JS, index arg is Not None, return Tag or None one_tag = await tab.querySelectorAll("#logo", index=0) assert isinstance(one_tag, Tag), type(one_tag) await tab.stop_loading_page() assert await tab.set_html("") current_html = await tab.current_html assert current_html == "<html><head></head><body></body></html>" # reload the page for _ in range(2): assert await tab.reload() await tab.wait_loading(5, timeout_stop_loading=True) current_html = await tab.html if current_html: break # print(current_html) assert len(current_html) > 100, repr(current_html) # test wait tags result = await tab.wait_tags("abcdabcdabcdabcd", max_wait_time=1) assert result == [] assert await tab.wait_tag("#logo", max_wait_time=3) assert await tab.includes("This repo is open-source. Licensed under MIT.") assert not (await tab.includes("abcdabcdabcdabcd")) assert await tab.wait_includes("This repo is open-source. Licensed under MIT.") assert (await tab.wait_includes("abcdabcdabcdabcd", max_wait_time=1)) is False assert await tab.wait_findall("This repo is open-source. Licensed under MIT.") assert (await tab.wait_findall("abcdabcdabcdabcd", max_wait_time=1)) == [] # test wait_console_value await tab.js("setTimeout(() => {console.log(123)}, 2);") assert (await tab.wait_console_value()) == 123
async def test_wait_response(tab: AsyncTab): # wait_response with filter_function # raw response: {'method': 'Network.responseReceived', 'params': {'requestId': '1000003000.69', 'loaderId': 'D7814CD633EDF3E699523AF0C4E9DB2C', 'timestamp': 207483.974238, 'type': 'Script', 'response': {'url': 'https://www.python.org/static/js/libs/masonry.pkgd.min.js', 'status': 200, 'statusText': '', 'headers': {'date': 'Sat, 05 Oct 2019 08:18:34 GMT', 'via': '1.1 vegur, 1.1 varnish, 1.1 varnish', 'last-modified': 'Tue, 24 Sep 2019 18:31:03 GMT', 'server': 'nginx', 'age': '290358', 'etag': '"5d8a60e7-6643"', 'x-served-by': 'cache-iad2137-IAD, cache-tyo19928-TYO', 'x-cache': 'HIT, HIT', 'content-type': 'application/x-javascript', 'status': '200', 'cache-control': 'max-age=604800, public', 'accept-ranges': 'bytes', 'x-timer': 'S1570263515.866582,VS0,VE0', 'content-length': '26179', 'x-cache-hits': '1, 170'}, 'mimeType': 'application/x-javascript', 'connectionReused': False, 'connectionId': 0, 'remoteIPAddress': '151.101.108.223', 'remotePort': 443, 'fromDiskCache': True, 'fromServiceWorker': False, 'fromPrefetchCache': False, 'encodedDataLength': 0, 'timing': {'requestTime': 207482.696803, 'proxyStart': -1, 'proxyEnd': -1, 'dnsStart': -1, 'dnsEnd': -1, 'connectStart': -1, 'connectEnd': -1, 'sslStart': -1, 'sslEnd': -1, 'workerStart': -1, 'workerReady': -1, 'sendStart': 0.079, 'sendEnd': 0.079, 'pushStart': 0, 'pushEnd': 0, 'receiveHeadersEnd': 0.836}, 'protocol': 'h2', 'securityState': 'unknown'}, 'frameId': 'A2971702DE69F008914F18EAE6514DD5'}}
# listening response
def filter_function(r):
url = r["params"]["response"]["url"]
ok = "cdn.staticfile.org/vue/2.6.10/vue.min.js" in url
logger.warning("get response url: %s %s" % (url, ok))
return ok
task = asyncio.ensure_future(
tab.wait_response(
filter_function=filter_function, response_body=True, timeout=10
)
)
await tab.set_url("https://cdn.staticfile.org/vue/2.6.10/vue.min.js")
result = await task
ok = "Released under the MIT License" in result["data"]
logger.warning(f"check wait_response callback, get_response {ok}")
assert ok
# test wait_response_context
def filter_function2(r):
try:
url = r["params"]["response"]["url"]
ok = "cdn.staticfile.org/vue/2.6.10/vue.min.js" in url
return ok
except KeyError:
pass
async with tab.wait_response_context(
filter_function=filter_function2, timeout=5
) as r:
await tab.goto("https://cdn.staticfile.org/vue/2.6.10/vue.min.js")
result = (await r) or {}
assert "Released under the MIT License" in result.get("data", ""), result
async def test_tab_js_onload(tab: AsyncTab): # add js onload js_id = await tab.add_js_onload(source="window.title=123456789") assert js_id await tab.set_url("https://bing.com") assert (await tab.get_variable("window.title")) == 123456789 # remove js onload assert await tab.remove_js_onload(js_id) await tab.set_url("https://bing.com") assert (await tab.get_variable("window.title")) != 123456789 assert (await tab.get_variable("[1, 2, 3]")) != [1, 2, 3] assert (await tab.get_variable("[1, 2, 3]", jsonify=True)) == [1, 2, 3] current_url = await tab.current_url url = await tab.url assert "bing.com" in url and "bing.com" in current_url, (url, current_url)
async def test_tab_current_html(tab: AsyncTab): html = await tab.get_html() assert "Customer name:" in html # alias current_html assert html == (await tab.current_html) == (await tab.html)
async def test_tab_screenshot(tab: AsyncTab): # screenshot screen = await tab.screenshot() part = await tab.screenshot_element("fieldset") assert screen assert part assert len(screen) > len(part)
async def test_tab_set_ua_headers(tab: AsyncTab): # test set_ua await tab.set_ua("Test UA") # test set_headers await tab.set_headers({"A": "1", "B": "2"}) for _ in range(3): if await tab.set_url("http://httpbin.org/get", timeout=10): break html = await tab.get_html() assert ( '"A": "1"' in html and '"B": "2"' in html and '"User-Agent": "Test UA"' in html )
async def test_tab_keyboard_mouse(tab: AsyncTab):
if "httpbin.org/forms/post" not in (await tab.current_url):
await tab.set_url("http://httpbin.org/forms/post", timeout=5)
rect = await tab.get_bounding_client_rect('[type="tel"]')
await tab.mouse_click(rect["left"], rect["top"], count=1)
await tab.keyboard_send(text="1")
await tab.keyboard_send(text="2")
await tab.keyboard_send(text="3")
await tab.keyboard_send(string="123")
await tab.mouse_click(rect["left"], rect["top"], count=2)
selection = await tab.get_variable("window.getSelection().toString()")
assert selection == "123123"
# test mouse_click_element
await tab.mouse_click_element_rect('[type="tel"]')
selection = await tab.get_variable("window.getSelection().toString()")
assert selection == ""
# test mouse_drag_rel_chain draw a square, sometime witeboard load failed.....
await tab.set_url("https://zhoushuo.me/drawingborad/", timeout=5)
await tab.mouse_click(5, 5)
# drag with moving mouse, released on each await
walker = (
await tab.mouse_drag_rel_chain(320, 145)
.move(50, 0, 0.2)
.move(0, 50, 0.2)
.move(-50, 0, 0.2)
.move(0, -50, 0.2)
)
await walker.move(50 * 1.414, 50 * 1.414, 0.2)
async def test_iter_events(tab: AsyncTab): async with tab.iter_events(["Page.loadEventFired"], timeout=8) as events_iter: await tab.goto("http://bing.com", timeout=0) data = await events_iter assert data, data
await tab.goto("http://bing.com", timeout=0)
data = await events_iter.get()
assert data, data
await tab.goto("http://bing.com", timeout=0)
async for data in events_iter:
assert data
break
def cb(event, tab, buffer):
return tab
async with tab.iter_events({"Page.loadEventFired": cb}, timeout=8) as events_iter:
await tab.goto("http://bing.com", timeout=0)
data = await events_iter
assert type(data) == type(tab), data
# test iter_fetch
async with tab.iter_fetch(
patterns=[{"urlPattern": "*postman-echo.com*", "resourceType": "Document"}]
) as f:
url = "https://postman-echo.com/get"
await tab.goto(url, timeout=0)
data = await f
# print(data, flush=True)
assert data
# test continueRequest
await f.continueRequest(data)
assert await tab.wait_includes('"host"', max_wait_time=5), await tab.html
await tab.goto(url, timeout=0)
data = await f
assert data
# test modify response
await f.fulfillRequest(data, 200, body=b"hello world.")
assert await tab.wait_includes("hello world.", max_wait_time=3)
await tab.goto(url, timeout=0)
data = await f
assert data
await f.failRequest(data, "AccessDenied")
assert (await tab.url).startswith("chrome-error://")
# response fetch
url = "https://postman-echo.com/get"
RequestPatternList = [
{"urlPattern": "*postman-echo.com*", "requestStage": "Response"}
]
async with tab.iter_fetch(RequestPatternList) as f:
await tab.goto(url, timeout=0)
# only one request could be catched
event = await f
assert f.match_event(event, RequestPatternList[0]), event
# print('request event:', json.dumps(event), flush=True)
response = await f.get_response(event, timeout=5)
# print('response body:', response)
assert '"host"' in response["data"], response
async def test_init_tab(chromed: AsyncChromeDaemon): # test init tab from chromed, create new tab and auto_close=True async with chromed.connect_tab(index=None, auto_close=True) as tab: TEST_DEFAULT_CB_OK = False
def test_default_cb(tab, data_dict):
nonlocal TEST_DEFAULT_CB_OK
TEST_DEFAULT_CB_OK = True
tab.default_recv_callback.append(test_default_cb)
# await tab.goto("https://staticfile.org/about_en.html", timeout=5)
# title = await tab.current_title
# assert TEST_DEFAULT_CB_OK, "test default_recv_callback failed"
# assert title == "About - StaticFile CDN", repr(title)
tab.default_recv_callback.clear()
async def test_fetch_context(tab: AsyncTab): async with tab.iter_fetch(patterns=[{"urlPattern": "bing.com"}]) as f: await tab.goto("http://bing.com", timeout=0) async for r in f: assert "bing.com" in r["params"]["request"]["url"] await f.continueRequest(r) break
async def cb(event, tab, buffer):
assert "bing.com" in r["params"]["request"]["url"]
await buffer.continueRequest(event)
async with tab.iter_fetch(
patterns=[{"urlPattern": "*bing.com*"}],
callback=cb,
) as f:
await tab.goto("http://bing.com", timeout=0)
async for r in f:
break
async def test_port_forwarding(host, port): from ichrome.utils import PortForwarder
dst_port = port + 100
async with PortForwarder((host, port), (host, dst_port)):
r = await Requests().get(f"http://{host}:{dst_port}/json/version", timeout=5)
assert "webSocketDebuggerUrl" in r.text, r.text
r = await Requests().get(f"http://{host}:{dst_port}/json/version", timeout=5)
assert "webSocketDebuggerUrl" not in r.text, r.text
async def test_duplicated_key_error(tab: AsyncTab): _task = asyncio.create_task(tab.wait_loading(timeout=3)) try: await asyncio.create_task(tab.wait_loading(timeout=3)) except ValueError: return await _task raise AssertionError
async def test_tab_new_tab(chromed: AsyncChromeDaemon): async with chromed.incognito_tab() as tab: url = "http://www.bing.com/" await tab.goto(url, timeout=3) MUIDB = (await tab.get_cookies_dict([url])).get("MUIDB") new_tab = await tab.new_tab() tab_exist = bool(await tab.chrome.get_tab(new_tab.tab_id)) assert tab_exist async with new_tab(auto_close=True) as tab: # same context, so same cookie MUIDB2 = (await tab.get_cookies_dict([url])).get("MUIDB") # print(MUIDB, MUIDB2, MUIDB == MUIDB2) assert MUIDB == MUIDB2 # the new_tab auto closed tab_exist = bool(await tab.chrome.get_tab(new_tab.tab_id)) assert not tab_exist
async def test_examples(): def on_startup(chromed): chromed.started = 1
def on_shutdown(chromed):
chromed.__class__.bye = 1
host = "127.0.0.1"
port = 9222
async with AsyncChromeDaemon(
host=host,
port=port,
headless=headless,
on_startup=on_startup,
on_shutdown=on_shutdown,
clear_after_shutdown=True,
) as chromed:
# test utils port forwarding
await test_port_forwarding(host, port)
await test_init_tab(chromed)
logger.info("test init tab from chromed OK.")
# test on_startup
assert chromed.started
logger.info("test on_startup OK.")
# test tab.new_tab
await test_tab_new_tab(chromed)
logger.info("test_tab_new_tab OK.")
# ===================== Chrome Test Cases =====================
async with AsyncChrome(host=host, port=port) as chrome:
# memory = chrome.get_memory()
# assert memory > 0
# logger.info("get_memory OK. (%s MB)" % memory)
# await test_chrome(chrome)
# logger.info("test_chrome OK.")
# # ===================== Tab Test Cases =====================
# # Duplicate, use async with chrome.connect_tab(None) instead
tab: AsyncTab = await chrome.new_tab()
# assert not tab._target_info
# await test_tab_ws(tab)
# # same as: async with tab.connect():
async with tab():
# assert tab.info
# # test duplicated event key issue
# await test_duplicated_key_error(tab)
# logger.info("test_duplicated_key_error OK.")
# # test send raw message
# await test_send_msg(tab)
# logger.info("test_send_msg OK.")
# # test cookies operations
# await test_tab_cookies(tab)
# logger.info("test_tab_cookies OK.")
# # test_browser_context
# await test_browser_context(tab, chrome, chromed)
# logger.info("test_browser_context OK.")
# # set url
# await test_tab_set_url(tab)
# logger.info("test_tab_set_url OK.")
# # test js
# await test_tab_js(tab)
# logger.info("test_tab_js OK.")
# # test wait_response
await test_wait_response(tab)
logger.info("test_wait_response OK.")
# # test fetch_context
# await test_fetch_context(tab)
# logger.info("test_fetch_context OK.")
# # test add_js_onload remove_js_onload
# await test_tab_js_onload(tab)
# logger.info("test_tab_js_onload OK.")
# # test iter_events iter_fetch
# await test_iter_events(tab)
# logger.info("test_iter_events OK.")
# # test set ua and set headers
# await test_tab_set_ua_headers(tab)
# logger.info("test_tab_set_ua_headers OK.")
# # load url for other tests
# await tab.set_url("http://httpbin.org/forms/post")
# # test current_html
# await test_tab_current_html(tab)
# logger.info("test_tab_current_html OK.")
# # test screenshot
# await test_tab_screenshot(tab)
# logger.info("test_tab_screenshot OK.")
# # test double click some positions. test keyboard_send input
# await test_tab_keyboard_mouse(tab)
# logger.info("test_tab_keyboard_mouse OK.")
# # clear cache
# assert await tab.clear_browser_cache()
# logger.info("clear_browser_cache OK.")
# # close tab
# await tab.close()
# test chrome.connect_tab
# async with chrome.connect_tab(chrome.server + "/json", True) as tab:
# await tab.wait_loading(2)
# assert "webSocketDebuggerUrl" in (await tab.current_html)
# logger.info("test connect_tab OK.")
# # close_browser gracefully, I have no more need of chrome instance
# await chrome.close_browser()
# # await chrome.kill()
# sep = f'\n{"=" * 80}\n'
# logger.critical(
# f"{sep}Congratulations, all test cases passed(flatten={AsyncTab._DEFAULT_FLATTEN}).{sep}"
# )
assert AsyncChromeDaemon.bye
# test clear_after_shutdown
_user_dir_path = AsyncChromeDaemon.DEFAULT_USER_DIR_PATH / "chrome_9222"
dir_cleared = not _user_dir_path.is_dir()
assert dir_cleared
await asyncio.sleep(1)
async def test_chrome_engine(): async def _test_chrome_engine(): tab_callback1 = r"""async def tab_callback(self, tab, url, timeout): await tab.set_url(url, timeout=5) return 'Bing' in (await tab.title)"""
async def tab_callback2(self, tab, url, timeout):
await tab.set_url(url, timeout=5)
return "Bing" in (await tab.title)
async with ChromeEngine(
max_concurrent_tabs=5,
headless=True,
disable_image=True,
after_shutdown=lambda cd: cd.clear_dir_with_shutil(cd.user_data_dir),
) as ce:
# test normal usage
tasks = [
asyncio.create_task(
ce.do("https://bing.com", tab_callback1, timeout=10)
)
for _ in range(3)
] + [
asyncio.create_task(
ce.do("https://bing.com", tab_callback2, timeout=10)
)
for _ in range(3)
]
for task in tasks:
assert (await task) is True
# test screenshot full screen and partial tag range.
tasks = [
asyncio.create_task(
ce.screenshot("https://bing.com", "#sbox", timeout=10)
),
asyncio.create_task(ce.screenshot("https://bing.com", timeout=10)),
]
results = [await task for task in tasks]
assert 1000 < len(results[0]) < len(results[1])
# test download
tasks = [
asyncio.create_task(
ce.download("https://bing.com", "#sbox", timeout=10)
),
asyncio.create_task(ce.download("https://bing.com", timeout=10)),
]
results = [await task for task in tasks]
assert 1000 < len(results[0]["tags"][0]) < len(results[1]["html"])
# test connect_tab
async with ce.connect_tab() as tab:
await tab.goto("http://pypi.org")
title = await tab.title
assert "PyPI" in title, title
logger.critical("test_chrome_engine OK")
_user_dir_path = (
AsyncChromeDaemon.DEFAULT_USER_DIR_PATH
/ f"chrome_{ChromeEngine.START_PORT}"
)
dir_cleared = not _user_dir_path.is_dir()
#assert dir_cleared
await _test_chrome_engine()
def test_all(): import time
AsyncChromeDaemon.DEFAULT_USER_DIR_PATH = Path("./ichrome_user_data")
loop = asyncio.new_event_loop()
try:
for flatten in [True, False, True]:
logger.critical("Start testing flatten=%s." % flatten)
AsyncTab._DEFAULT_FLATTEN = flatten
# loop.run_until_complete(test_chrome_engine())
# time.sleep(1)
loop.run_until_complete(test_examples())
time.sleep(1)
finally:
loop.close()
if name == "main": test_all()