Refactor connector profile updates

tests/test_connector_profile_store.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+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.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 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__])
+        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))
+
+
+if __name__ == "__main__":
+    unittest.main()