Mercurial > libervia-backend
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 |