erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests: reformat tests using black.

Browse source
This commit is contained in:
Goffi 2024-09-26 16:11:56 +02:00
commit 5005ca346b
6 changed files with 161 additions and 140 deletions
Download patch file Download diff file Expand all files Collapse all files

41
tests/e2e/conftest.py
View file

@ -60,7 +60,7 @@ class FakeFile:
The file should be deleted after use. If not, it will be deleted at the end of
session with the whole temporary test files directory.
"""
name = ''.join(random.choices(self.ALPHABET, k=8))
name = "".join(random.choices(self.ALPHABET, k=8))
return self.dest_files / name
def size(self, size: int, use_cache: bool = True):
@ -72,7 +72,7 @@ class FakeFile:
if not use_cache or not dest_path.exists():
hash_ = hashlib.sha256()
remaining = size
with dest_path.open('wb') as f:
with dest_path.open("wb") as f:
while remaining:
if remaining > self.BUF_SIZE:
to_get = self.BUF_SIZE
@ -92,7 +92,7 @@ class FakeFile:
def get_dest_hash(self, dest_file: Path) -> str:
"""Calculate hash of file at given path"""
hash_ = hashlib.sha256()
with dest_file.open('rb') as f:
with dest_file.open("rb") as f:
while True:
buf = f.read(self.BUF_SIZE)
if not buf:
@ -105,15 +105,15 @@ class TestMessage(EmailMessage):
@property
def subject(self):
return self['subject']
return self["subject"]
@property
def from_(self):
return self['from']
return self["from"]
@property
def to(self):
return self['to']
return self["to"]
@property
def body(self):
@ -150,10 +150,10 @@ def test_profiles():
2: 1,
3: 1,
}
for server_idx in range(1, nb_servers+1):
for server_idx in range(1, nb_servers + 1):
account_stop = accounts_by_servers[server_idx] + 1
for account_idx in range(1, account_stop):
profile_suff = f"_s{server_idx}" if server_idx>1 else ""
profile_suff = f"_s{server_idx}" if server_idx > 1 else ""
profile = f"account{account_idx}{profile_suff}"
profiles.append(profile)
try:
@ -161,7 +161,7 @@ def test_profiles():
f"account{account_idx}@server{server_idx}.test",
"test",
profile=profile,
host=f"server{server_idx}.test"
host=f"server{server_idx}.test",
)
except sh.ErrorReturnCode_19:
# this is the conlict exit code, this can happen when tests are run
@ -186,25 +186,20 @@ def pubsub_nodes(test_profiles):
One node will be on account1's PEP, the other one on pubsub.server1.test.
"""
li.pubsub.node.create(
"-f", "access_model", "open",
node="test",
profile="account1", connect=True
"-f", "access_model", "open", node="test", profile="account1", connect=True
)
li.pubsub.node.create(
"-f", "access_model", "open",
service="pubsub.server1.test", node="test",
profile="account1"
"-f",
"access_model",
"open",
service="pubsub.server1.test",
node="test",
profile="account1",
)
yield
li.pubsub.node.delete(node="test", profile="account1", connect=True, force=True)
li.pubsub.node.delete(
node="test",
profile="account1", connect=True,
force=True
)
li.pubsub.node.delete(
service="pubsub.server1.test", node="test",
profile="account1",
force=True
service="pubsub.server1.test", node="test", profile="account1", force=True
)

18
tests/e2e/libervia-cli/conftest.py
View file

@ -35,13 +35,13 @@ class LiberviaCliJson:
def __call__(self, *args, **kwargs):
args = self.subcommands + list(args)
self.subcommands.clear()
kwargs['output'] = 'json-raw'
kwargs['_tty_out'] = False
kwargs["output"] = "json-raw"
kwargs["_tty_out"] = False
cmd = li(*args, **kwargs)
return json.loads(cmd)
def __getattr__(self, name):
if name.startswith('_'):
if name.startswith("_"):
# no li subcommand starts with a "_",
# and pytest uses some attributes with this name scheme
return super().__getattr__(name)
@ -55,13 +55,14 @@ class LiberviaCliElt(LiberviaCliJson):
def __init__(self):
super().__init__()
from libervia.backend.tools.xml_tools import ElementParser
self.parser = ElementParser()
def __call__(self, *args, **kwargs):
args = self.subcommands + list(args)
self.subcommands.clear()
kwargs['output'] = 'xml-raw'
kwargs['_tty_out'] = False
kwargs["output"] = "xml-raw"
kwargs["_tty_out"] = False
cmd = li(*args, **kwargs)
return self.parser(cmd.strip())
@ -79,7 +80,8 @@ class Editor:
self.editor_path = self.tmp_dir_path / "editor.py"
self.ori_content_path = self.tmp_dir_path / "original_content"
self.new_content_path = self.tmp_dir_path / "new_content"
self.base_script = dedent(f"""\
self.base_script = dedent(
f"""\
#!{sys.executable}
import sys
@ -111,9 +113,9 @@ class Editor:
If filter_ is not specified, original content is written unmodified.
Code must be on a single line.
"""
if '\n' in content_filter:
if "\n" in content_filter:
raise ValueError("new lines can't be used in filter_")
with self.editor_path.open('w') as f:
with self.editor_path.open("w") as f:
f.write(self.base_script.format(content_filter=content_filter))
self.editor_path.chmod(0o700)
self.editor_set = True

157
tests/e2e/libervia-cli/test_libervia-cli.py
View file

@ -34,7 +34,7 @@ from libervia.backend.tools.common import uri
if os.getenv("LIBERVIA_TEST_ENV_E2E") is None:
pytest.skip(
"skipping end-to-end tests, we are not in a test environment",
allow_module_level=True
allow_module_level=True,
)
@ -52,10 +52,7 @@ class TestLiberviaCliAccount:
def test_create_and_delete(self, li_json):
"""Create an account in-band, connect it, then delete it and its profile"""
li.account.create(
"test_create@server1.test",
"test",
profile="test_create",
host="server1.test"
"test_create@server1.test", "test", profile="test_create", host="server1.test"
)
profiles = li_json.profile.list()
assert "test_create" in profiles
@ -96,15 +93,15 @@ class TestLiberviaCliPubsub:
item3_id = li.pubsub.set(node="test", quiet=True, _in=payload)
parsed_elt = li_elt.pubsub.get(node="test", item=item1_id)
payload = parsed_elt.firstChildElement()
assert payload.name == 'test'
assert payload.name == "test"
assert str(payload) == content
parsed_elt = li_elt.pubsub.get(node="test", item=item2_id)
payload = parsed_elt.firstChildElement()
assert payload.name == 'test'
assert payload.name == "test"
assert str(payload) == content
parsed_elt = li_elt.pubsub.get(node="test", item=item3_id)
payload = parsed_elt.firstChildElement()
assert payload.name == 'test'
assert payload.name == "test"
assert str(payload) == content
# deleting first item should work
@ -131,7 +128,7 @@ class TestLiberviaCliPubsub:
parsed_elt = li_elt.pubsub.get(node="test", item=item_id)
edited_payload = parsed_elt.firstChildElement()
expected_edited_content = content.replace("original", "edited")
assert edited_payload.name == 'test'
assert edited_payload.name == "test"
assert str(edited_payload) == expected_edited_content
def test_affiliations(self, li_json):
@ -139,9 +136,7 @@ class TestLiberviaCliPubsub:
assert affiliations["test"] == "owner"
def test_uri(self):
built_uri = li.pubsub.uri(
service="pubsub.example.net", node="some_node"
).strip()
built_uri = li.pubsub.uri(service="pubsub.example.net", node="some_node").strip()
assert built_uri == "xmpp:pubsub.example.net?;node=some_node"
built_uri = li.pubsub.uri(
service="pubsub.example.net", node="some_node", item="some_item"
@ -154,26 +149,17 @@ class TestLiberviaCliPubsub:
fr_txt = "this is a blog post about France"
nc_txt = "this is a blog post about New Caledonia"
au_txt = "this is a blog post about Australia"
li.blog.set("-t", "travel", "-t", "europe", _in=sk_txt, syntax="markdown")
li.blog.set("-t", "travel", "-t", "europe", _in=fr_txt, syntax="markdown")
li.blog.set("-t", "travel", "-t", "south pacific", _in=nc_txt, syntax="markdown")
li.blog.set(
"-t", "travel", "-t", "europe",
_in=sk_txt,
syntax="markdown"
)
li.blog.set(
"-t", "travel", "-t", "europe",
_in=fr_txt,
syntax="markdown"
)
li.blog.set(
"-t", "travel", "-t", "south pacific",
_in=nc_txt,
syntax="markdown"
)
li.blog.set(
"-t", "travel", "-t", "south pacific",
"-t",
"travel",
"-t",
"south pacific",
_in="this is a blog post about Australia",
title=au_txt,
syntax="markdown"
syntax="markdown",
)
# we get the blog to activate the cache for it
li.blog.get(max_items=1)
@ -215,25 +201,28 @@ class TestLiberviaCliBlog:
item_data = li_json.blog.get(max=1, before="")
item = item_data[0][0]
metadata = item_data[1]
assert metadata['service'] == "account1@server1.test"
assert metadata['node'] == self.MICROBLOG_NS
assert metadata['rsm'].keys() <= {"first", "last", "index", "count"}
item_id = item['id']
assert metadata["service"] == "account1@server1.test"
assert metadata["node"] == self.MICROBLOG_NS
assert metadata["rsm"].keys() <= {"first", "last", "index", "count"}
item_id = item["id"]
expected_uri = uri.build_xmpp_uri(
'pubsub', subtype="microblog", path="account1@server1.test",
node=self.MICROBLOG_NS, item=item_id
"pubsub",
subtype="microblog",
path="account1@server1.test",
node=self.MICROBLOG_NS,
item=item_id,
)
assert item['uri'] == expected_uri
assert item['content_xhtml'] == (
'<div><p>markdown <strong>bold</strong> '
assert item["uri"] == expected_uri
assert item["content_xhtml"] == (
"<div><p>markdown <strong>bold</strong> "
'<a href="https://example.net">link</a></p></div>'
)
assert isinstance(item['published'], int)
assert isinstance(item['updated'], int)
assert isinstance(item['comments'], list)
assert isinstance(item['tags'], list)
assert item['author'] == 'account1'
assert item['author_jid'] == 'account1@server1.test'
assert isinstance(item["published"], int)
assert isinstance(item["updated"], int)
assert isinstance(item["comments"], list)
assert isinstance(item["tags"], list)
assert item["author"] == "account1"
assert item["author_jid"] == "account1@server1.test"
def test_edit(self, editor, li_json):
payload_md = "content in **markdown**"
@ -243,9 +232,9 @@ class TestLiberviaCliBlog:
assert editor.new_content == payload_md
items_data = li_json.blog.get(max_items=1)
last_item = items_data[0][0]
last_item_id = last_item['id']
assert last_item['content'] == "content in markdown"
assert last_item['content_xhtml'] == (
last_item_id = last_item["id"]
assert last_item["content"] == "content in markdown"
assert last_item["content_xhtml"] == (
"<div><p>content in <strong>markdown</strong></p></div>"
)
editor.set_filter('f"{content} extended"')
@ -255,9 +244,9 @@ class TestLiberviaCliBlog:
items_data = li_json.blog.get(max_items=1)
last_item = items_data[0][0]
# we check that the id hasn't been modified
assert last_item['id'] == last_item_id
assert last_item['content'] == "content in markdown extended"
assert last_item['content_xhtml'] == (
assert last_item["id"] == last_item_id
assert last_item["content"] == "content in markdown extended"
assert last_item["content_xhtml"] == (
"<div><p>content in <strong>markdown</strong> extended</p></div>"
)
@ -286,7 +275,8 @@ class TestLiberviaCliFile:
dest_path.mkdir()
try:
li.file.receive(
"account1@server1.test", profile="account1_s2", path=dest_path)
"account1@server1.test", profile="account1_s2", path=dest_path
)
dest_file = dest_path / source_file.name
dest_file_hash = fake_file.get_dest_hash(dest_file)
finally:
@ -305,7 +295,8 @@ class TestLiberviaCliFile:
dest_path.mkdir()
try:
li.file.receive(
"account1@server1.test", profile="account1_s2", path=dest_path)
"account1@server1.test", profile="account1_s2", path=dest_path
)
dest_file = dest_path / source_file.name
dest_file_hash = fake_file.get_dest_hash(dest_file)
finally:
@ -315,7 +306,6 @@ class TestLiberviaCliFile:
assert source_file_hash == dest_file_hash
class TestE2EEncryption:
def test_pubsub_encryption_oxps(self, li_elt):
@ -324,9 +314,7 @@ class TestE2EEncryption:
li.blog.set(_in=secret_blog, node="e2ee_blog", item="test_e2ee", encrypt=True)
# the item should be transparently decrypted
parsed_decrypted = li_elt.pubsub.get(
node=node, item="test_e2ee", no_cache=True
)
parsed_decrypted = li_elt.pubsub.get(node=node, item="test_e2ee", no_cache=True)
entry_elt = parsed_decrypted.firstChildElement()
assert entry_elt.name == "entry"
assert entry_elt.uri == NS_ATOM
@ -344,14 +332,17 @@ class TestE2EEncryption:
def test_pubsub_secrets_sharing_oxps(self, li_elt):
secret_blog = "this is a secret blog post"
node="secret_sharing"
node = "secret_sharing"
li.blog.set(_in=secret_blog, node=node, item="test_e2ee", encrypt=True)
# the item must not be decrypted for account1_s2 (secret is not known)
parsed_item = li_elt.pubsub.get(
service="account1@server1.test", node=node, item="test_e2ee", no_cache=True,
profile="account1_s2"
service="account1@server1.test",
node=node,
item="test_e2ee",
no_cache=True,
profile="account1_s2",
)
encrypted_elt = parsed_item.firstChildElement()
assert encrypted_elt.name == "encrypted"
@ -360,12 +351,17 @@ class TestE2EEncryption:
assert secret_blog not in parsed_item.toXml()
# we share the secrets
li.pubsub.secret.share("account1@server2.test", service="account1@server1.test", node=node)
li.pubsub.secret.share(
"account1@server2.test", service="account1@server1.test", node=node
)
# and get the item again
parsed_item = li_elt.pubsub.get(
service="account1@server1.test", node=node, item="test_e2ee", no_cache=True,
profile="account1_s2"
service="account1@server1.test",
node=node,
item="test_e2ee",
no_cache=True,
profile="account1_s2",
)
# now it should be decrypted
entry_elt = parsed_item.firstChildElement()
@ -376,9 +372,9 @@ class TestE2EEncryption:
def test_pubsub_signature(self, li_json):
"""A pubsub item can be signed, and the signature can be verified"""
body = "this message is signed"
service="account1@server1.test"
node ="blog_signing"
item="signed_item"
service = "account1@server1.test"
node = "blog_signing"
item = "signed_item"
li.blog.set(_in=body, service=service, node=node, item=item, sign=True)
attachments = li_json.pubsub.attachments.get(
service=service, node=node, item=item
@ -388,7 +384,10 @@ class TestE2EEncryption:
assert attachment["from"] == "account1@server1.test"
signature_json = attachment["signature"]
sign_data = li_json.pubsub.signature.check(
json.dumps(signature_json), service=service, node=node, item=item,
json.dumps(signature_json),
service=service,
node=node,
item=item,
)
assert sign_data["signer"] == "account1@server1.test"
assert sign_data["validated"] == True
@ -402,13 +401,18 @@ class TestE2EEncryption:
source_file = fake_file.size(10240)
source_file_hash = fake_file.get_source_hash(source_file)
send_cmd = li.file.send(
source_file, "account1@server2.test", encrypt=True, _bg=True,
source_file,
"account1@server2.test",
encrypt=True,
_bg=True,
)
dest_path = fake_file.dest_files / "test_send_receive"
dest_path.mkdir()
try:
li.file.receive(
"account1@server1.test", profile="account1_s2", path=dest_path,
"account1@server1.test",
profile="account1_s2",
path=dest_path,
)
dest_file = dest_path / source_file.name
dest_file_hash = fake_file.get_dest_hash(dest_file)
@ -425,18 +429,21 @@ class TestE2EEncryption:
node = "e2ee_blog"
item = "test_pte"
li.encryption.start("account1@server2.test", name="omemo")
li.encryption.start(
"account1@server1.test", name="omemo", profile="account1_s2"
)
li.encryption.start("account1@server1.test", name="omemo", profile="account1_s2")
li.blog.set(
_in=secret_blog, node="e2ee_blog", item=item,
encrypt_for="account1@server2.test"
_in=secret_blog,
node="e2ee_blog",
item=item,
encrypt_for="account1@server2.test",
)
# the item should be transparently decrypted
parsed_decrypted = li_elt.pubsub.get(
service="account1@server1.test", node=node, item=item, no_cache=True,
profile="account1_s2"
service="account1@server1.test",
node=node,
item=item,
no_cache=True,
profile="account1_s2",
)
entry_elt = parsed_decrypted.firstChildElement()
assert entry_elt.name == "entry"

8
tests/e2e/libervia-web/conftest.py
View file

@ -49,10 +49,10 @@ def screenshot_on_failure(request):
if request.node.rep_setup.passed:
if request.node.rep_call.failed:
report_dir = Path(os.getenv("LIBERVIA_TEST_REPORT_DIR", "/tmp/tests_report"))
dest_dir = report_dir/"screenshots"
dest_dir = report_dir / "screenshots"
dest_dir.mkdir(parents=True, exist_ok=True)
filename = f"{datetime.now().isoformat()}_{request.node.name}.png"
dest_path = dest_dir/filename
dest_path = dest_dir / filename
helium.get_driver().save_screenshot(str(dest_path))
print(f" 📸screenshot saved to {dest_path}")
@ -92,6 +92,7 @@ def browser():
def nobody_logged_in(browser):
browser.get_driver().delete_all_cookies()
def log_in(browser, account):
try:
account_cookies = accounts_cookies[account]
@ -105,14 +106,17 @@ def log_in(browser, account):
else:
browser.get_driver().add_cookie(account_cookies)
@pytest.fixture
def log_in_account1(browser):
log_in(browser, "account1")
@pytest.fixture
def log_in_account1_s2(browser):
log_in(browser, "account1_s2")
@pytest.fixture
def mobile_screen(browser):
browser.get_driver().set_window_size(*SIZE_MOBILE)

14
tests/e2e/libervia-web/test_libervia-web.py
View file

@ -44,7 +44,7 @@ from selenium.webdriver.common.by import By
if os.getenv("LIBERVIA_TEST_ENV_E2E_WEB") is None:
pytest.skip(
"skipping end-to-end tests, we are not in a test environment for Libervia",
allow_module_level=True
allow_module_level=True,
)
pytestmark = pytest.mark.usefixtures("test_profiles", "screenshot_on_failure")
@ -111,9 +111,9 @@ class TestPhotos:
wait_until(lambda: S("#file_drop").exists())
wait_until(lambda: not S("#loading_screen").exists())
drag_file("/src/libervia-backend/tests/_files/test_1.jpg", "drop photos here")
wait_until(lambda: len(find_all(S("div.progress_finished")))==1)
wait_until(lambda: len(find_all(S("div.progress_finished"))) == 1)
drag_file("/src/libervia-backend/tests/_files/test_2.jpg", "drop photos here")
wait_until(lambda: len(find_all(S("div.progress_finished")))==2)
wait_until(lambda: len(find_all(S("div.progress_finished"))) == 2)
assert S('img[alt="test_1.jpg"]').exists()
assert S('img[alt="test_2.jpg"]').exists()
@ -133,8 +133,7 @@ class TestPhotos:
active_slide_1_elt = active_slide_1.web_element
click(S(".swiper-button-next"))
wait_until(
lambda:
"swiper-slide-active" not in active_slide_1_elt.get_attribute("class")
lambda: "swiper-slide-active" not in active_slide_1_elt.get_attribute("class")
)
active_slide_2 = S("div.swiper-slide-active")
assert active_slide_2.exists()
@ -149,7 +148,9 @@ class TestPhotos:
assert Text("Unauthorized").exists()
assert "Error" in get_driver().title
@pytest.mark.dependency(name="invite_ext_user", depends=["create_album", "ext_user_no_access"])
@pytest.mark.dependency(
name="invite_ext_user", depends=["create_album", "ext_user_no_access"]
)
def test_invitation_of_external_user(self, log_in_account1):
"""User can invite somebody not in its roster by its full JID"""
go_to(self.TEST_ALBUM_URL)
@ -212,7 +213,6 @@ class TestLists:
click("create list")
wait_until(Text("Success").exists)
@pytest.mark.dependency(
name="create_generic_list_item", depends=["create_generic_list"]
)

63
tests/e2e/run_e2e.py
View file

@ -30,6 +30,7 @@ from libervia import web
from libervia.backend.core import exceptions
from libervia.backend.tools.common import regex
import yaml
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
@ -43,7 +44,8 @@ OPT_KEEP_BROWSER = "--keep-browser"
OPT_VISUAL = "--visual"
OPT_DEV_MODE = "--dev-mode"
dev_mode_inst = dedent("""\
dev_mode_inst = dedent(
"""\
Here is a short script to start working with a logged account:
from helium import *
@ -52,7 +54,8 @@ dev_mode_inst = dedent("""\
write("account1", "login")
write("test", "password")
click("log in")
""")
"""
)
report_buffer = io.StringIO()
live_out_buf = []
live_err_buf = []
@ -63,7 +66,7 @@ def live_out(data):
# we may get bytes when buffer is reached and we are in the middle of an unicode
# sequence. In this case we buffer it, and print it when it's complete
if isinstance(data, str):
data = b''.join(live_out_buf).decode() + data
data = b"".join(live_out_buf).decode() + data
live_out_buf.clear()
else:
live_out_buf.append(data)
@ -80,7 +83,7 @@ def live_out(data):
def live_err(data):
if live_err_buf:
if isinstance(data, str):
data = b''.join(live_err_buf).decode() + data
data = b"".join(live_err_buf).decode() + data
live_err_buf.clear()
else:
live_err_buf.append(data)
@ -111,22 +114,24 @@ def set_env(override, name, value="1"):
environment = override["services"]["backend"].setdefault("environment", {})
environment[name] = value
def write_report_log(path, log_raw, with_ansi=False):
log_raw = str(log_raw)
if with_ansi:
# we save 2 versions: one with ANSI escape codes
report_ansi = path.with_suffix(".ansi")
with report_ansi.open('w') as f:
with report_ansi.open("w") as f:
f.write(log_raw)
# and one cleaner, without them
report_log = path.with_suffix(".log")
with report_log.open('w') as f:
f.write(regex.RE_ANSI_REMOVE.sub('', log_raw))
with report_log.open("w") as f:
f.write(regex.RE_ANSI_REMOVE.sub("", log_raw))
else:
report_log = path.with_suffix(".log")
with report_log.open('w') as f:
with report_log.open("w") as f:
f.write(log_raw)
def use_e2e_env():
rev = os.getenv("DOCKER_LIBERVIA_REV", "dev")
print(f"Running tests for Libervia {rev}")
@ -181,7 +186,8 @@ def use_e2e_env():
with tempfile.TemporaryDirectory(prefix="libervia_test_e2e_") as temp_dir:
override_path = Path(temp_dir) / "test_override.yml"
override = yaml.load(
dedent(f"""\
dedent(
f"""\
version: "3.6"
services:
backend:
@ -205,8 +211,8 @@ def use_e2e_env():
target: {libervia_templates_target}
read_only: true
"""
),
Loader=Loader
),
Loader=Loader,
)
if keep_profiles:
@ -222,24 +228,35 @@ def use_e2e_env():
yaml.dump(override, f, Dumper=Dumper)
docker_compose = sh.docker_compose.bake(
"-f", compose_e2e_path, "-f", override_path)
"-f", compose_e2e_path, "-f", override_path
)
docker_compose.up("-d")
p = docker_compose.exec(
"--workdir", "/src/libervia-backend/tests", "backend",
"pytest", "-o", "cache_dir=/tmp", *sys.argv[1:], color="yes",
_in=sys.stdin, _out=live_out, _out_bufsize=0, _err=live_err, _err_bufsize=0,
_bg=True
"--workdir",
"/src/libervia-backend/tests",
"backend",
"pytest",
"-o",
"cache_dir=/tmp",
*sys.argv[1:],
color="yes",
_in=sys.stdin,
_out=live_out,
_out_bufsize=0,
_err=live_err,
_err_bufsize=0,
_bg=True,
)
if visual:
vnc_port = docker_compose.port("backend", "5900").split(':', 1)[1].strip()
vnc_port = docker_compose.port("backend", "5900").split(":", 1)[1].strip()
p_vnc = sh.vncviewer(
f"localhost:{vnc_port}",
_bg=True,
# vncviewer exits with 1 when we send an SIGTERM to it, and it's printed
# before we can catch it (it's happening in a thread). Thus we set exit
# code 1 as OK to avoid the backtrace.
_ok_code=[0, 1]
_ok_code=[0, 1],
)
else:
p_vnc = None
@ -254,17 +271,13 @@ def use_e2e_env():
# with `isoformat()`.
sh.docker.cp(f"{libervia_cont_id}:/reports", f"./{report_dest}")
write_report_log(
report_dest/"report",
report_buffer.getvalue(),
with_ansi=True
report_dest / "report", report_buffer.getvalue(), with_ansi=True
)
write_report_log(
report_dest/"backend",
docker_compose.logs("--no-log-prefix", "backend")
report_dest / "backend", docker_compose.logs("--no-log-prefix", "backend")
)
write_report_log(
report_dest/"web",
docker_compose.logs("--no-log-prefix", "web")
report_dest / "web", docker_compose.logs("--no-log-prefix", "web")
)
print(f"report saved to {report_dest}")
sys.exit(e.exit_code)