Mercurial > libervia-backend

comparison tests/unit/test_email_gateway.py @ 4318:27bb22eace65

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
tests (unit/email gateway): add test for XEP-0131 handling: rel 451
author Goffi <goffi@goffi.org>
date Sat, 28 Sep 2024 15:59:48 +0200
parents d27228b3c704
children
comparison
equal deleted inserted replaced
4317:055930cc81f9 4318:27bb22eace65
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 from email.message import EmailMessage
20 from email.parser import BytesParser
19 from email.utils import formataddr 21 from email.utils import formataddr
20 from unittest.mock import AsyncMock, MagicMock 22 from unittest.mock import AsyncMock, MagicMock
23
21 import pytest 24 import pytest
22 from pytest_twisted import ensureDeferred as ed 25 from pytest_twisted import ensureDeferred as ed
23 from twisted.words.protocols.jabber import jid 26 from twisted.words.protocols.jabber import jid
24 from email.message import EmailMessage 27
25 from libervia.backend.plugins.plugin_comp_email_gateway import EmailGatewayComponent 28 from libervia.backend.plugins.plugin_comp_email_gateway import (
29 EmailGatewayComponent,
30 SendMailExtra,
31 )
26 from libervia.backend.plugins.plugin_comp_email_gateway.models import ( 32 from libervia.backend.plugins.plugin_comp_email_gateway.models import (
27 Credentials, 33 Credentials,
28 UserData, 34 UserData,
29 ) 35 )
36 from libervia.backend.plugins.plugin_xep_0131 import HeadersData, Urgency
30 37
31 38
32 class TestEmailGatewayComponent: 39 class TestEmailGatewayComponent:
33 @pytest.fixture 40 @pytest.fixture
34 def email_gw(self, host): 41 def email_gw(self, host):
120 call_args = client.sendMessage.call_args[0] 127 call_args = client.sendMessage.call_args[0]
121 assert call_args[0] == to_jid 128 assert call_args[0] == to_jid
122 assert call_args[1] == {"": "Hello, world!\n"} 129 assert call_args[1] == {"": "Hello, world!\n"}
123 assert call_args[2] == None 130 assert call_args[2] == None
124 client.sendMessage.assert_called_once_with( 131 client.sendMessage.assert_called_once_with(
125 to_jid, {"": "Hello, world!\n"}, None, extra=None 132 to_jid, {"": "Hello, world!\n"}, None, extra={}
126 ) 133 )
127 134
128 @ed 135 @ed
129 async def test_on_new_email_multiple_recipients(self, email_gw): 136 async def test_on_new_email_multiple_recipients(self, email_gw):
130 """Email with multiple recipients is correctly processed and sent .""" 137 """Email with multiple recipients is correctly processed and sent ."""
166 } 173 }
167 ], 174 ],
168 } 175 }
169 }, 176 },
170 ) 177 )
178
179 @ed
180 async def test_send_email_with_headers(self, email_gw, monkeypatch):
181 """Email is sent with correct headers."""
182 email_gw.client = MagicMock()
183 email_gw.client.jid = jid.JID("gateway.example.org")
184 email_gw.storage = MagicMock()
185 email_gw.storage.get = AsyncMock(
186 return_value={
187 "user_email": "user@example.org",
188 "user_name": "Sender Name",
189 "smtp_host": "smtp.example.org",
190 "smtp_port": "587",
191 "smtp_username": "sender",
192 "smtp_password": "password",
193 }
194 )
195
196 from_jid = jid.JID("user@example.org")
197 to_email = "recipient@example.com"
198 body = "Hello, world!"
199 subject = "Test email"
200 headers = HeadersData(keywords="important,urgent", urgency=Urgency.high)
201
202 # Mock the smtp.sendmail function
203 sendmail_mock = AsyncMock()
204 monkeypatch.setattr("twisted.mail.smtp.sendmail", sendmail_mock)
205
206 await email_gw.send_email(
207 from_jid, to_email, body, subject, extra=SendMailExtra(headers=headers)
208 )
209
210 sendmail_mock.assert_called_once()
211
212 # Extract the email content from the call arguments
213 call_args = sendmail_mock.call_args[0]
214 _, _, _, email_content_bytes = call_args
215
216 # Parse the email content
217 parser = BytesParser()
218 msg = parser.parsebytes(email_content_bytes)
219
220 # Assert the headers are correctly set
221 assert msg["Keywords"] == headers.keywords
222 assert msg["Importance"] == "high"
223
224 @ed
225 async def test_on_new_email_with_headers(self, email_gw):
226 """Headers from the email are correctly processed and included in the message."""
227 client = MagicMock()
228 client.get_virtual_client = lambda __: client
229 client.jid = jid.JID("gateway.example.org")
230 client.sendMessage = AsyncMock()
231 email_gw.client = client
232
233 email = EmailMessage()
234 email["from"] = formataddr(("User Name", "sender@somewhere.example"))
235 email["to"] = "user@example.org"
236 email.set_content("Hello, world!")
237 email["Keywords"] = "test, example"
238 email["Importance"] = "high"
239
240 to_jid = jid.JID("gw-user@example.org")
241 user_email = "user@example.org"
242 user_data = UserData(Credentials({"user_email": user_email}))
243 await email_gw.on_new_email(user_data, to_jid, email)
244
245 client.sendMessage.assert_called_once_with(
246 to_jid,
247 {"": "Hello, world!\n"},
248 None,
249 extra={"headers": {"keywords": "test, example", "urgency": "high"}},
250 )