from __future__ import annotations import json import unittest from unittest.mock import patch from sqlalchemy import create_engine 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.db.base import Base from govoplan_core.security.credential_envelopes import CredentialEnvelope from govoplan_core.security.secrets import decrypt_secret, encrypt_secret from govoplan_files.backend.db.models import FileConnectorProfile from govoplan_files.backend.storage.common import FileStorageError from govoplan_files.backend.storage.connector_profile_store import ( list_database_connector_profiles, update_connector_profile_row, ) class ConnectorProfileStoreUpdateTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ FileConnectorProfile.__table__, CredentialEnvelope.__table__, ], ) self.session = sessionmaker(bind=self.engine)() self.row = FileConnectorProfile( id="profile-1", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", label="Original profile", provider="webdav", endpoint_url="https://dav.example.test/original", base_path="original", enabled=True, credential_mode="basic", username="original-user", password_encrypted=encrypt_secret("original-password"), token_encrypted=encrypt_secret("original-token"), capabilities=["browse"], policy={"allow": {"providers": ["webdav"]}}, metadata_={"original": True}, created_by_user_id=None, updated_by_user_id=None, ) self.session.add(self.row) self.session.commit() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all(bind=self.engine) self.engine.dispose() def test_update_preserves_normalization_and_flush_only_transaction_boundary(self) -> None: updated = update_connector_profile_row( self.session, self.row, user_id="user-2", label=" Updated profile ", provider="NEXTCLOUD", endpoint_url=" https://cloud.example.test/dav ", base_path=" shared/reports ", enabled=False, credential_profile_id=" credential-2 ", credential_mode="TOKEN", username=" updated-user ", password="updated-password", token="updated-token", capabilities=["browse", " import ", ""], policy={"deny": {"external_paths": ["private"]}}, metadata={"department": "reports"}, ) self.assertIs(self.row, updated) self.assertEqual("Updated profile", updated.label) self.assertEqual("nextcloud", updated.provider) self.assertEqual("https://cloud.example.test/dav", updated.endpoint_url) self.assertEqual("shared/reports", updated.base_path) self.assertFalse(updated.enabled) self.assertEqual("credential-2", updated.credential_profile_id) self.assertEqual("token", updated.credential_mode) self.assertEqual("updated-user", updated.username) self.assertEqual("updated-password", decrypt_secret(updated.password_encrypted)) self.assertEqual("updated-token", decrypt_secret(updated.token_encrypted)) self.assertEqual(["browse", "import"], updated.capabilities) self.assertEqual({"deny": {"external_paths": ["private"]}}, updated.policy) self.assertEqual({"department": "reports"}, updated.metadata_) self.assertEqual("user-2", updated.updated_by_user_id) self.session.rollback() persisted = self.session.get(FileConnectorProfile, self.row.id) assert persisted is not None self.assertEqual("Original profile", persisted.label) self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted)) self.assertTrue(persisted.enabled) def test_secret_replacement_wins_over_clear_while_clear_removes_an_omitted_token(self) -> None: update_connector_profile_row( self.session, self.row, user_id="user-2", password="replacement-password", clear_password=True, clear_token=True, ) self.assertEqual("replacement-password", decrypt_secret(self.row.password_encrypted)) self.assertIsNone(self.row.token_encrypted) self.assertEqual("https://dav.example.test/original", self.row.endpoint_url) self.assertEqual("original-user", self.row.username) def test_legacy_external_secret_reference_cannot_be_cleared_or_partially_mutate_the_row(self) -> None: self.row.secret_ref = "vault:tenant-1:files:profile" self.session.commit() with self.assertRaisesRegex(FileStorageError, "provider-side deletion"): update_connector_profile_row( self.session, self.row, user_id="user-2", label="Must not be applied", secret_ref="", ) self.assertEqual("Original profile", self.row.label) self.assertEqual("vault:tenant-1:files:profile", self.row.secret_ref) def test_flush_failure_remains_rollback_safe(self) -> None: with patch.object(self.session, "flush", side_effect=RuntimeError("database unavailable")), self.assertRaisesRegex( RuntimeError, "database unavailable", ): update_connector_profile_row( self.session, self.row, user_id="user-2", label="Uncommitted profile", password="uncommitted-password", ) self.session.rollback() persisted = self.session.get(FileConnectorProfile, self.row.id) assert persisted is not None self.assertEqual("Original profile", persisted.label) self.assertEqual("original-password", decrypt_secret(persisted.password_encrypted)) def test_visibility_filter_runs_before_connector_secrets_are_decrypted(self) -> None: invisible = FileConnectorProfile( id="invisible-profile", tenant_id="tenant-1", scope_type="user", scope_id="another-user", label="Invisible profile", provider="webdav", endpoint_url="https://invisible.example.test", enabled=True, credential_mode="basic", password_encrypted="not-a-valid-encrypted-secret", capabilities=["browse"], policy={}, metadata_={}, ) self.session.add(invisible) self.session.commit() profiles = list_database_connector_profiles( self.session, tenant_id="tenant-1", row_visible=lambda row: row.id == self.row.id, ) self.assertEqual([self.row.id], [profile.id for profile in profiles]) self.assertEqual("original-password", profiles[0].password_value) def test_reusable_credential_resolves_and_revocation_keeps_profile_visible(self) -> None: credential = CredentialEnvelope( id="shared-credential", tenant_id="tenant-1", scope_type="tenant", scope_id="tenant-1", name="Shared WebDAV login", credential_kind="username_password", public_data={"username": "ada"}, secret_data_encrypted=encrypt_secret(json.dumps({"password": "secret"})), secret_keys=["password"], allowed_modules=["files"], allowed_server_refs=[], inherit_to_lower_scopes=True, is_active=True, revision="revision-1", ) self.row.credential_profile_id = "credential-envelope:shared-credential" self.row.username = None self.row.password_encrypted = None self.row.token_encrypted = None self.session.add(credential) self.session.commit() profile = list_database_connector_profiles( self.session, tenant_id="tenant-1", )[0] self.assertEqual("ada", profile.username) self.assertEqual("secret", profile.password_value) self.assertTrue(profile.credentials_configured) credential.is_active = False self.session.commit() profile = list_database_connector_profiles( self.session, tenant_id="tenant-1", )[0] self.assertIsNone(profile.password_value) self.assertFalse(profile.credentials_configured) self.assertTrue(profile.metadata["credential_unavailable"]) if __name__ == "__main__": unittest.main()