Mercurial > libervia-backend
diff tests/e2e/conftest.py @ 3429:d4558f3cbf13
tests, docker(e2e): added e2e tests for Libervia:
- moved jp tests to `e2e/jp`
- new fixtures
- adapted docker-compose
- improved `run_e2e` with several flags + report on failure
- doc to come
author | Goffi <goffi@goffi.org> |
---|---|
date | Fri, 27 Nov 2020 16:39:40 +0100 |
parents | 814e118d9ef3 |
children | f9011d62a87a |
line wrap: on
line diff
--- a/tests/e2e/conftest.py Fri Nov 27 16:32:40 2020 +0100 +++ b/tests/e2e/conftest.py Fri Nov 27 16:39:40 2020 +0100 @@ -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 @@ 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 @@ 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 @@ ) -@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