fix(installer): delete provider secrets before retirement
This commit is contained in:
@@ -4,12 +4,13 @@ import unittest
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from govoplan_access.backend.db import models as access_models # noqa: F401 - populate users/accounts tables
|
||||
from govoplan_calendar.backend.caldav import CalDAVClient, CalDAVError, CalDAVNotFound, CalDAVObject, CalDAVPreconditionFailed, CalDAVReportResult, CalDAVWriteResult, parse_multistatus
|
||||
from govoplan_calendar.backend.db.models import CalendarEvent, CalendarOutboxOperation, CalendarSyncCredential
|
||||
from govoplan_calendar.backend.manifest import manifest
|
||||
from govoplan_calendar.backend.outbox import dispatch_calendar_outbox
|
||||
from govoplan_calendar.backend.schemas import (
|
||||
CalendarCalDavDiscoveryRequest,
|
||||
@@ -767,6 +768,65 @@ class CalDAVSyncTests(unittest.TestCase):
|
||||
self.assertIsNotNone(credential.deleted_at)
|
||||
self.assertEqual(audit.call_count, 1)
|
||||
|
||||
def test_destructive_module_retirement_deletes_external_credentials_before_tables(self) -> None:
|
||||
class Provider:
|
||||
def __init__(self) -> None:
|
||||
self.values: dict[str, str] = {}
|
||||
self.deleted: list[str] = []
|
||||
|
||||
def store_secret(self, *, scope: str, name: str, value: str) -> str:
|
||||
reference = f"vault:{scope}:{name}"
|
||||
self.values[reference] = value
|
||||
return reference
|
||||
|
||||
def read_secret(self, secret_ref: str) -> str | None:
|
||||
return self.values.get(secret_ref)
|
||||
|
||||
def delete_secret(self, secret_ref: str) -> None:
|
||||
self.deleted.append(secret_ref)
|
||||
self.values.pop(secret_ref, None)
|
||||
|
||||
provider = Provider()
|
||||
session = self.session()
|
||||
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
|
||||
calendar = create_calendar(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarCollectionCreateRequest(name="Remote"),
|
||||
)
|
||||
with patch("govoplan_calendar.backend.service.secret_provider", return_value=provider):
|
||||
create_caldav_source(
|
||||
session,
|
||||
tenant_id="tenant-1",
|
||||
user_id=None,
|
||||
payload=CalendarCalDavSourceCreateRequest(
|
||||
calendar_id=calendar.id,
|
||||
collection_url="https://dav.example.test/module-retirement",
|
||||
auth_type="bearer",
|
||||
bearer_token="do-not-log-this",
|
||||
),
|
||||
)
|
||||
session.commit()
|
||||
credential = session.query(CalendarSyncCredential).one()
|
||||
provider_ref = str((credential.metadata_ or {})["provider_ref"])
|
||||
retirement_provider = manifest.migration_spec.retirement_provider
|
||||
assert retirement_provider is not None
|
||||
plan = retirement_provider(session, "calendar")
|
||||
assert plan.destroy_data_executor is not None
|
||||
with patch("govoplan_calendar.backend.service.audit_event") as audit:
|
||||
plan.destroy_data_executor(session, "calendar")
|
||||
|
||||
self.assertEqual(provider.deleted, [provider_ref])
|
||||
self.assertEqual(provider.values, {})
|
||||
self.assertEqual(audit.call_count, 1)
|
||||
self.assertEqual(
|
||||
audit.call_args.kwargs["details"]["deletion_reason"],
|
||||
"module_data_retired",
|
||||
)
|
||||
self.assertFalse(inspect(self.engine).has_table("calendar_sync_credentials"))
|
||||
self.assertFalse(inspect(self.engine).has_table("calendar_sync_sources"))
|
||||
|
||||
def test_create_source_retires_orphaned_source_for_deleted_calendar(self) -> None:
|
||||
session = self.session()
|
||||
session.add(Tenant(id="tenant-1", slug="tenant-1", name="Tenant"))
|
||||
|
||||
Reference in New Issue
Block a user