144 lines
6.5 KiB
Python
144 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_mail.backend.router import create_profile, deactivate_profile, update_profile
|
|
from govoplan_mail.backend.schemas import MailServerProfileCreateRequest, MailServerProfileUpdateRequest
|
|
|
|
|
|
class _Session:
|
|
def __init__(self) -> None:
|
|
self.commits = 0
|
|
self.rollbacks = 0
|
|
|
|
def add(self, _value) -> None:
|
|
return None
|
|
|
|
def commit(self) -> None:
|
|
self.commits += 1
|
|
|
|
def rollback(self) -> None:
|
|
self.rollbacks += 1
|
|
|
|
def refresh(self, _value) -> None:
|
|
return None
|
|
|
|
|
|
class MailProfileDeletionRouteTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.principal = SimpleNamespace(
|
|
tenant_id="tenant-1",
|
|
user=SimpleNamespace(id="user-1"),
|
|
api_key_id=None,
|
|
)
|
|
self.profile = SimpleNamespace(
|
|
id="profile-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
is_active=False,
|
|
)
|
|
|
|
def test_repeated_delete_does_not_emit_a_false_change(self) -> None:
|
|
session = _Session()
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
|
|
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
|
|
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
|
|
):
|
|
result = deactivate_profile("profile-1", principal=self.principal, session=session)
|
|
|
|
self.assertIs(result, self.profile)
|
|
self.assertEqual(session.commits, 1)
|
|
self.assertEqual(session.rollbacks, 0)
|
|
record_change.assert_not_called()
|
|
|
|
def test_change_feed_reports_whether_credentials_were_deleted(self) -> None:
|
|
self.profile.is_active = True
|
|
session = _Session()
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
|
|
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
|
|
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
|
|
):
|
|
deactivate_profile("profile-1", principal=self.principal, session=session)
|
|
|
|
self.assertFalse(record_change.call_args.kwargs["payload"]["credentials_deleted"])
|
|
self.assertEqual(record_change.call_args.kwargs["payload"]["deleted_credential_protocols"], [])
|
|
|
|
def test_generic_secret_deletion_failure_rolls_back(self) -> None:
|
|
self.profile.is_active = True
|
|
session = _Session()
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
|
|
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", side_effect=RuntimeError("audit unavailable")),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "audit unavailable"):
|
|
deactivate_profile("profile-1", principal=self.principal, session=session)
|
|
|
|
self.assertEqual(session.commits, 0)
|
|
self.assertEqual(session.rollbacks, 1)
|
|
|
|
def test_system_profile_changes_are_instance_wide_in_the_change_feed(self) -> None:
|
|
system_profile = SimpleNamespace(
|
|
id="profile-system",
|
|
tenant_id=None,
|
|
scope_type="system",
|
|
scope_id=None,
|
|
is_active=True,
|
|
smtp_config={"host": "smtp.example.test"},
|
|
imap_config=None,
|
|
)
|
|
session = _Session()
|
|
create_payload = MailServerProfileCreateRequest.model_validate(
|
|
{
|
|
"name": "System profile",
|
|
"scope_type": "system",
|
|
"smtp": {"host": "smtp.example.test"},
|
|
}
|
|
)
|
|
with (
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router.create_mail_server_profile", return_value=system_profile),
|
|
patch("govoplan_mail.backend.router._record_mail_change") as create_change,
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
|
|
):
|
|
create_profile(create_payload, principal=self.principal, session=session)
|
|
self.assertIsNone(create_change.call_args.kwargs["tenant_id"])
|
|
|
|
update_payload = MailServerProfileUpdateRequest(name="Renamed system profile")
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=system_profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router.update_mail_server_profile", return_value=system_profile),
|
|
patch("govoplan_mail.backend.router._record_mail_change") as update_change,
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
|
|
):
|
|
update_profile("profile-system", update_payload, principal=self.principal, session=session)
|
|
self.assertIsNone(update_change.call_args.kwargs["tenant_id"])
|
|
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=system_profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
|
|
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
|
|
patch("govoplan_mail.backend.router._record_mail_change") as delete_change,
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
|
|
):
|
|
deactivate_profile("profile-system", principal=self.principal, session=session)
|
|
self.assertIsNone(delete_change.call_args.kwargs["tenant_id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|