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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
import asyncio
import os
import platform
import signal
import subprocess
import choreographer as choreo
import logistro
import pytest
from async_timeout import timeout
from choreographer import errors
# allows to create a browser pool for tests
pytestmark = pytest.mark.asyncio(loop_scope="function")
_logger = logistro.getLogger(__name__)
@pytest.mark.asyncio(loop_scope="function")
async def test_context(headless, sandbox, gpu):
_logger.info("testing...")
if sandbox and "debian" in platform.version().lower():
pytest.skip(
"Debian doesn't support sandbox.",
)
elif sandbox:
_logger.info(
"Not skipping sandbox: "
f"Sandbox: {sandbox},"
f"Version: {platform.version().lower()}",
)
async with timeout(pytest.default_timeout): # type: ignore[reportAttributeAccessIssue]
async with choreo.Browser(
headless=headless,
enable_sandbox=sandbox,
enable_gpu=gpu,
) as browser:
response = await browser.send_command(command="Target.getTargets")
assert "result" in response and "targetInfos" in response["result"] # noqa: PT018 combined assert
if len(response["result"]["targetInfos"]) == 0:
await browser.create_tab()
assert isinstance(browser.get_tab(), choreo.Tab)
tab = browser.get_tab()
assert tab is not None
assert len(tab.sessions) == 1
# let asyncio do some cleaning up if it wants, may prevent warnings
await asyncio.sleep(0)
if hasattr(browser._browser_impl, "tmp_dir"): # noqa: SLF001
assert not browser._browser_impl.tmp_dir.exists # type: ignore[reportAttributeAccessIssue] # noqa: SLF001
@pytest.mark.asyncio(loop_scope="function")
async def test_no_context(headless, sandbox, gpu):
_logger.info("testing...")
if sandbox and "debian" in platform.version().lower():
pytest.skip("Debian doesn't support sandbox.")
browser = await choreo.Browser(
headless=headless,
enable_sandbox=sandbox,
enable_gpu=gpu,
)
try:
async with timeout(pytest.default_timeout): # type: ignore[reportAttributeAccessIssue]
response = await browser.send_command(command="Target.getTargets")
assert "result" in response and "targetInfos" in response["result"] # noqa: PT018 combined assert
if len(response["result"]["targetInfos"]) == 0:
await browser.create_tab()
assert isinstance(browser.get_tab(), choreo.Tab)
tab = browser.get_tab()
assert tab is not None
assert len(tab.sessions) == 1
finally:
await browser.close()
await asyncio.sleep(0)
if hasattr(browser._browser_impl, "tmp_dir"): # noqa: SLF001
assert not browser._browser_impl.tmp_dir.exists # type: ignore[reportAttributeAccessIssue] # noqa: SLF001
# Harass choreographer with a kill in this test to see if its clean in a way
# tempdir may survive protected by chromium subprocess surviving the kill
@pytest.mark.asyncio(loop_scope="function")
async def test_watchdog(headless):
_logger.info("testing...")
browser = await choreo.Browser(
headless=headless,
)
if platform.system() == "Windows":
# Blocking process here because it ensures the kill will occur rn
subprocess.call( # noqa: S603, ASYNC221 sanitize input, blocking process
["taskkill", "/F", "/T", "/PID", str(browser.subprocess.pid)], # noqa: S607
stderr=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
)
else:
os.kill(browser.subprocess.pid, signal.SIGKILL)
await asyncio.sleep(0.5)
with pytest.raises(
(errors.ChannelClosedError, errors.BrowserClosedError, asyncio.CancelledError),
):
await browser.send_command(command="Target.getTargets")
await browser.close()
await asyncio.sleep(0)
|
