import dataclasses
import functools
import math
import os
import pdb
import re
import sys
import time
import traceback
import urllib
from dataclasses import dataclass
import py
import pytest
import toml
from playwright.sync_api import Error as PlaywrightError
ROOT = py.path.local(__file__).dirpath("..", "..", "..")
BUILD = ROOT.join("pyscript.core").join("dist")
def params_with_marks(params):
"""
Small helper to automatically apply to each param a pytest.mark with the
same name of the param itself. E.g.:
params_with_marks(['aaa', 'bbb'])
is equivalent to:
[pytest.param('aaa', marks=pytest.mark.aaa),
pytest.param('bbb', marks=pytest.mark.bbb)]
This makes it possible to use 'pytest -m aaa' to run ONLY the tests which
uses the param 'aaa'.
"""
return [pytest.param(name, marks=getattr(pytest.mark, name)) for name in params]
def with_execution_thread(*values):
"""
Class decorator to override config.execution_thread.
By default, we run each test twice:
- execution_thread = 'main'
- execution_thread = 'worker'
If you want to execute certain tests with only one specific values of
execution_thread, you can use this class decorator. For example:
@with_execution_thread('main')
class TestOnlyMainThread:
...
@with_execution_thread('worker')
class TestOnlyWorker:
...
If you use @with_execution_thread(None), the logic to inject the
execution_thread config is disabled.
"""
if values == (None,):
@pytest.fixture
def execution_thread(self, request):
return None
else:
for value in values:
assert value in ("main", "worker")
@pytest.fixture(params=params_with_marks(values))
def execution_thread(self, request):
return request.param
def with_execution_thread_decorator(cls):
cls.execution_thread = execution_thread
return cls
return with_execution_thread_decorator
def skip_worker(reason):
"""
Decorator to skip a test if self.execution_thread == 'worker'
"""
if callable(reason):
# this happens if you use @skip_worker instead of @skip_worker("bla bla bla")
raise Exception(
"You need to specify a reason for skipping, "
"please use: @skip_worker('...')"
)
def decorator(fn):
@functools.wraps(fn)
def decorated(self, *args):
if self.execution_thread == "worker":
pytest.skip(reason)
return fn(self, *args)
return decorated
return decorator
def only_main(fn):
"""
Decorator to mark a test which make sense only in the main thread
"""
@functools.wraps(fn)
def decorated(self, *args):
if self.execution_thread == "worker":
return
return fn(self, *args)
return decorated
def filter_inner_text(text, exclude=None):
return "\n".join(filter_page_content(text.splitlines(), exclude=exclude))
def filter_page_content(lines, exclude=None):
"""Remove lines that are not relevant for the test. By default, ignores:
('', 'execution_thread = "main"', 'execution_thread = "worker"')
Args:
lines (list): list of strings
exclude (list): list of strings to exclude
Returns:
list: list of strings
"""
if exclude is None:
exclude = {"", 'execution_thread = "main"', 'execution_thread = "worker"'}
return [line for line in lines if line not in exclude]
@pytest.mark.usefixtures("init")
@with_execution_thread("main", "worker")
class PyScriptTest:
"""
Base class to write PyScript integration tests, based on playwright.
It provides a simple API to generate HTML files and load them in
playwright.
It also provides a Pythonic API on top of playwright for the most
common tasks; in particular:
- self.console collects all the JS console.* messages. Look at the doc
of ConsoleMessageCollection for more details.
- self.check_js_errors() checks that no JS errors have been thrown
- after each test, self.check_js_errors() is automatically run to ensure
that no JS error passes uncaught.
- self.wait_for_console waits until the specified message appears in the
console
- self.wait_for_pyscript waits until all the PyScript tags have been
evaluated
- self.pyscript_run is the main entry point for pyscript tests: it
creates an HTML page to run the specified snippet.
"""
DEFAULT_TIMEOUT = 20000
@pytest.fixture()
def init(self, request, tmpdir, logger, page, execution_thread):
"""
Fixture to automatically initialize all the tests in this class and its
subclasses.
The magic is done by the decorator @pytest.mark.usefixtures("init"),
which tells pytest to automatically use this fixture for all the test
method of this class.
Using the standard pytest behavior, we can request more fixtures:
tmpdir, and page; 'page' is a fixture provided by pytest-playwright.
Then, we save these fixtures on the self and proceed with more
initialization. The end result is that the requested fixtures are
automatically made available as self.xxx in all methods.
"""
self.testname = request.function.__name__.replace("test_", "")
self.tmpdir = tmpdir
# create a symlink to BUILD inside tmpdir
tmpdir.join("build").mksymlinkto(BUILD)
self.tmpdir.chdir()
self.tmpdir.join("favicon.ico").write("")
self.logger = logger
self.execution_thread = execution_thread
self.dev_server = None
if request.config.option.no_fake_server:
# use a real HTTP server. Note that as soon as we request the
# fixture, the server automatically starts in its own thread.
self.dev_server = request.getfixturevalue("dev_server")
self.http_server_addr = self.dev_server.base_url
self.router = None
else:
# use the internal playwright routing
self.http_server_addr = "https://fake_server"
self.router = SmartRouter(
"fake_server",
cache=request.config.cache,
logger=logger,
usepdb=request.config.option.usepdb,
)
self.router.install(page)
#
self.init_page(page)
#
# this extra print is useful when using pytest -s, else we start printing
# in the middle of the line
print()
#
# if you use pytest --headed you can see the browser page while
# playwright executes the tests, but the page is closed very quickly
# as soon as the test finishes. To avoid that, we automatically start
# a pdb so that we can wait as long as we want.
yield
if request.config.option.headed:
pdb.Pdb.intro = (
"\n"
"This (Pdb) was started automatically because you passed --headed:\n"
"the execution of the test pauses here to give you the time to inspect\n"
"the browser. When you are done, type one of the following commands:\n"
" (Pdb) continue\n"
" (Pdb) cont\n"
" (Pdb) c\n"
)
pdb.set_trace()
def init_page(self, page):
self.page = page
# set default timeout to 60000 millliseconds from 30000
page.set_default_timeout(self.DEFAULT_TIMEOUT)
self.console = ConsoleMessageCollection(self.logger)
self._js_errors = []
self._py_errors = []
page.on("console", self._on_console)
page.on("pageerror", self._on_pageerror)
@property
def headers(self):
if self.dev_server is None:
return self.router.headers
return self.dev_server.RequestHandlerClass.my_headers()
def disable_cors_headers(self):
if self.dev_server is None:
self.router.enable_cors_headers = False
else:
self.dev_server.RequestHandlerClass.enable_cors_headers = False
def run_js(self, code):
"""
allows top level await to be present in the `code` parameter
"""
self.page.evaluate(
"""(async () => {
try {%s}
catch(e) {
console.error(e);
}
})();"""
% code
)
def teardown_method(self):
# we call check_js_errors on teardown: this means that if there are still
# non-cleared errors, the test will fail. If you expect errors in your
# page and they should not cause the test to fail, you should call
# self.check_js_errors() in the test itself.
self.check_js_errors()
self.check_py_errors()
def _on_console(self, msg):
if msg.type == "error" and "Traceback (most recent call last)" in msg.text:
# this is a Python traceback, let's record it as a py_error
self._py_errors.append(msg.text)
self.console.add_message(msg.type, msg.text)
def _on_pageerror(self, error):
# apparently, playwright Error.stack contains all the info that we
# want: exception name, message and stacktrace. The docs say that
# error.stack is optional, so fallback to the standard repr if it's
# unavailable.
error_msg = error.stack or str(error)
self.console.add_message("js_error", error_msg)
self._js_errors.append(error_msg)
def _check_page_errors(self, kind, expected_messages):
"""
Check whether the page raised any 'JS' or 'Python' error.
expected_messages is a list of strings of errors that you expect they
were raised in the page. They are checked using a simple 'in' check,
equivalent to this:
if expected_message in actual_error_message:
...
If an error was expected but not found, it raises PageErrorsDidNotRaise.
If there are MORE errors other than the expected ones, it raises PageErrors.
Upon return, all the errors are cleared, so a subsequent call to
check_{js,py}_errors will not raise, unless NEW errors have been reported
in the meantime.
"""
assert kind in ("JS", "Python")
if kind == "JS":
actual_errors = self._js_errors[:]
else:
actual_errors = self._py_errors[:]
expected_messages = list(expected_messages)
for i, msg in enumerate(expected_messages):
for j, error in enumerate(actual_errors):
if msg is not None and error is not None and msg in error:
# we matched one expected message with an error, remove both
expected_messages[i] = None
actual_errors[j] = None
# if everything is find, now expected_messages and actual_errors contains
# only Nones. If they contain non-None elements, it means that we
# either have messages which are expected-but-not-found or
# found-but-not-expected.
not_found = [msg for msg in expected_messages if msg is not None]
unexpected = [err for err in actual_errors if err is not None]
if kind == "JS":
self.clear_js_errors()
else:
self.clear_py_errors()
if not_found:
# expected-but-not-found
raise PageErrorsDidNotRaise(kind, not_found, unexpected)
if unexpected:
# found-but-not-expected
raise PageErrors(kind, unexpected)
def check_js_errors(self, *expected_messages):
"""
Check whether JS errors were reported.
See the docstring for _check_page_errors for more details.
"""
self._check_page_errors("JS", expected_messages)
def check_py_errors(self, *expected_messages):
"""
Check whether Python errors were reported.
See the docstring for _check_page_errors for more details.
"""
self._check_page_errors("Python", expected_messages)
def clear_js_errors(self):
"""
Clear all JS errors.
"""
self._js_errors = []
def clear_py_errors(self):
self._py_errors = []
def writefile(self, filename, content):
"""
Very thin helper to write a file in the tmpdir
"""
f = self.tmpdir.join(filename)
f.dirpath().ensure(dir=True)
f.write(content)
def goto(self, path):
self.logger.reset()
self.logger.log("page.goto", path, color="yellow")
url = f"{self.http_server_addr}/{path}"
self.page.goto(url, timeout=0)
def wait_for_console(
self,
text,
*,
match_substring=False,
timeout=None,
check_js_errors=True,
):
"""
Wait until the given message appear in the console. If the message was
already printed in the console, return immediately.
By default "text" must be the *exact* string as printed by a single
call to e.g. console.log. If match_substring is True, it is enough
that the console contains the given text anywhere.
timeout is expressed in milliseconds. If it's None, it will use
the same default as playwright, which is 30 seconds.
If check_js_errors is True (the default), it also checks that no JS
errors were raised during the waiting.
Return the elapsed time in ms.
"""
if match_substring:
def find_text():
return text in self.console.all.text
else:
def find_text():
return text in self.console.all.lines
if timeout is None:
timeout = 10 * 1000
# NOTE: we cannot use playwright's own page.expect_console_message(),
# because if you call it AFTER the text has already been emitted, it
# waits forever. Instead, we have to use our own custom logic.
try:
t0 = time.time()
while True:
elapsed_ms = (time.time() - t0) * 1000
if elapsed_ms > timeout:
raise TimeoutError(f"{elapsed_ms:.2f} ms")
#
if find_text():
# found it!
return elapsed_ms
#
self.page.wait_for_timeout(50)
finally:
# raise JsError if there were any javascript exception. Note that
# this might happen also in case of a TimeoutError. In that case,
# the JsError will shadow the TimeoutError but this is correct,
# because it's very likely that the console message never appeared
# precisely because of the exception in JS.
if check_js_errors:
self.check_js_errors()
def wait_for_pyscript(self, *, timeout=None, check_js_errors=True):
"""
Wait until pyscript has been fully loaded.
Timeout is expressed in milliseconds. If it's None, it will use
playwright's own default value, which is 30 seconds).
If check_js_errors is True (the default), it also checks that no JS
errors were raised during the waiting.
"""
scripts = (
self.page.locator("script[type=py]").all()
+ self.page.locator("py-script").all()
)
n_scripts = len(scripts)
# this is printed by core.js:onAfterRun
elapsed_ms = self.wait_for_console(
"---py:all-done---",
timeout=timeout,
check_js_errors=check_js_errors,
)
self.logger.log(
"wait_for_pyscript", f"Waited for {elapsed_ms/1000:.2f} s", color="yellow"
)
# We still don't know why this wait is necessary, but without it
# events aren't being triggered in the tests.
self.page.wait_for_timeout(100)
SCRIPT_TAG_REGEX = re.compile('(
{extra_head}
{snippet}