from __future__ import annotations import unittest from types import SimpleNamespace from unittest.mock import patch from sqlalchemy import create_engine, inspect from sqlalchemy.orm import sessionmaker from govoplan_access.backend.db import models as access_models # noqa: F401 - resolve Files user foreign keys from govoplan_core.core.change_sequence import ChangeSequenceEntry from govoplan_core.db.base import Base from govoplan_core.security.secrets import encrypt_secret from govoplan_files.backend.db.models import FileConnectorCredential, FileConnectorProfile from govoplan_files.backend.router import deactivate_connector_credential, deactivate_connector_profile class Principal: tenant_id = "tenant-1" user = SimpleNamespace(id="user-1") api_key = None def has(self, scope: str) -> bool: return scope == "files:file:admin" class ConnectorCredentialDeletionTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ FileConnectorCredential.__table__, FileConnectorProfile.__table__, ChangeSequenceEntry.__table__, ], ) self.session = sessionmaker(bind=self.engine)() self.principal = Principal() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def test_delete_credential_scrubs_material_audits_and_disables_dependents(self) -> None: credential = self._credential() profile = self._profile(credential_profile_id=credential.id) self.session.add_all([credential, profile]) self.session.commit() with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event" ) as audit: response = deactivate_connector_credential( credential.id, session=self.session, principal=self.principal, # type: ignore[arg-type] ) self.assertFalse(response.enabled) self.assertFalse(response.credentials_configured) self.assertIsNone(response.credential_secret_source) self.session.refresh(credential) self.session.refresh(profile) self._assert_credential_scrubbed(credential) self._assert_profile_scrubbed(profile) calls = {call.kwargs["action"]: call.kwargs for call in audit.call_args_list} self.assertEqual( {"files.connector_profile_deleted", "files.connector_credential_deleted"}, set(calls), ) credential_audit = calls["files.connector_credential_deleted"] self.assertEqual("encrypted_database", credential_audit["details"]["storage_backend"]) self.assertEqual(["password", "token"], credential_audit["details"]["deleted_secret_kinds"]) self.assertEqual(1, credential_audit["details"]["affected_profile_count"]) self.assertNotIn("do-not-audit", repr(audit.call_args_list)) changes = self.session.query(ChangeSequenceEntry).order_by(ChangeSequenceEntry.id.asc()).all() self.assertEqual([credential.id, profile.id], [change.resource_id for change in changes]) self.assertEqual(["deleted", "updated"], [change.operation for change in changes]) def test_delete_profile_scrubs_inline_credentials_and_audits(self) -> None: profile = self._profile(credential_profile_id="shared-credential") self.session.add(profile) self.session.commit() with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event" ) as audit: response = deactivate_connector_profile( profile.id, session=self.session, principal=self.principal, # type: ignore[arg-type] ) self.assertFalse(response.enabled) self.assertFalse(response.credentials_configured) self.assertIsNone(response.credential_profile_id) self.session.refresh(profile) self._assert_profile_scrubbed(profile) audit.assert_called_once() call = audit.call_args.kwargs self.assertEqual("files.connector_profile_deleted", call["action"]) self.assertEqual("api_delete", call["details"]["deletion_reason"]) self.assertIn("credential_profile", call["details"]["removed_reference_kinds"]) self.assertNotIn("do-not-audit", repr(call)) def test_delete_rolls_back_when_audit_fails(self) -> None: credential = self._credential() self.session.add(credential) self.session.commit() original_password = credential.password_encrypted original_token = credential.token_encrypted with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event", side_effect=RuntimeError("audit unavailable"), ), self.assertRaisesRegex(RuntimeError, "audit unavailable"): deactivate_connector_credential( credential.id, session=self.session, principal=self.principal, # type: ignore[arg-type] ) persisted = self.session.get(FileConnectorCredential, credential.id) assert persisted is not None self.assertTrue(persisted.enabled) self.assertEqual(original_password, persisted.password_encrypted) self.assertEqual(original_token, persisted.token_encrypted) self.assertEqual(0, self.session.query(ChangeSequenceEntry).count()) def test_delete_detaches_unowned_legacy_secret_reference_without_claiming_provider_deletion(self) -> None: credential = self._credential(secret_ref="vault:tenant-1:files:credential") self.session.add(credential) self.session.commit() with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event" ) as audit: response = deactivate_connector_credential( credential.id, session=self.session, principal=self.principal, # type: ignore[arg-type] ) self.assertFalse(response.enabled) persisted = self.session.get(FileConnectorCredential, credential.id) assert persisted is not None self._assert_credential_scrubbed(persisted) audit.assert_called_once() details = audit.call_args.kwargs["details"] self.assertEqual("unowned_external_reference_detached", details["storage_backend"]) self.assertIn("unowned_external_secret_ref", details["removed_reference_kinds"]) self.assertNotIn("vault:tenant-1:files:credential", repr(audit.call_args)) def test_retirement_scrubs_and_audits_before_tables_are_dropped(self) -> None: from govoplan_files.backend.manifest import manifest self.session.add_all([self._credential(), self._profile()]) self.session.commit() retirement_provider = manifest.migration_spec.retirement_provider assert retirement_provider is not None plan = retirement_provider(self.session, "files") assert plan.destroy_data_executor is not None with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event" ) as audit: plan.destroy_data_executor(self.session, "files") self.assertEqual(2, audit.call_count) self.assertEqual( {"module_data_retired"}, {call.kwargs["details"]["deletion_reason"] for call in audit.call_args_list}, ) self.assertNotIn("do-not-audit", repr(audit.call_args_list)) self.assertFalse(inspect(self.engine).has_table(FileConnectorCredential.__tablename__)) self.assertFalse(inspect(self.engine).has_table(FileConnectorProfile.__tablename__)) def test_retirement_detaches_unowned_external_reference_before_drop(self) -> None: from govoplan_files.backend.manifest import manifest self.session.add(self._credential(secret_ref="vault:tenant-1:files:credential")) self.session.commit() retirement_provider = manifest.migration_spec.retirement_provider assert retirement_provider is not None plan = retirement_provider(self.session, "files") assert plan.destroy_data_executor is not None with patch( "govoplan_files.backend.storage.connector_credential_deletion.audit_event" ) as audit: plan.destroy_data_executor(self.session, "files") audit.assert_called_once() details = audit.call_args.kwargs["details"] self.assertIn("unowned_external_secret_ref", details["removed_reference_kinds"]) self.assertNotIn("vault:tenant-1:files:credential", repr(audit.call_args)) self.assertFalse(inspect(self.engine).has_table(FileConnectorCredential.__tablename__)) self.assertFalse(inspect(self.engine).has_table(FileConnectorProfile.__tablename__)) @staticmethod def _credential(*, secret_ref: str | None = None) -> FileConnectorCredential: return FileConnectorCredential( id="credential-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", label="Credential", provider="webdav", enabled=True, credential_mode="basic", username="connector-user", password_encrypted=encrypt_secret("do-not-audit-password"), token_encrypted=encrypt_secret("do-not-audit-token"), password_env="DEPLOYMENT_PASSWORD", token_env="DEPLOYMENT_TOKEN", secret_ref=secret_ref, policy={}, metadata_={"private_hint": "do-not-audit-metadata"}, ) @staticmethod def _profile(*, credential_profile_id: str | None = None) -> FileConnectorProfile: return FileConnectorProfile( id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", label="Profile", provider="webdav", endpoint_url="https://dav.example.test", enabled=True, credential_profile_id=credential_profile_id, credential_mode="basic", username="connector-user", password_encrypted=encrypt_secret("do-not-audit-profile-password"), token_encrypted=encrypt_secret("do-not-audit-profile-token"), password_env="DEPLOYMENT_PROFILE_PASSWORD", token_env="DEPLOYMENT_PROFILE_TOKEN", policy={}, metadata_={"private_hint": "do-not-audit-profile-metadata"}, ) def _assert_credential_scrubbed(self, row: FileConnectorCredential) -> None: self.assertFalse(row.enabled) self.assertEqual("none", row.credential_mode) self.assertIsNone(row.username) self.assertIsNone(row.password_encrypted) self.assertIsNone(row.token_encrypted) self.assertIsNone(row.password_env) self.assertIsNone(row.token_env) self.assertIsNone(row.secret_ref) self.assertEqual({}, row.metadata_) def _assert_profile_scrubbed(self, row: FileConnectorProfile) -> None: self.assertFalse(row.enabled) self.assertIsNone(row.credential_profile_id) self.assertEqual("none", row.credential_mode) self.assertIsNone(row.username) self.assertIsNone(row.password_encrypted) self.assertIsNone(row.token_encrypted) self.assertIsNone(row.password_env) self.assertIsNone(row.token_env) self.assertIsNone(row.secret_ref) self.assertEqual({}, row.metadata_) if __name__ == "__main__": unittest.main()