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, docker(e2e): added e2e tests for Libervia:

Browse source 
- moved jp tests to `e2e/jp`
- new fixtures
- adapted docker-compose
- improved `run_e2e` with several flags + report on failure
- doc to come
This commit is contained in:
Goffi 2020-11-27 16:39:40 +01:00
commit fb216829c3
8 changed files with 601 additions and 159 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docker/backend_e2e/sat.conf
View file

@ -1,5 +1,8 @@
[DEFAULT]
xmpp_domain = server1.test
email_server = sat.test
email_port = 8025
email_admins_list = admin@server1.test
[component file_sharing]
tls_certificate = /usr/share/sat/certificates/server1.test.pem

8
docker/docker-compose_e2e.yml
View file

@ -42,6 +42,10 @@ services:
image: salutatoi/sat_e2e
environment:
SAT_TEST_ENV_E2E: "1"
SAT_TEST_ENV_E2E_LIBERVIA: "1"
ports:
# VNC server for Libervia e2e tests visual mode
- 5900
networks:
default:
aliases:
@ -57,3 +61,7 @@ services:
ports:
- "8080"
- "8443"
networks:
default:
aliases:
- libervia.test

205
tests/e2e/conftest.py
View file

@ -16,142 +16,17 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import sys
import os
import tempfile
import string
import hashlib
import random
from pathlib import Path
from textwrap import dedent
import json
import pytest
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Message
from email.message import EmailMessage
from sh import jp
class JpJson:
"""jp like commands parsing result as JSON"""
def __init__(self):
self.subcommands = []
def __call__(self, *args, **kwargs):
args = self.subcommands + list(args)
self.subcommands.clear()
kwargs['output'] = 'json_raw'
kwargs['_tty_out'] = False
cmd = jp(*args, **kwargs)
return json.loads(cmd.stdout)
def __getattr__(self, name):
if name.startswith('_'):
# no jp subcommand starts with a "_",
# and pytest uses some attributes with this name scheme
return super().__getattr__(name)
self.subcommands.append(name)
return self
class JpElt(JpJson):
"""jp like commands parsing result as domishElement"""
def __init__(self):
super().__init__()
from sat.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
cmd = jp(*args, **kwargs)
return self.parser(cmd.stdout.decode().strip())
class Editor:
def __init__(self):
# temporary directory will be deleted Automatically when this object will be
# destroyed
self.tmp_dir_obj = tempfile.TemporaryDirectory(prefix="sat_e2e_test_editor_")
self.tmp_dir_path = Path(self.tmp_dir_obj.name)
if not sys.executable:
raise Exception("Can't find python executable")
self.editor_set = False
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"""\
#!{sys.executable}
import sys
def content_filter(content):
return {{content_filter}}
with open(sys.argv[1], 'r+') as f:
original_content = f.read()
f.seek(0)
new_content = content_filter(original_content)
f.write(new_content)
f.truncate()
with open("{self.ori_content_path}", "w") as f:
f.write(original_content)
with open("{self.new_content_path}", "w") as f:
f.write(new_content)
"""
)
self._env = os.environ.copy()
self._env["EDITOR"] = str(self.editor_path)
def set_filter(self, content_filter: str = "content"):
"""Python code to modify original content
The code will be applied to content received by editor.
The original content received by editor is in the "content" variable.
If filter_ is not specified, original content is written unmodified.
Code must be on a single line.
"""
if '\n' in content_filter:
raise ValueError("new lines can't be used in filter_")
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
@property
def env(self):
"""Get environment variable with the editor set"""
if not self.editor_set:
self.set_filter()
return self._env
@property
def original_content(self):
"""Last content received by editor, before any modification
returns None if editor has not yet been called
"""
try:
with self.ori_content_path.open() as f:
return f.read()
except FileNotFoundError:
return None
@property
def new_content(self):
"""Last content writen by editor
This is the final content, after filter has been applied to original content
returns None if editor has not yet been called
"""
try:
with self.new_content_path.open() as f:
return f.read()
except FileNotFoundError:
return None
import pytest
class FakeFile:
@ -225,16 +100,33 @@ class FakeFile:
return hash_.hexdigest()
@pytest.fixture(scope="session")
def jp_json():
"""Run jp with "json_raw" output, and returns the parsed value"""
return JpJson()
class TestMessage(EmailMessage):
@property
def subject(self):
return self['subject']
@property
def from_(self):
return self['from']
@property
def to(self):
return self['to']
@property
def body(self):
return self.get_payload(decode=True).decode()
@pytest.fixture(scope="session")
def jp_elt():
"""Run jp with "xml_raw" output, and returns the parsed value"""
return JpElt()
class SMTPMessageHandler(Message):
messages = []
def __init__(self):
super().__init__(message_class=TestMessage)
def handle_message(self, message):
self.messages.append(message)
@pytest.fixture(scope="session")
@ -272,16 +164,18 @@ def test_profiles():
jp.profile.modify(profile="account1", default=True, connect=True)
jp.profile.connect(profile="account1_s2", connect=True)
yield tuple(profiles)
for profile in profiles:
jp.account.delete(profile=profile, connect=True, force=True)
jp.profile.delete(profile, force=True)
# This environment may be used during tests development
if os.getenv("SAT_TEST_E2E_KEEP_PROFILES") == None:
for profile in profiles:
jp.account.delete(profile=profile, connect=True, force=True)
jp.profile.delete(profile, force=True)
@pytest.fixture(scope="class")
def pubsub_nodes(test_profiles):
"""Create 2 testing nodes
Both nodes will be created with "account1" profile, named "test" and have and "open"
Both nodes will be created with "account1" profile, named "test" and have an "open"
access model.
One node will account1's PEP, the other one on pubsub.server1.test.
"""
@ -308,13 +202,30 @@ def pubsub_nodes(test_profiles):
)
@pytest.fixture()
def editor():
"""Create a fake editor to automatise edition from CLI"""
return Editor()
@pytest.fixture(scope="session")
def fake_file():
"""Manage dummy files creation and destination path"""
return FakeFile()
@pytest.fixture(scope="session")
def test_files():
"""Return a Path to test files directory"""
return Path(__file__).parent.parent / "_files"
@pytest.fixture(scope="session")
def fake_smtp():
"""Create a fake STMP server to check sent emails"""
controller = Controller(SMTPMessageHandler())
controller.hostname = "0.0.0.0"
controller.start()
yield
controller.stop()
@pytest.fixture
def sent_emails(fake_smtp):
"""Catch email sent during the tests"""
SMTPMessageHandler.messages.clear()
return SMTPMessageHandler.messages

169
tests/e2e/jp/conftest.py Normal file
View file

@ -0,0 +1,169 @@
#!/usr/bin/env python3
# SàT: an XMPP client
# Copyright (C) 2009-2020 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import sys
import os
import tempfile
from pathlib import Path
from textwrap import dedent
import json
import pytest
from sh import jp
class JpJson:
"""jp like commands parsing result as JSON"""
def __init__(self):
self.subcommands = []
def __call__(self, *args, **kwargs):
args = self.subcommands + list(args)
self.subcommands.clear()
kwargs['output'] = 'json_raw'
kwargs['_tty_out'] = False
cmd = jp(*args, **kwargs)
return json.loads(cmd.stdout)
def __getattr__(self, name):
if name.startswith('_'):
# no jp subcommand starts with a "_",
# and pytest uses some attributes with this name scheme
return super().__getattr__(name)
self.subcommands.append(name)
return self
class JpElt(JpJson):
"""jp like commands parsing result as domishElement"""
def __init__(self):
super().__init__()
from sat.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
cmd = jp(*args, **kwargs)
return self.parser(cmd.stdout.decode().strip())
class Editor:
def __init__(self):
# temporary directory will be deleted Automatically when this object will be
# destroyed
self.tmp_dir_obj = tempfile.TemporaryDirectory(prefix="sat_e2e_test_editor_")
self.tmp_dir_path = Path(self.tmp_dir_obj.name)
if not sys.executable:
raise Exception("Can't find python executable")
self.editor_set = False
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"""\
#!{sys.executable}
import sys
def content_filter(content):
return {{content_filter}}
with open(sys.argv[1], 'r+') as f:
original_content = f.read()
f.seek(0)
new_content = content_filter(original_content)
f.write(new_content)
f.truncate()
with open("{self.ori_content_path}", "w") as f:
f.write(original_content)
with open("{self.new_content_path}", "w") as f:
f.write(new_content)
"""
)
self._env = os.environ.copy()
self._env["EDITOR"] = str(self.editor_path)
def set_filter(self, content_filter: str = "content"):
"""Python code to modify original content
The code will be applied to content received by editor.
The original content received by editor is in the "content" variable.
If filter_ is not specified, original content is written unmodified.
Code must be on a single line.
"""
if '\n' in content_filter:
raise ValueError("new lines can't be used in filter_")
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
@property
def env(self):
"""Get environment variable with the editor set"""
if not self.editor_set:
self.set_filter()
return self._env
@property
def original_content(self):
"""Last content received by editor, before any modification
returns None if editor has not yet been called
"""
try:
with self.ori_content_path.open() as f:
return f.read()
except FileNotFoundError:
return None
@property
def new_content(self):
"""Last content writen by editor
This is the final content, after filter has been applied to original content
returns None if editor has not yet been called
"""
try:
with self.new_content_path.open() as f:
return f.read()
except FileNotFoundError:
return None
@pytest.fixture(scope="session")
def jp_json():
"""Run jp with "json_raw" output, and returns the parsed value"""
return JpJson()
@pytest.fixture(scope="session")
def jp_elt():
"""Run jp with "xml_raw" output, and returns the parsed value"""
return JpElt()
@pytest.fixture()
def editor():
"""Create a fake editor to automatise edition from CLI"""
return Editor()

0
tests/e2e/test_jp.py → tests/e2e/jp/test_jp.py
View file

117
tests/e2e/libervia/conftest.py Normal file
View file

@ -0,0 +1,117 @@
#!/usr/bin/env python3
# SàT: an XMPP client
# Copyright (C) 2009-2020 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import sys
import os
import socket
import pytest
import time
from datetime import datetime
from pathlib import Path
import helium
LIBERVIA_HOST = "libervia.test"
LIBERVIA_PORT_HTTPS = 8443
BASE_URL = f"https://{LIBERVIA_HOST}:{LIBERVIA_PORT_HTTPS}"
SIZE_DESKTOP = (1024, 728)
SIZE_MOBILE = (380, 640)
account_1_cookies = None
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item, call):
# needed to get test results in request fixture
# cf. https://docs.pytest.org/en/latest/example/simple.html#making-test-result-information-available-in-fixtures
outcome = yield
rep = outcome.get_result()
setattr(item, "rep_" + rep.when, rep)
@pytest.fixture
def screenshot_on_failure(request):
yield
if request.node.rep_setup.passed:
if request.node.rep_call.failed:
report_dir = Path(os.getenv("SAT_TEST_REPORT_DIR", "/tmp/tests_report"))
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
helium.get_driver().save_screenshot(str(dest_path))
print(f"screenshot saved to {dest_path}")
def wait_for_socket(host, port, retries=30):
sock = socket.socket()
while True:
try:
sock.connect((host, port))
except ConnectionRefusedError as e:
retries -= 1
if retries < 0:
print(f"Can't access server at {host}:{port}", file=sys.stderr)
raise e
else:
print(f"Can't connect to {host}:{port}, retrying ({retries})")
time.sleep(1)
else:
break
@pytest.fixture(scope="session")
def browser():
if os.getenv("SAT_TEST_E2E_LIBERVIA_NO_HEADLESS") is not None:
kwargs = {}
else:
kwargs = {"headless": True}
driver = helium.start_firefox(**kwargs)
driver.set_window_size(*SIZE_DESKTOP)
wait_for_socket(LIBERVIA_HOST, LIBERVIA_PORT_HTTPS)
yield helium
if os.getenv("SAT_TEST_E2E_LIBERVIA_KEEP_BROWSER") is None:
helium.kill_browser()
@pytest.fixture
def nobody_logged_in(browser):
browser.get_driver().delete_all_cookies()
@pytest.fixture
def log_in_account1(browser):
global account_1_cookies
if account_1_cookies is None:
browser.get_driver().delete_all_cookies()
browser.go_to("https://libervia.test:8443/login")
browser.write("account1", into="login")
browser.write("test", into="password")
browser.click("log in")
account_1_cookies = browser.get_driver().get_cookies()
else:
browser.get_driver().add_cookie(account_1_cookies)
@pytest.fixture
def mobile_screen(browser):
browser.get_driver().set_window_size(*SIZE_MOBILE)
@pytest.fixture
def desktop_screen(browser):
browser.get_driver().set_window_size(*SIZE_DESKTOP)

93
tests/e2e/libervia/test_libervia.py Normal file
View file

@ -0,0 +1,93 @@
#!/usr/bin/env python3
# SàT: an XMPP client
# Copyright (C) 2009-2020 Jérôme Poisson (goffi@goffi.org)
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import pytest
from helium import (
go_to, write, click, drag_file, find_all, wait_until, S, Text, Link, get_driver
)
if os.getenv("SAT_TEST_ENV_E2E_LIBERVIA") is None:
pytest.skip(
"skipping end-to-end tests, we are not in a test environment for Libervia",
allow_module_level=True
)
pytestmark = pytest.mark.usefixtures("test_profiles", "screenshot_on_failure")
class TestLogin:
def test_user_can_create_account(self, nobody_logged_in, sent_emails):
go_to("https://libervia.test:8443/login")
click("no account yet")
write("new_account", into="login")
write("some_email@example.net", into="email")
write("testtest", into="password")
click("register new account")
wait_until(lambda: get_driver().current_url.endswith("/login"))
write("testtest", into="password")
click("log in")
wait_until(Text("you are logged").exists)
wait_until(lambda: len(sent_emails) == 2)
if sent_emails[0].to == "admin@server1.test":
admin_email, user_email = sent_emails
else:
user_email, admin_email = sent_emails
assert admin_email.to == "admin@server1.test"
# profile name must be specified in admin email
assert "new_account" in admin_email.body
assert user_email.to == "some_email@example.net"
# user jid must be specified in the email
assert "new_account@server1.test" in user_email.body
# use can now log-in
def test_user_can_log_in(self, nobody_logged_in):
go_to("https://libervia.test:8443/login")
write("account1_s2", into="login")
write("test", into="password")
click("log in")
assert Text("you are logged").exists()
def test_wrong_password_fails(self, nobody_logged_in):
go_to("https://libervia.test:8443/login")
write("account1_s2", into="login")
write("wrong_password", into="password")
click("log in")
assert Text("Your login and/or password is incorrect. Please try again.")
class TestPhotos:
def test_user_can_create_album(self, log_in_account1):
go_to("https://libervia.test:8443/photos")
wait_until(Link("create").exists)
click("create")
write("test album", into="album name")
click("create")
album_link = Link("test album")
wait_until(album_link.exists)
click(album_link)
wait_until(lambda: not S("#loading_screen").exists())
drag_file("/src/sat/tests/_files/test_1.jpg", "drop photos here")
wait_until(lambda: len(find_all(S("div.progress_finished")))==1)
drag_file("/src/sat/tests/_files/test_2.jpg", "drop photos here")
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()

165
tests/e2e/run_e2e.py
View file

@ -22,28 +22,85 @@ import sys
from pathlib import Path
import tempfile
from textwrap import dedent
from datetime import datetime
import sh
import io
import re
import sat_templates
import libervia
from sat.core import exceptions
import yaml
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
KEEP_OPT = "--keep"
OPT_KEEP_CONTAINERS = "--keep-containers"
OPT_KEEP_PROFILES = "--keep-profiles"
OPT_KEEP_VNC = "--keep-vnc"
OPT_KEEP_BROWSER = "--keep-browser"
OPT_VISUAL = "--visual"
OPT_DEV_MODE = "--dev-mode"
dev_mode_inst = dedent("""\
Here is a short script to start working with a logged account:
from helium import *
start_firefox()
go_to("https://libervia.test:8443/login")
write("account1", "login")
write("test", "password")
click("log in")
""")
report_buffer = io.StringIO()
def live_out(data):
sys.stdout.write(data)
sys.stdout.flush()
report_buffer.write(data)
def live_err(data):
sys.stderr.write(data)
sys.stderr.flush()
report_buffer.write(data)
def get_opt(opt_name):
"""Check is an option flag is set, and remove it for sys.argv
This allow to have simple flags without interfering with pytest options
"""
if opt_name in sys.argv:
sys.argv.remove(opt_name)
return True
else:
return False
def set_env(override, name, value="1"):
"""Set environement variable"""
environment = override["services"]["sat"].setdefault("environment", {})
environment[name] = value
def use_e2e_env():
if KEEP_OPT in sys.argv:
visual = get_opt(OPT_VISUAL)
keep_containers = get_opt(OPT_KEEP_CONTAINERS)
keep_profiles = get_opt(OPT_KEEP_PROFILES)
keep_vnc = get_opt(OPT_KEEP_VNC)
keep_browser = get_opt(OPT_KEEP_BROWSER)
if keep_browser:
keep_containers = True
sys.argv.remove(KEEP_OPT)
else:
keep_containers = False
keep_vnc = True
if keep_vnc:
visual = True
dev_mode = get_opt(OPT_DEV_MODE)
if dev_mode:
keep_containers = keep_profiles = keep_vnc = visual = True
for p in Path.cwd().parents:
package_path = p / "sat"
docker_path = p / "docker"
@ -56,14 +113,32 @@ def use_e2e_env():
"from the backend repository?"
)
libervia_path = Path(libervia.__file__).parent.resolve()
libervia_root_path = libervia_path.parent
if (libervia_root_path / ".hg").is_dir():
libervia_source = libervia_root_path
libervia_target = "/src/libervia"
else:
libervia_source = libervia_path
libervia_target = "/src/libervia/libervia"
sat_templates_path = Path(sat_templates.__file__).parent.resolve()
sat_templates_root_path = sat_templates_path.parent
if (sat_templates_root_path / ".hg").is_dir():
sat_templates_source = sat_templates_root_path
sat_templates_target = "/src/sat_templates"
else:
sat_templates_source = sat_templates_path
sat_templates_target = "/src/sat_templates/sat_templates"
compose_e2e_path = docker_path / "docker-compose_e2e.yml"
if not compose_e2e_path.is_file():
raise exceptions.NotFound('"docker-compose_e2e.yml" file can\'t be found')
with tempfile.TemporaryDirectory() as temp_dir:
with tempfile.TemporaryDirectory(prefix="sat_test_e2e_") as temp_dir:
override_path = Path(temp_dir) / "test_override.yml"
with override_path.open("w") as f:
f.write(dedent(f"""\
override = yaml.load(
dedent(f"""\
version: "3.6"
services:
sat:
@ -72,23 +147,89 @@ def use_e2e_env():
source: {sat_root_path}
target: /src/sat
read_only: true
"""))
libervia:
volumes:
- type: bind
source: {sat_root_path}
target: /src/sat
read_only: true
- type: bind
source: {libervia_source}
target: {libervia_target}
read_only: true
- type: bind
source: {sat_templates_source}
target: {sat_templates_target}
read_only: true
"""
),
Loader=Loader
)
if keep_profiles:
set_env(override, "SAT_TEST_E2E_KEEP_PROFILES")
if visual:
set_env(override, "SAT_TEST_E2E_LIBERVIA_NO_HEADLESS")
if keep_browser:
set_env(override, "SAT_TEST_E2E_LIBERVIA_KEEP_BROWSER")
with override_path.open("w") as f:
yaml.dump(override, f, Dumper=Dumper)
docker_compose = sh.docker_compose.bake(
"-f", compose_e2e_path, "-f", override_path)
docker_compose.up("-d")
try:
docker_compose.exec(
p = docker_compose.exec(
"-T", "--workdir", "/src/sat/tests", "sat",
"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
_in=sys.stdin, _out=live_out, _out_bufsize=0, _err=live_err, _err_bufsize=0,
_bg=True
)
if visual:
vnc_port = docker_compose.port("sat", "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]
)
else:
p_vnc = None
try:
p.wait()
except sh.ErrorReturnCode as e:
sat_cont_id = docker_compose.ps("-q", "sat").strip()
report_dest = Path(f"reports_{datetime.now().isoformat()}/")
# we need to make `report_dest` explicitely local with "./", otherwise
# docker parse takes it as a container path due to the presence of ":"
# with `isoformat()`.
sh.docker.cp(f"{sat_cont_id}:/reports", f"./{report_dest}")
# we save 2 versions: one with ANSI escape codes
report_ansi = report_dest / "report.ansi"
with report_ansi.open('w') as f:
f.write(report_buffer.getvalue())
# and one without (cf. https://stackoverflow.com/a/14693789)
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
report_log = report_dest / "report.log"
with report_log.open('w') as f:
f.write(ansi_escape.sub('', report_buffer.getvalue()))
print(f"report saved to {report_dest}")
sys.exit(e.exit_code)
finally:
if p_vnc is not None and p_vnc.is_alive() and not keep_vnc:
p_vnc.terminate()
if not keep_containers:
docker_compose.down(volumes=True)
if dev_mode:
print(dev_mode_inst)
if __name__ == "__main__":
use_e2e_env()