mirror of
https://gitlab.archlinux.org/archlinux/aurweb.git
synced 2025-02-03 10:43:03 +01:00
feat(testing.email): add Email.dump
Signed-off-by: Kevin Morris <kevr@0cost.org>
This commit is contained in:
parent
60b098a2f2
commit
85e6ad03db
3 changed files with 110 additions and 4 deletions
|
@ -1,8 +1,12 @@
|
||||||
import base64
|
import base64
|
||||||
|
import binascii
|
||||||
import copy
|
import copy
|
||||||
import email
|
import email
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
|
import sys
|
||||||
|
|
||||||
|
from typing import TextIO
|
||||||
|
|
||||||
|
|
||||||
class Email:
|
class Email:
|
||||||
|
@ -26,10 +30,13 @@ class Email:
|
||||||
"""
|
"""
|
||||||
TEST_DIR = "test-emails"
|
TEST_DIR = "test-emails"
|
||||||
|
|
||||||
def __init__(self, serial: int = 1):
|
def __init__(self, serial: int = 1, autoparse: bool = True):
|
||||||
self.serial = serial
|
self.serial = serial
|
||||||
self.content = self._get()
|
self.content = self._get()
|
||||||
|
|
||||||
|
if autoparse:
|
||||||
|
self._parse()
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def email_prefix(suite: bool = False) -> str:
|
def email_prefix(suite: bool = False) -> str:
|
||||||
"""
|
"""
|
||||||
|
@ -77,7 +84,7 @@ class Email:
|
||||||
with open(path) as f:
|
with open(path) as f:
|
||||||
return f.read()
|
return f.read()
|
||||||
|
|
||||||
def parse(self) -> "Email":
|
def _parse(self) -> "Email":
|
||||||
"""
|
"""
|
||||||
Parse this email and base64-decode the body.
|
Parse this email and base64-decode the body.
|
||||||
|
|
||||||
|
@ -94,7 +101,10 @@ class Email:
|
||||||
|
|
||||||
# aurweb email notifications always have base64 encoded content.
|
# aurweb email notifications always have base64 encoded content.
|
||||||
# Decode it here so self.body is human readable.
|
# Decode it here so self.body is human readable.
|
||||||
self.body = base64.b64decode(self.message.get_payload()).decode()
|
try:
|
||||||
|
self.body = base64.b64decode(self.message.get_payload()).decode()
|
||||||
|
except (binascii.Error, UnicodeDecodeError):
|
||||||
|
self.body = self.message.get_payload()
|
||||||
|
|
||||||
path = self._email_path()
|
path = self._email_path()
|
||||||
with open(path, "w") as f:
|
with open(path, "w") as f:
|
||||||
|
@ -102,6 +112,9 @@ class Email:
|
||||||
|
|
||||||
return self
|
return self
|
||||||
|
|
||||||
|
def parse(self) -> "Email":
|
||||||
|
return self
|
||||||
|
|
||||||
def glue(self) -> str:
|
def glue(self) -> str:
|
||||||
"""
|
"""
|
||||||
Glue parsed content back into a complete email document, but
|
Glue parsed content back into a complete email document, but
|
||||||
|
@ -110,7 +123,9 @@ class Email:
|
||||||
:return: Email document as a string
|
:return: Email document as a string
|
||||||
"""
|
"""
|
||||||
headers = copy.copy(self.headers)
|
headers = copy.copy(self.headers)
|
||||||
del headers["Content-Transfer-Encoding"]
|
|
||||||
|
if "Content-Transfer-Encoding" in headers:
|
||||||
|
headers.pop("Content-Transfer-Encoding")
|
||||||
|
|
||||||
output = []
|
output = []
|
||||||
for k, v in headers.items():
|
for k, v in headers.items():
|
||||||
|
@ -118,3 +133,23 @@ class Email:
|
||||||
output.append("")
|
output.append("")
|
||||||
output.append(self.body)
|
output.append(self.body)
|
||||||
return "\n".join(output)
|
return "\n".join(output)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def dump(file: TextIO = sys.stdout) -> None:
|
||||||
|
"""
|
||||||
|
Dump emails content to `file`.
|
||||||
|
|
||||||
|
This function is intended to be used to debug email issues
|
||||||
|
while testing something relevent to email.
|
||||||
|
|
||||||
|
:param file: Writable file object
|
||||||
|
"""
|
||||||
|
lines = []
|
||||||
|
for i in range(Email.count()):
|
||||||
|
email = Email(i + 1)
|
||||||
|
lines += [
|
||||||
|
f"== Email #{i + 1} ==",
|
||||||
|
email.glue(),
|
||||||
|
f"== End of Email #{i + 1}"
|
||||||
|
]
|
||||||
|
print("\n".join(lines), file=file)
|
||||||
|
|
|
@ -217,3 +217,15 @@ def db_test(db_session: scoped_session) -> None:
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def git(tmpdir: py.path.local) -> GitRepository:
|
def git(tmpdir: py.path.local) -> GitRepository:
|
||||||
yield GitRepository(tmpdir)
|
yield GitRepository(tmpdir)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def email_test() -> None:
|
||||||
|
"""
|
||||||
|
A decoupled test email setup fixture.
|
||||||
|
|
||||||
|
When using the `db_test` fixture, this fixture is redundant. Otherwise,
|
||||||
|
email tests need to run through our `setup_email` function to ensure
|
||||||
|
that we set them up to be used via aurweb.testing.email.Email.
|
||||||
|
"""
|
||||||
|
setup_email()
|
||||||
|
|
59
test/test_email.py
Normal file
59
test/test_email.py
Normal file
|
@ -0,0 +1,59 @@
|
||||||
|
import io
|
||||||
|
|
||||||
|
from subprocess import PIPE, Popen
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from aurweb import config
|
||||||
|
from aurweb.testing.email import Email
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def setup(email_test):
|
||||||
|
return
|
||||||
|
|
||||||
|
|
||||||
|
def sendmail(from_: str, to_: str, content: str) -> Email:
|
||||||
|
binary = config.get("notifications", "sendmail")
|
||||||
|
proc = Popen(binary, stdin=PIPE, stdout=PIPE, stderr=PIPE)
|
||||||
|
content = f"From: {from_}\nTo: {to_}\n\n{content}"
|
||||||
|
proc.communicate(content.encode())
|
||||||
|
proc.wait()
|
||||||
|
assert proc.returncode == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_email_glue():
|
||||||
|
""" Test that Email.glue() decodes both base64 and decoded content. """
|
||||||
|
body = "Test email."
|
||||||
|
sendmail("test@example.org", "test@example.org", body)
|
||||||
|
assert Email.count() == 1
|
||||||
|
|
||||||
|
email1 = Email(1)
|
||||||
|
email2 = Email(1)
|
||||||
|
assert email1.glue() == email2.glue()
|
||||||
|
|
||||||
|
|
||||||
|
def test_email_dump():
|
||||||
|
""" Test that Email.dump() dumps a single email. """
|
||||||
|
body = "Test email."
|
||||||
|
sendmail("test@example.org", "test@example.org", body)
|
||||||
|
assert Email.count() == 1
|
||||||
|
|
||||||
|
stdout = io.StringIO()
|
||||||
|
Email.dump(file=stdout)
|
||||||
|
content = stdout.getvalue()
|
||||||
|
assert "== Email #1 ==" in content
|
||||||
|
|
||||||
|
|
||||||
|
def test_email_dump_multiple():
|
||||||
|
""" Test that Email.dump() dumps multiple emails. """
|
||||||
|
body = "Test email."
|
||||||
|
sendmail("test@example.org", "test@example.org", body)
|
||||||
|
sendmail("test2@example.org", "test2@example.org", body)
|
||||||
|
assert Email.count() == 2
|
||||||
|
|
||||||
|
stdout = io.StringIO()
|
||||||
|
Email.dump(file=stdout)
|
||||||
|
content = stdout.getvalue()
|
||||||
|
assert "== Email #1 ==" in content
|
||||||
|
assert "== Email #2 ==" in content
|
Loading…
Add table
Reference in a new issue