EHN: add send_mail method to sss helper

setup.py

@@ -154,6 +154,7 @@ def setup_package():
         long_description_content_type="text/markdown",
         author='Henrik tom Wörden',
         author_email='h.tomwoerden@indiscale.com',
+        install_requires=["caosdb>=0.4.0", "openpyxl"],
         packages=find_packages('src'),
         package_dir={'': 'src'},
         setup_requires=["pytest-runner>=2.0,<3dev"],

src/caosadvancedtools/serverside/helper.py

@@ -26,6 +26,8 @@ import json
 import logging
 import os
 import sys
+import subprocess
+from email import message, policy, utils
 
 import caosdb as db
 
@@ -312,3 +314,46 @@ def get_shared_filename(filename):
     filename = os.path.join(randname, filename)
 
     return filename, filepath
+
+
+def send_mail(from_addr, to, subject, body, cc=None, bcc=None):
+    """ Send an email via the configured send_mail client.
+
+    The relevant options in the pycaosdb.ini are:
+
+        [Misc]
+        sendmail = ...
+
+    Parameters:
+    -----------
+    from_addr : str
+        The sender's email address.
+    to : str or list of str
+        The recipient's email address.
+    subject : str
+        Subject of the email.
+    body : str
+        The mail body, i.e. the text message.
+    cc : str or list of str (optional)
+        Single or list of cc-recipients. Defaults to None.
+    bcc : str or list of str (optional)
+        Single or list of bcc-recipients. Defaults to None.
+    """
+
+    caosdb_config = db.configuration.get_config()
+    sendmail = caosdb_config["Misc"]["sendmail"]
+    mail = message.EmailMessage(policy=policy.SMTP)
+    mail.set_content(body)
+    mail["From"] = from_addr
+    mail["To"] = to if isinstance(to, str) else ", ".join(to)
+    mail["Subject"] = subject
+    mail['Date'] = utils.formatdate(localtime=True)
+
+    if cc is not None:
+        mail["CC"] = cc if isinstance(cc, str) else ", ".join(cc)
+    if bcc is not None:
+        mail["BCC"] = bcc if isinstance(cc, str) else ", ".join(cc)
+
+    p = subprocess.Popen([sendmail, "-t", "-oi"],
+                         stdin=subprocess.PIPE)
+    p.communicate(mail.as_bytes())

unittests/sss_helper_example_data.json

+{
+    "loan": 12345,
+    "box": 2345,
+    "first_name": "Anna",
+    "last_name": "Lytik",
+    "email": "a.lytik@example.com",
+    "expected_return_date": "2020-12-24",
+    "exhaust_contents": "true",
+    "comment": "this is a comment",
+    "destination": "blablabla destination",
+    "current_location": "blublublub"
+}

unittests/test_sss_helper.py

+from os.path import abspath, dirname, join, isfile, exists
+from os import listdir, remove
+from email import message_from_file, policy
+from caosdb import configure_connection, RecordType
+from caosdb.connection.mockup import (MockUpServerConnection, MockUpResponse)
+from caosadvancedtools.serverside.helper import (parse_arguments, get_data,
+                                                 init_data_model, send_mail)
+
+
+def get_data_example():
+    return abspath(join(dirname(__file__), "sss_helper_example_data.json"))
+
+
+def setup_module():
+    connection = configure_connection(url="unittests", username="testuser",
+                                      password_method="plain",
+                                      password="testpassword", timeout=200,
+                                      implementation=MockUpServerConnection)
+    entities = '<Response><RecordType name="Test" id="1234"/></Response>'
+    connection._delegate_connection.resources.append(
+        lambda **kwargs: MockUpResponse(200, {}, entities))
+
+
+def teardown_module():
+    for m in get_tmp_mails():
+        remove(m)
+
+
+def get_tmp_mails():
+    tmpmail = "/tmp/mail"
+    if not exists(tmpmail):
+        return []
+    mails = [join(tmpmail, f) for f in listdir(tmpmail) if isfile(join(tmpmail,
+                                                                       f))]
+    return mails
+
+
+def test_parse_arguments():
+    args = parse_arguments(["--auth-token", "1234ABCD", "test.json"])
+    assert args.filename == "test.json"
+    assert args.auth_token == "1234ABCD"
+
+
+def test_get_data():
+    data = get_data(filename=get_data_example())
+    # default={"test": "bla"})
+    assert data["box"] == 2345
+
+    data = get_data(filename=get_data_example(),
+                    default={"test": "bla", "comment": "no comment"})
+    assert data["box"] == 2345
+    assert data["test"] == "bla"
+    assert data["comment"] == "this is a comment"
+
+
+def test_init_data_model():
+    rt = RecordType(name="Test")
+    assert rt.id is None
+    init_data_model([rt])
+    assert rt.id == 1234
+
+
+def test_send_mail():
+    assert len(get_tmp_mails()) == 0
+    send_mail("me@example.com", "you@example.com", "the subject", "hello!")
+    mails = get_tmp_mails()
+    assert len(mails) == 1
+
+    with open(mails[0], "r") as f:
+        msg = message_from_file(f, policy=policy.SMTP)
+
+    assert msg["From"] == "me@example.com"
+    assert msg["To"] == "you@example.com"
+    assert msg["Subject"] == "the subject"
+    assert msg.get_content() == "hello!\n"