comparison 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
comparison
equal deleted inserted replaced
3428:a6ea53248c14 3429:d4558f3cbf13
14 # GNU Affero General Public License for more details. 14 # GNU Affero General Public License for more details.
15 15
16 # You should have received a copy of the GNU Affero General Public License 16 # You should have received a copy of the GNU Affero General Public License
17 # along with this program. If not, see <http://www.gnu.org/licenses/>. 17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
18 18
19 import sys
20 import os 19 import os
21 import tempfile 20 import tempfile
22 import string 21 import string
23 import hashlib 22 import hashlib
24 import random 23 import random
25 from pathlib import Path 24 from pathlib import Path
26 from textwrap import dedent 25 from aiosmtpd.controller import Controller
27 import json 26 from aiosmtpd.handlers import Message
27 from email.message import EmailMessage
28 from sh import jp
28 import pytest 29 import pytest
29 from sh import jp
30
31
32 class JpJson:
33 """jp like commands parsing result as JSON"""
34
35 def __init__(self):
36 self.subcommands = []
37
38 def __call__(self, *args, **kwargs):
39 args = self.subcommands + list(args)
40 self.subcommands.clear()
41 kwargs['output'] = 'json_raw'
42 kwargs['_tty_out'] = False
43 cmd = jp(*args, **kwargs)
44 return json.loads(cmd.stdout)
45
46 def __getattr__(self, name):
47 if name.startswith('_'):
48 # no jp subcommand starts with a "_",
49 # and pytest uses some attributes with this name scheme
50 return super().__getattr__(name)
51 self.subcommands.append(name)
52 return self
53
54
55 class JpElt(JpJson):
56 """jp like commands parsing result as domishElement"""
57
58 def __init__(self):
59 super().__init__()
60 from sat.tools.xml_tools import ElementParser
61 self.parser = ElementParser()
62
63 def __call__(self, *args, **kwargs):
64 args = self.subcommands + list(args)
65 self.subcommands.clear()
66 kwargs['output'] = 'xml_raw'
67 kwargs['_tty_out'] = False
68 cmd = jp(*args, **kwargs)
69 return self.parser(cmd.stdout.decode().strip())
70
71
72 class Editor:
73
74 def __init__(self):
75 # temporary directory will be deleted Automatically when this object will be
76 # destroyed
77 self.tmp_dir_obj = tempfile.TemporaryDirectory(prefix="sat_e2e_test_editor_")
78 self.tmp_dir_path = Path(self.tmp_dir_obj.name)
79 if not sys.executable:
80 raise Exception("Can't find python executable")
81 self.editor_set = False
82 self.editor_path = self.tmp_dir_path / "editor.py"
83 self.ori_content_path = self.tmp_dir_path / "original_content"
84 self.new_content_path = self.tmp_dir_path / "new_content"
85 self.base_script = dedent(f"""\
86 #!{sys.executable}
87 import sys
88
89 def content_filter(content):
90 return {{content_filter}}
91
92 with open(sys.argv[1], 'r+') as f:
93 original_content = f.read()
94 f.seek(0)
95 new_content = content_filter(original_content)
96 f.write(new_content)
97 f.truncate()
98
99 with open("{self.ori_content_path}", "w") as f:
100 f.write(original_content)
101
102 with open("{self.new_content_path}", "w") as f:
103 f.write(new_content)
104 """
105 )
106 self._env = os.environ.copy()
107 self._env["EDITOR"] = str(self.editor_path)
108
109 def set_filter(self, content_filter: str = "content"):
110 """Python code to modify original content
111
112 The code will be applied to content received by editor.
113 The original content received by editor is in the "content" variable.
114 If filter_ is not specified, original content is written unmodified.
115 Code must be on a single line.
116 """
117 if '\n' in content_filter:
118 raise ValueError("new lines can't be used in filter_")
119 with self.editor_path.open('w') as f:
120 f.write(self.base_script.format(content_filter=content_filter))
121 self.editor_path.chmod(0o700)
122 self.editor_set = True
123
124 @property
125 def env(self):
126 """Get environment variable with the editor set"""
127 if not self.editor_set:
128 self.set_filter()
129 return self._env
130
131 @property
132 def original_content(self):
133 """Last content received by editor, before any modification
134
135 returns None if editor has not yet been called
136 """
137 try:
138 with self.ori_content_path.open() as f:
139 return f.read()
140 except FileNotFoundError:
141 return None
142
143 @property
144 def new_content(self):
145 """Last content writen by editor
146
147 This is the final content, after filter has been applied to original content
148 returns None if editor has not yet been called
149 """
150 try:
151 with self.new_content_path.open() as f:
152 return f.read()
153 except FileNotFoundError:
154 return None
155 30
156 31
157 class FakeFile: 32 class FakeFile:
158 ALPHABET = f"{string.ascii_letters}{string.digits}_" 33 ALPHABET = f"{string.ascii_letters}{string.digits}_"
159 BUF_SIZE = 65535 34 BUF_SIZE = 65535
223 break 98 break
224 hash_.update(buf) 99 hash_.update(buf)
225 return hash_.hexdigest() 100 return hash_.hexdigest()
226 101
227 102
228 @pytest.fixture(scope="session") 103 class TestMessage(EmailMessage):
229 def jp_json(): 104
230 """Run jp with "json_raw" output, and returns the parsed value""" 105 @property
231 return JpJson() 106 def subject(self):
232 107 return self['subject']
233 108
234 @pytest.fixture(scope="session") 109 @property
235 def jp_elt(): 110 def from_(self):
236 """Run jp with "xml_raw" output, and returns the parsed value""" 111 return self['from']
237 return JpElt() 112
113 @property
114 def to(self):
115 return self['to']
116
117 @property
118 def body(self):
119 return self.get_payload(decode=True).decode()
120
121
122 class SMTPMessageHandler(Message):
123 messages = []
124
125 def __init__(self):
126 super().__init__(message_class=TestMessage)
127
128 def handle_message(self, message):
129 self.messages.append(message)
238 130
239 131
240 @pytest.fixture(scope="session") 132 @pytest.fixture(scope="session")
241 def test_profiles(): 133 def test_profiles():
242 """Test accounts created using in-band registration 134 """Test accounts created using in-band registration
270 host=f"server{server_idx}.test" 162 host=f"server{server_idx}.test"
271 ) 163 )
272 jp.profile.modify(profile="account1", default=True, connect=True) 164 jp.profile.modify(profile="account1", default=True, connect=True)
273 jp.profile.connect(profile="account1_s2", connect=True) 165 jp.profile.connect(profile="account1_s2", connect=True)
274 yield tuple(profiles) 166 yield tuple(profiles)
275 for profile in profiles: 167 # This environment may be used during tests development
276 jp.account.delete(profile=profile, connect=True, force=True) 168 if os.getenv("SAT_TEST_E2E_KEEP_PROFILES") == None:
277 jp.profile.delete(profile, force=True) 169 for profile in profiles:
170 jp.account.delete(profile=profile, connect=True, force=True)
171 jp.profile.delete(profile, force=True)
278 172
279 173
280 @pytest.fixture(scope="class") 174 @pytest.fixture(scope="class")
281 def pubsub_nodes(test_profiles): 175 def pubsub_nodes(test_profiles):
282 """Create 2 testing nodes 176 """Create 2 testing nodes
283 177
284 Both nodes will be created with "account1" profile, named "test" and have and "open" 178 Both nodes will be created with "account1" profile, named "test" and have an "open"
285 access model. 179 access model.
286 One node will account1's PEP, the other one on pubsub.server1.test. 180 One node will account1's PEP, the other one on pubsub.server1.test.
287 """ 181 """
288 jp.pubsub.node.create( 182 jp.pubsub.node.create(
289 "-f", "access_model", "open", 183 "-f", "access_model", "open",
306 profile="account1", 200 profile="account1",
307 force=True 201 force=True
308 ) 202 )
309 203
310 204
311 @pytest.fixture()
312 def editor():
313 """Create a fake editor to automatise edition from CLI"""
314 return Editor()
315
316
317 @pytest.fixture(scope="session") 205 @pytest.fixture(scope="session")
318 def fake_file(): 206 def fake_file():
319 """Manage dummy files creation and destination path""" 207 """Manage dummy files creation and destination path"""
320 return FakeFile() 208 return FakeFile()
209
210
211 @pytest.fixture(scope="session")
212 def test_files():
213 """Return a Path to test files directory"""
214 return Path(__file__).parent.parent / "_files"
215
216
217 @pytest.fixture(scope="session")
218 def fake_smtp():
219 """Create a fake STMP server to check sent emails"""
220 controller = Controller(SMTPMessageHandler())
221 controller.hostname = "0.0.0.0"
222 controller.start()
223 yield
224 controller.stop()
225
226
227 @pytest.fixture
228 def sent_emails(fake_smtp):
229 """Catch email sent during the tests"""
230 SMTPMessageHandler.messages.clear()
231 return SMTPMessageHandler.messages