Differential D8818 Diff 31796 swh/vault/tests/test_backend.py

Changeset View

Standalone View

swh/vault/tests/test_backend.py

# Copyright (C) 2017-2020 The Software Heritage developers		# Copyright (C) 2017-2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import contextlib		import contextlib
import datetime		import datetime
import re		import re
import smtplib		import smtplib
▲ Show 20 Lines • Show All 47 Lines • ▼ Show 20 Lines

def fail_cook(backend, bundle_type, swhid, failure_reason):		def fail_cook(backend, bundle_type, swhid, failure_reason):
with mock_cooking(backend):		with mock_cooking(backend):
backend.create_task(bundle_type, swhid)		backend.create_task(bundle_type, swhid)
backend.set_status(bundle_type, swhid, "failed")		backend.set_status(bundle_type, swhid, "failed")
backend.set_progress(bundle_type, swhid, failure_reason)		backend.set_progress(bundle_type, swhid, failure_reason)


TEST_TYPE = "gitfast"		TEST_TYPE = "git_bare"
TEST_SWHID = CoreSWHID.from_string("swh:1:rev:4a4b9771542143cf070386f86b4b92d42966bdbc")		TEST_SWHID = CoreSWHID.from_string("swh:1:rev:4a4b9771542143cf070386f86b4b92d42966bdbc")
TEST_PROGRESS = (		TEST_PROGRESS = (
"Mr. White, You're telling me you're cooking again? \N{ASTONISHED FACE} "		"Mr. White, You're telling me you're cooking again? \N{ASTONISHED FACE} "
)		)
TEST_EMAIL = "ouiche@lorraine.fr"		TEST_EMAIL = "ouiche@lorraine.fr"


@pytest.fixture		@pytest.fixture
▲ Show 20 Lines • Show All 142 Lines • ▼ Show 20 Lines	def test_send_all_emails(swh_vault):
swh_vault.set_status(TEST_TYPE, TEST_SWHID, "done")		swh_vault.set_status(TEST_TYPE, TEST_SWHID, "done")

with patch.object(swh_vault, "_smtp_send") as m:		with patch.object(swh_vault, "_smtp_send") as m:
swh_vault.send_notif(TEST_TYPE, TEST_SWHID)		swh_vault.send_notif(TEST_TYPE, TEST_SWHID)

sent_emails = {k[0][0] for k in m.call_args_list}		sent_emails = {k[0][0] for k in m.call_args_list}
assert {k["To"] for k in sent_emails} == set(emails)		assert {k["To"] for k in sent_emails} == set(emails)

		download_url = (
		"https://archive.softwareheritage.org/api/1/vault/"
		f"{TEST_TYPE.replace('_', '-')}/{str(TEST_SWHID)}/raw"
		)

for e in sent_emails:		for e in sent_emails:
assert "bot@softwareheritage.org" in e["From"]		assert "bot@softwareheritage.org" in e["From"]
assert TEST_TYPE in e["Subject"]		assert TEST_TYPE in e["Subject"]
assert TEST_SWHID.object_id.hex()[:5] in e["Subject"]		assert TEST_SWHID.object_id.hex()[:5] in e["Subject"]
assert TEST_TYPE in str(e)		assert TEST_TYPE in str(e)
assert "https://archive.softwareheritage.org/" in str(e)		assert download_url in str(e)
assert TEST_SWHID.object_id.hex()[:5] in str(e)		assert TEST_SWHID.object_id.hex()[:5] in str(e)
assert "--\x20\n" in str(e) # Well-formated signature!!!		assert "--\x20\n" in str(e) # Well-formated signature!!!

# Check that the entries have been deleted and recalling the		# Check that the entries have been deleted and recalling the
# function does not re-send the e-mails		# function does not re-send the e-mails
m.reset_mock()		m.reset_mock()
swh_vault.send_notif(TEST_TYPE, TEST_SWHID)		swh_vault.send_notif(TEST_TYPE, TEST_SWHID)
m.assert_not_called()		m.assert_not_called()
Show All 11 Lines	def test_send_email_error_no_smtp(swh_vault):
swh_vault.send_notif(TEST_TYPE, TEST_SWHID)		swh_vault.send_notif(TEST_TYPE, TEST_SWHID)

assert len(reports) == 6		assert len(reports) == 6
for i, email in enumerate(emails):		for i, email in enumerate(emails):
# first report is the logger.error		# first report is the logger.error
assert reports[2 * i]["level"] == "error"		assert reports[2 * i]["level"] == "error"
assert reports[2 * i]["logger"] == "swh.vault.backend"		assert reports[2 * i]["logger"] == "swh.vault.backend"
reg = re.compile(		reg = re.compile(
"Unable to send SMTP message 'Bundle ready: gitfast [0-9a-f]{7}' "		f"Unable to send SMTP message 'Bundle ready: {TEST_TYPE} [0-9a-f]{{7}}' "
f"to {email.replace('+', '[+]')}: cannot connect to server"		f"to {email.replace('+', '[+]')}: cannot connect to server"
)		)
assert reg.match(reports[2 * i]["logentry"]["message"])		assert reg.match(reports[2 * i]["logentry"]["message"])
# second is the sentry_sdk.capture_message		# second is the sentry_sdk.capture_message
assert reports[2 * i + 1]["level"] == "error"		assert reports[2 * i + 1]["level"] == "error"
assert reg.match(reports[2 * i + 1]["message"])		assert reg.match(reports[2 * i + 1]["message"])


Show All 18 Lines	def test_send_email_error_send_failed(swh_vault):
# first one is the captured exception		# first one is the captured exception
assert reports[0]["level"] == "error"		assert reports[0]["level"] == "error"
assert reports[0]["exception"]["values"][0]["type"] == "SMTPHeloError"		assert reports[0]["exception"]["values"][0]["type"] == "SMTPHeloError"

# the following 3 ones are the sentry_sdk.capture_message() calls		# the following 3 ones are the sentry_sdk.capture_message() calls
for i, email in enumerate(emails, start=1):		for i, email in enumerate(emails, start=1):
assert reports[i]["level"] == "error"		assert reports[i]["level"] == "error"
reg = re.compile(		reg = re.compile(
"Unable to send SMTP message 'Bundle ready: gitfast [0-9a-f]{7}' "		f"Unable to send SMTP message 'Bundle ready: {TEST_TYPE} [0-9a-f]{{7}}' "
f"to {email.replace('+', '[+]')}: [(]404, 'HELO Failed'[)]"		f"to {email.replace('+', '[+]')}: [(]404, 'HELO Failed'[)]"
)		)
assert reg.match(reports[i]["message"])		assert reg.match(reports[i]["message"])


def test_available(swh_vault):		def test_available(swh_vault):
assert not swh_vault.is_available(TEST_TYPE, TEST_SWHID)		assert not swh_vault.is_available(TEST_TYPE, TEST_SWHID)

▲ Show 20 Lines • Show All 114 Lines • Show Last 20 Lines