266 lines
9.2 KiB
Python
266 lines
9.2 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from dataclasses import replace
|
|
from unittest.mock import patch
|
|
|
|
from govoplan_files.backend.storage.connector_policy import ConnectorPolicySource
|
|
from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
|
|
from govoplan_files.backend.storage.connector_visibility import (
|
|
connector_profile_usable_for_import,
|
|
visible_connector_profiles_for_actor,
|
|
)
|
|
|
|
|
|
def _profile(
|
|
profile_id: str,
|
|
*,
|
|
scope_type: str = "system",
|
|
scope_id: str | None = None,
|
|
provider: str = "webdav",
|
|
enabled: bool = True,
|
|
) -> ConnectorProfile:
|
|
return ConnectorProfile(
|
|
id=profile_id,
|
|
label=profile_id,
|
|
provider=provider,
|
|
scope_type=scope_type,
|
|
scope_id=scope_id,
|
|
endpoint_url=f"https://{profile_id}.example.invalid",
|
|
enabled=enabled,
|
|
credential_mode="anonymous",
|
|
)
|
|
|
|
|
|
class ConnectorVisibilityTests(unittest.TestCase):
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources",
|
|
return_value=[],
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings"
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
|
)
|
|
def test_profiles_are_filtered_to_actor_scopes_and_database_definition_wins(
|
|
self,
|
|
database_profiles,
|
|
configured_profiles,
|
|
_policy_sources,
|
|
) -> None:
|
|
profiles = [
|
|
_profile("system"),
|
|
_profile("tenant", scope_type="tenant", scope_id="tenant-1"),
|
|
_profile("other-tenant", scope_type="tenant", scope_id="tenant-2"),
|
|
_profile("user", scope_type="user", scope_id="user-1"),
|
|
_profile("other-user", scope_type="user", scope_id="user-2"),
|
|
_profile("group", scope_type="group", scope_id="group-1"),
|
|
_profile("campaign", scope_type="campaign", scope_id="campaign-1"),
|
|
_profile("disabled", enabled=False),
|
|
_profile("duplicate", provider="webdav"),
|
|
]
|
|
database_profiles.return_value = (profiles, {profile.id for profile in profiles})
|
|
configured_profiles.return_value = [_profile("duplicate", provider="nextcloud")]
|
|
|
|
visible = visible_connector_profiles_for_actor(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
group_ids={"group-1"},
|
|
settings=object(),
|
|
campaign_visible=lambda campaign_id: campaign_id == "campaign-1",
|
|
)
|
|
|
|
self.assertEqual(
|
|
{"system", "tenant", "user", "group", "campaign", "duplicate"},
|
|
{profile.id for profile in visible},
|
|
)
|
|
duplicate = next(profile for profile in visible if profile.id == "duplicate")
|
|
self.assertEqual("webdav", duplicate.provider)
|
|
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources",
|
|
return_value=[],
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings",
|
|
return_value=[],
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
|
)
|
|
def test_campaign_and_admin_visibility_remain_explicit(
|
|
self,
|
|
database_profiles,
|
|
_configured_profiles,
|
|
_policy_sources,
|
|
) -> None:
|
|
profiles = [
|
|
_profile("campaign-a", scope_type="campaign", scope_id="campaign-a"),
|
|
_profile("campaign-b", scope_type="campaign", scope_id="campaign-b"),
|
|
_profile("other-user", scope_type="user", scope_id="user-2"),
|
|
]
|
|
database_profiles.return_value = (profiles, {profile.id for profile in profiles})
|
|
|
|
campaign_only = visible_connector_profiles_for_actor(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
group_ids=(),
|
|
settings=None,
|
|
campaign_id="campaign-a",
|
|
campaign_visible=lambda _campaign_id: True,
|
|
)
|
|
self.assertEqual(["campaign-a"], [profile.id for profile in campaign_only])
|
|
|
|
admin = visible_connector_profiles_for_actor(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
group_ids=(),
|
|
settings=None,
|
|
campaign_visible=None,
|
|
include_admin_scopes=True,
|
|
)
|
|
self.assertEqual(
|
|
{"campaign-a", "campaign-b", "other-user"},
|
|
{profile.id for profile in admin},
|
|
)
|
|
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.connector_profiles_from_settings",
|
|
return_value=[],
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.select_database_connector_profiles"
|
|
)
|
|
@patch(
|
|
"govoplan_files.backend.storage.connector_visibility.effective_connector_policy_sources"
|
|
)
|
|
def test_effective_policy_is_attached_without_mutating_profile(
|
|
self,
|
|
policy_sources,
|
|
database_profiles,
|
|
_configured_profiles,
|
|
) -> None:
|
|
profile = _profile("profile")
|
|
source = ConnectorPolicySource(
|
|
scope_type="system",
|
|
label="System",
|
|
policy={"allow": {"providers": ["webdav"]}},
|
|
)
|
|
database_profiles.return_value = ([profile], {profile.id})
|
|
policy_sources.return_value = [source]
|
|
|
|
visible = visible_connector_profiles_for_actor(
|
|
object(), # type: ignore[arg-type]
|
|
tenant_id="tenant-1",
|
|
user_id="user-1",
|
|
group_ids=(),
|
|
settings=None,
|
|
)
|
|
|
|
self.assertEqual((), profile.policy_sources)
|
|
self.assertEqual((source,), visible[0].policy_sources)
|
|
|
|
def test_import_usability_requires_safe_provider_credentials_endpoint_and_identity_policy(
|
|
self,
|
|
) -> None:
|
|
self.assertTrue(connector_profile_usable_for_import(_profile("webdav")))
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(_profile("s3", provider="s3"))
|
|
)
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(_profile("smb", provider="smb"))
|
|
)
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(
|
|
_profile("sharepoint", provider="sharepoint")
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(
|
|
ConnectorProfile(
|
|
id="no-endpoint",
|
|
label="No endpoint",
|
|
provider="webdav",
|
|
credential_mode="anonymous",
|
|
)
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(
|
|
ConnectorProfile(
|
|
id="no-credentials",
|
|
label="No credentials",
|
|
provider="webdav",
|
|
endpoint_url="https://files.example.invalid",
|
|
credential_mode="basic",
|
|
)
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
connector_profile_usable_for_import(
|
|
ConnectorProfile(
|
|
id="metadata-endpoint",
|
|
label="Metadata endpoint",
|
|
provider="webdav",
|
|
credential_mode="anonymous",
|
|
metadata={"webdav_endpoint_url": "https://files.example.invalid"},
|
|
)
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
connector_profile_usable_for_import(
|
|
ConnectorProfile(
|
|
id="unresolved-secret-reference",
|
|
label="Unresolved secret reference",
|
|
provider="webdav",
|
|
endpoint_url="https://files.example.invalid",
|
|
credential_mode="basic",
|
|
secret_ref="vault://files/webdav",
|
|
)
|
|
)
|
|
)
|
|
|
|
denied = replace(
|
|
_profile("denied"),
|
|
policy_sources=(
|
|
ConnectorPolicySource(
|
|
scope_type="system",
|
|
label="System",
|
|
policy={"deny": {"providers": ["webdav"]}},
|
|
),
|
|
),
|
|
)
|
|
self.assertFalse(connector_profile_usable_for_import(denied))
|
|
|
|
path_limited = replace(
|
|
_profile("path-limited"),
|
|
policy_sources=(
|
|
ConnectorPolicySource(
|
|
scope_type="system",
|
|
label="System",
|
|
policy={"allow": {"external_paths": ["/approved/*"]}},
|
|
),
|
|
),
|
|
)
|
|
self.assertFalse(connector_profile_usable_for_import(path_limited))
|
|
|
|
endpoint_allowed = replace(
|
|
_profile("endpoint-allowed"),
|
|
policy_sources=(
|
|
ConnectorPolicySource(
|
|
scope_type="system",
|
|
label="System",
|
|
policy={"allow": {"external_urls": ["https://*.example.invalid"]}},
|
|
),
|
|
),
|
|
)
|
|
self.assertTrue(connector_profile_usable_for_import(endpoint_allowed))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|