102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, patch
|
|
|
|
from govoplan_files.backend.storage.common import FileStorageError
|
|
from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
|
|
from govoplan_files.backend.storage.connector_spaces import (
|
|
soft_delete_connector_space,
|
|
validate_connector_space_write_mode,
|
|
)
|
|
|
|
|
|
class ConnectorSpaceDeletionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.session = Mock()
|
|
self.space = SimpleNamespace(
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_user_id="owner-1",
|
|
owner_group_id=None,
|
|
deleted_at=None,
|
|
is_active=True,
|
|
)
|
|
|
|
@patch("govoplan_files.backend.storage.connector_spaces.utcnow")
|
|
@patch("govoplan_files.backend.storage.connector_spaces.ensure_owner_access")
|
|
def test_owner_can_soft_delete_local_space_definition(
|
|
self, ensure_owner_access: Mock, utcnow: Mock
|
|
) -> None:
|
|
deleted_at = object()
|
|
utcnow.return_value = deleted_at
|
|
|
|
result = soft_delete_connector_space(
|
|
self.session, self.space, user_id="owner-1"
|
|
)
|
|
|
|
self.assertIs(result, self.space)
|
|
self.assertIs(deleted_at, self.space.deleted_at)
|
|
self.assertFalse(self.space.is_active)
|
|
ensure_owner_access.assert_called_once_with(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
owner_type="user",
|
|
owner_id="owner-1",
|
|
user_id="owner-1",
|
|
is_admin=False,
|
|
)
|
|
self.session.add.assert_called_once_with(self.space)
|
|
self.session.flush.assert_called_once_with()
|
|
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_spaces.ensure_owner_access",
|
|
side_effect=FileStorageError("File space access denied"),
|
|
)
|
|
def test_inaccessible_space_is_blocked_without_mutation(
|
|
self, _ensure_owner_access: Mock
|
|
) -> None:
|
|
with self.assertRaisesRegex(FileStorageError, "access denied"):
|
|
soft_delete_connector_space(
|
|
self.session, self.space, user_id="other-user"
|
|
)
|
|
|
|
self.assertIsNone(self.space.deleted_at)
|
|
self.assertTrue(self.space.is_active)
|
|
self.session.add.assert_not_called()
|
|
self.session.flush.assert_not_called()
|
|
|
|
def test_read_only_mode_does_not_require_remote_write_capability(self) -> None:
|
|
profile = ConnectorProfile(
|
|
id="webdav-read",
|
|
label="Read only",
|
|
provider="webdav",
|
|
capabilities=("browse", "import"),
|
|
)
|
|
|
|
validate_connector_space_write_mode(profile, read_only=True)
|
|
|
|
def test_two_way_mode_requires_explicit_supported_write_capability(self) -> None:
|
|
missing_capability = ConnectorProfile(
|
|
id="s3-read",
|
|
label="S3 read only",
|
|
provider="s3",
|
|
capabilities=("browse",),
|
|
)
|
|
unsupported_provider = ConnectorProfile(
|
|
id="webdav-write",
|
|
label="WebDAV write",
|
|
provider="webdav",
|
|
capabilities=("write",),
|
|
)
|
|
|
|
for profile in (missing_capability, unsupported_provider):
|
|
with self.subTest(profile=profile.id):
|
|
with self.assertRaisesRegex(FileStorageError, "requires an S3"):
|
|
validate_connector_space_write_mode(profile, read_only=False)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|