Merge branch 'staging' into azure-subscriptions
This commit is contained in:
@@ -164,6 +164,12 @@ def pdf_upload2():
|
||||
yield FileStorage(fp, content_type="application/pdf")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def downloaded_task_order():
|
||||
with open(PDF_FILENAME, "rb") as fp:
|
||||
yield {"name": "mock.pdf", "content": fp.read()}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def extended_financial_verification_data(pdf_upload):
|
||||
return {
|
||||
|
22
tests/domain/cloud/test_azure_file_service.py
Normal file
22
tests/domain/cloud/test_azure_file_service.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from atst.domain.csp.files import AzureFileService
|
||||
from azure.storage.blob.models import Blob
|
||||
|
||||
|
||||
class MockBlockBlobService(object):
|
||||
def __init__(self, exception=None, **kwargs):
|
||||
self.exception = exception
|
||||
|
||||
def get_blob_to_bytes(self, blob_name="test.pdf", **kwargs):
|
||||
if self.exception:
|
||||
raise self.exception
|
||||
else:
|
||||
return Blob(name=blob_name, content=b"mock content")
|
||||
|
||||
|
||||
def test_download_task_order_success(app, monkeypatch):
|
||||
file_service = AzureFileService(config=app.config)
|
||||
file_service.BlockBlobService = MockBlockBlobService
|
||||
|
||||
task_order = file_service.download_task_order("test.pdf")
|
||||
assert task_order["name"] == "test.pdf"
|
||||
assert task_order["content"] == b"mock content"
|
@@ -1,10 +1,17 @@
|
||||
import pytest
|
||||
from atst.utils.mailer import Mailer, Mailer, MailConnection, RedisConnection
|
||||
from atst.utils.mailer import (
|
||||
Mailer,
|
||||
MailConnection,
|
||||
RedisConnection,
|
||||
)
|
||||
from atst.utils.localization import translate
|
||||
from email.mime.base import MIMEBase
|
||||
|
||||
|
||||
class MockConnection(MailConnection):
|
||||
def __init__(self):
|
||||
self._messages = []
|
||||
self.sender = "mock@mock.com"
|
||||
|
||||
def send(self, message):
|
||||
self._messages.append(message)
|
||||
@@ -46,3 +53,55 @@ def test_redis_mailer_can_save_messages(app):
|
||||
assert message_data["recipients"][0] in message
|
||||
assert message_data["subject"] in message
|
||||
assert message_data["body"] in message
|
||||
|
||||
|
||||
def test_send_with_attachment(app, mailer, downloaded_task_order):
|
||||
to_number = "11111111111111"
|
||||
attachment = {
|
||||
"maintype": "application",
|
||||
"subtype": "pdf",
|
||||
"filename": downloaded_task_order["name"],
|
||||
"content": downloaded_task_order["content"],
|
||||
}
|
||||
mailer.send(
|
||||
recipients=[app.config["MICROSOFT_TASK_ORDER_EMAIL_ADDRESS"]],
|
||||
subject=translate("email.task_order_sent.subject", {"to_number": to_number}),
|
||||
body=translate("email.task_order_sent.body", {"to_number": to_number}),
|
||||
attachments=[attachment],
|
||||
)
|
||||
# one email was sent
|
||||
assert len(mailer.messages) == 1
|
||||
|
||||
# the email was sent to Microsoft with the correct subject line
|
||||
message = mailer.messages[0]
|
||||
assert message["To"] == app.config["MICROSOFT_TASK_ORDER_EMAIL_ADDRESS"]
|
||||
assert message["Subject"] == translate(
|
||||
"email.task_order_sent.subject", {"to_number": to_number}
|
||||
)
|
||||
|
||||
# the email was sent as a multipart message with two parts -- the message
|
||||
# body and the attachment
|
||||
assert message.is_multipart()
|
||||
message_payload = message.get_payload()
|
||||
assert len(message_payload) == 2
|
||||
|
||||
# A body and attachment were in the email
|
||||
body = next(
|
||||
(
|
||||
part
|
||||
for part in message_payload
|
||||
if part["Content-Type"] == 'text/plain; charset="utf-8"'
|
||||
),
|
||||
None,
|
||||
)
|
||||
attachment = next(
|
||||
(part for part in message_payload if part["Content-Type"] == "application/pdf"),
|
||||
None,
|
||||
)
|
||||
assert body
|
||||
assert attachment
|
||||
|
||||
assert (
|
||||
attachment["Content-Disposition"]
|
||||
== f"attachment; filename=\"{downloaded_task_order['name']}\""
|
||||
)
|
||||
|
Reference in New Issue
Block a user