Add function to send email and tests

.github/workflows/pipeline.yml

@@ -90,6 +90,8 @@ jobs:
             LITELLM_MODEL: ${{ secrets.LITELLM_MODEL }}
             LITELLM_API_BASE: ${{ secrets.LITELLM_API_BASE }}
             LITELLM_API_KEY: ${{ secrets.LITELLM_API_KEY }}
+            INFOBIP_API_KEY: ${{ secrets.INFOBIP_API_KEY }}
+            INFOBIP_BASE_URL: ${{ secrets.INFOBIP_BASE_URL }}
         steps:
             - uses: actions/checkout@v3 # Don't change it to cheackout@v4. V4 is not working with container image.
             # This is to fix GIT not liking owner of the checkout dir - https://github.com/actions/runner/issues/2033#issuecomment-1204205989

captn/email/__init__.py

@@ -0,0 +1,5 @@
+from captn.email.send_email import (
+    send_email,
+)
+
+__all__ = ("send_email",)

captn/email/send_email.py

@@ -0,0 +1,77 @@
+import http.client
+import json
+from codecs import encode
+from os import environ
+from typing import Any, Dict
+
+INFOBIP_API_KEY = environ["INFOBIP_API_KEY"]
+INFOBIP_BASE_URL = environ["INFOBIP_BASE_URL"]
+
+
+# ToDo: Create support@captn.ai and add to infobip portal
+def send_email(
+    *, from_email: str = "harish@airt.ai", to_email: str, subject: str, body_text: str
+) -> Dict[str, Any]:
+    conn = http.client.HTTPSConnection(  # nosemgrep: python.lang.security.audit.httpsconnection-detected.httpsconnection-detected
+        INFOBIP_BASE_URL
+    )
+    dataList = []
+    boundary = "wL36Yn8afVp8Ag7AmP8qZ0SA4n1v9T"
+    dataList.append(encode("--" + boundary))
+    dataList.append(encode("Content-Disposition: form-data; name=from;"))
+
+    dataList.append(encode("Content-Type: {}".format("text/plain")))
+    dataList.append(encode(""))
+
+    dataList.append(encode(from_email))
+    dataList.append(encode("--" + boundary))
+    dataList.append(encode("Content-Disposition: form-data; name=subject;"))
+
+    dataList.append(encode("Content-Type: {}".format("text/plain")))
+    dataList.append(encode(""))
+
+    dataList.append(encode(subject))
+    dataList.append(encode("--" + boundary))
+    dataList.append(encode("Content-Disposition: form-data; name=to;"))
+
+    dataList.append(encode("Content-Type: {}".format("text/plain")))
+    dataList.append(encode(""))
+
+    dataList.append(encode('{"to":"' + to_email + '"}'))
+    dataList.append(encode("--" + boundary))
+    dataList.append(encode("Content-Disposition: form-data; name=text;"))
+
+    dataList.append(encode("Content-Type: {}".format("text/plain")))
+    dataList.append(encode(""))
+
+    dataList.append(encode(body_text))
+    dataList.append(encode("--" + boundary + "--"))
+    dataList.append(encode(""))
+    body = b"\r\n".join(dataList)
+    payload = body
+    headers = {
+        "Authorization": f"App {INFOBIP_API_KEY}",
+        "Accept": "application/json",
+        "Content-type": f"multipart/form-data; boundary={boundary}",
+    }
+
+    conn.request("POST", "/email/3/send", payload, headers)
+    res = conn.getresponse()
+
+    data = res.read()
+    decoded: Dict[str, Any] = json.loads(data.decode("utf-8"))
+
+    if res.status != 200:
+        raise Exception(f"Failed to send email: {res.status}, {decoded}")
+
+    return decoded
+
+
+if __name__ == "__main__":
+    r = send_email(
+        from_email="harish@airt.ai",
+        to_email="kumaran@airt.ai",
+        subject="Hi there!",
+        body_text="It is me, SDK!",
+    )
+    print(r)

scripts/deploy.sh

@@ -136,4 +136,5 @@ ssh -o StrictHostKeyChecking=no -i key.pem azureuser@"$DOMAIN" "docker run --nam
 	-e AZURE_MODEL='$AZURE_MODEL' -e AZURE_OPENAI_API_KEY_SWEEDEN='$AZURE_OPENAI_API_KEY_SWEEDEN' \
 	-e AZURE_OPENAI_API_KEY_CANADA='$AZURE_OPENAI_API_KEY_CANADA' -e OPENAI_API_KEY='$OPENAI_API_KEY' \
 	-e LITELLM_MODEL='$LITELLM_MODEL' -e LITELLM_API_BASE='$LITELLM_API_BASE' -e LITELLM_API_KEY='$LITELLM_API_KEY' \
+	-e INFOBIP_BASE_URL='$INFOBIP_BASE_URL' -e INFOBIP_API_KEY='$INFOBIP_API_KEY' \
 	-d ghcr.io/$GITHUB_REPOSITORY:$TAG"

tests/captn_agents/test_send_email.py

@@ -0,0 +1,103 @@
+import http.client
+import json
+from typing import Any, Dict
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class DummyResponse:
+    def __init__(self, status: int, d: Dict[str, Any], **kwargs) -> None:  # type: ignore [no-untyped-def]
+        self.status = status
+        self.d = d
+
+    def read(self, *args, **kwargs):  # type: ignore [no-untyped-def]
+        return str.encode(json.dumps(self.d))
+
+
+@pytest.fixture
+def mock_https_connection(monkeypatch):  # type: ignore [no-untyped-def]
+    monkeypatch.setattr(http.client.HTTPSConnection, "request", MagicMock())
+
+
+@pytest.fixture
+def set_env_variables(monkeypatch):  # type: ignore [no-untyped-def]
+    monkeypatch.setenv("INFOBIP_API_KEY", "123")
+    monkeypatch.setenv("INFOBIP_BASE_URL", "dummy.base.com")
+
+
+@pytest.fixture
+def mock_response(monkeypatch, status_code, response_data):  # type: ignore [no-untyped-def]
+    def _mock_response(*args, **kwargs):  # type: ignore [no-untyped-def]
+        return DummyResponse(status=status_code, d=response_data)
+
+    monkeypatch.setattr(http.client.HTTPSConnection, "getresponse", _mock_response)
+
+
+@pytest.mark.parametrize(
+    "status_code, response_data, expected_exception",
+    [
+        (
+            200,
+            {
+                "bulkId": "t3j7tho5rk69t2f3bruh",
+                "messages": [
+                    {
+                        "to": "test@airt.ai",
+                        "messageId": "692g3652bq2tpvps2nk2",
+                        "status": {
+                            "groupId": 1,
+                            "groupName": "PENDING",
+                            "id": 26,
+                            "name": "PENDING_ACCEPTED",
+                            "description": "Message accepted, pending for delivery.",
+                        },
+                    }
+                ],
+            },
+            None,
+        ),
+        (
+            500,
+            {
+                "bulkId": "t3j7tho5rk69t2f3bruh",
+                "messages": [
+                    {
+                        "to": "test@airt.ai",
+                        "messageId": "692g3652bq2tpvps2nk2",
+                        "status": {
+                            "groupId": 1,
+                            "groupName": "PENDING",
+                            "id": 26,
+                            "name": "REJECTED",
+                            "description": "Message rejected.",
+                        },
+                    }
+                ],
+            },
+            "Failed to send email: 500",
+        ),
+    ],
+)
+def test_send_email(  # type: ignore [no-untyped-def]
+    mock_https_connection,
+    set_env_variables,
+    mock_response,
+    status_code,
+    response_data,
+    expected_exception,
+):
+    from captn.email import send_email
+
+    if expected_exception:
+        with pytest.raises(Exception) as exc_info:
+            send_email(
+                to_email="test@airt.ai", subject="Hi there!", body_text="It is me, SDK!"
+            )
+
+        assert expected_exception in str(exc_info.value)
+    else:
+        response = send_email(
+            to_email="test@airt.ai", subject="Hi there!", body_text="It is me, SDK!"
+        )
+        assert response == response_data