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("pyscriptjs", "build")
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
@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.
"""
@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.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(60000)
self.console = ConsoleMessageCollection(self.logger)
self._js_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()
def _on_console(self, msg):
self.console.add_message(msg.type, msg.text)
def _on_pageerror(self, error):
self.console.add_message("js_error", error.stack)
self._js_errors.append(error)
def check_js_errors(self, *expected_messages):
"""
Check whether JS errors were reported.
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
DidNotRaiseJsError().
If there are MORE errors other than the expected ones, it raises JsErrors.
Upon return, all the errors are cleared, so a subsequent call to
check_js_errors will not raise, unless NEW JS errors have been reported
in the meantime.
"""
expected_messages = list(expected_messages)
js_errors = self._js_errors[:]
for i, msg in enumerate(expected_messages):
for j, error in enumerate(js_errors):
if msg is not None and error is not None and msg in error.message:
# we matched one expected message with an error, remove both
expected_messages[i] = None
js_errors[j] = None
# if everything is find, now expected_messages and js_errors contains
# only Nones. If they contain non-None elements, it means that we
# either have messages which are expected-but-not-found or errors
# which are found-but-not-expected.
expected_messages = [msg for msg in expected_messages if msg is not None]
js_errors = [err for err in js_errors if err is not None]
self.clear_js_errors()
if expected_messages:
# expected-but-not-found
raise JsErrorsDidNotRaise(expected_messages, js_errors)
if js_errors:
# found-but-not-expected
raise JsErrors(js_errors)
def clear_js_errors(self):
"""
Clear all JS errors.
"""
self._js_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 = 30 * 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.
"""
# this is printed by interpreter.ts:Interpreter.initialize
elapsed_ms = self.wait_for_console(
"[pyscript/main] PyScript page fully initialized",
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)
def _parse_py_config(self, doc):
configs = re.findall("(.*?)", doc, flags=re.DOTALL)
configs = [cfg.strip() for cfg in configs]
if len(configs) == 0:
return None
elif len(configs) == 1:
return toml.loads(configs[0])
else:
raise AssertionError("Too many ")
def _inject_execution_thread_config(self, snippet, execution_thread):
"""
If snippet already contains a py-config, let's try to inject
execution_thread automatically. Note that this works only for plain
with inline config: type="json" and src="..." are not
supported by this logic, which should remain simple.
"""
cfg = self._parse_py_config(snippet)
if cfg is None:
# we don't have any , let's add one
py_config_maybe = f"""
execution_thread = "{execution_thread}"
"""
else:
cfg["execution_thread"] = execution_thread
dumped_cfg = toml.dumps(cfg)
new_py_config = f"""
{dumped_cfg}
"""
snippet = re.sub(
".*", new_py_config, snippet, flags=re.DOTALL
)
# no need for extra config, it's already in the snippet
py_config_maybe = ""
#
return snippet, py_config_maybe
def _pyscript_format(self, snippet, *, execution_thread, extra_head=""):
if execution_thread is None:
py_config_maybe = ""
else:
snippet, py_config_maybe = self._inject_execution_thread_config(
snippet, execution_thread
)
doc = f"""
{extra_head}
{py_config_maybe}
{snippet}
"""
return doc
def pyscript_run(
self, snippet, *, extra_head="", wait_for_pyscript=True, timeout=None
):
"""
Main entry point for pyscript tests.
snippet contains a fragment of HTML which will be put inside a full
HTML document. In particular, the automatically contains the
correct