146 lines
6.0 KiB
Python
146 lines
6.0 KiB
Python
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()
|