201 lines
9.2 KiB
Python
201 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from fastapi import HTTPException
|
|
|
|
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,
|
|
has=lambda _scope: False,
|
|
)
|
|
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") as require_credentials,
|
|
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
|
|
patch("govoplan_mail.backend.router.clear_mailbox_index") as clear_index,
|
|
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)
|
|
clear_index.assert_called_once_with(session, profile_id="profile-1")
|
|
record_change.assert_not_called()
|
|
require_credentials.assert_not_called()
|
|
|
|
def test_delete_requires_secret_authority_when_credentials_will_be_deleted(self) -> None:
|
|
self.profile.is_active = True
|
|
self.profile.smtp_password_encrypted = "existing-ciphertext"
|
|
self.profile.imap_password_encrypted = 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") as require_credentials,
|
|
patch(
|
|
"govoplan_mail.backend.router.delete_mail_profile_credentials",
|
|
return_value=("smtp",),
|
|
),
|
|
patch("govoplan_mail.backend.router.clear_mailbox_index"),
|
|
patch("govoplan_mail.backend.router._record_mail_change"),
|
|
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
|
|
):
|
|
deactivate_profile("profile-1", principal=self.principal, session=session)
|
|
|
|
require_credentials.assert_called_once_with(
|
|
self.principal,
|
|
"tenant",
|
|
"tenant-1",
|
|
)
|
|
|
|
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.clear_mailbox_index") as clear_index,
|
|
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"], [])
|
|
clear_index.assert_called_once_with(session, profile_id="profile-1")
|
|
|
|
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_patch_deactivation_is_rejected_in_favor_of_audited_delete(self) -> None:
|
|
self.profile.is_active = True
|
|
self.profile.smtp_password_encrypted = "existing-ciphertext"
|
|
session = _Session()
|
|
payload = MailServerProfileUpdateRequest(is_active=False)
|
|
with (
|
|
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
|
|
patch("govoplan_mail.backend.router._require_profile_write_scope"),
|
|
self.assertRaises(HTTPException) as captured,
|
|
):
|
|
update_profile(
|
|
self.profile.id,
|
|
payload,
|
|
principal=self.principal,
|
|
session=session,
|
|
)
|
|
|
|
self.assertEqual(captured.exception.status_code, 422)
|
|
self.assertTrue(self.profile.is_active)
|
|
self.assertEqual(self.profile.smtp_password_encrypted, "existing-ciphertext")
|
|
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.clear_mailbox_index"),
|
|
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()
|