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