119 lines
3.7 KiB
Python
119 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_idm.backend.db.models import IdmTypedGroup
|
|
from govoplan_idm.backend.search_source import IdmSearchSource, PROVIDER_ID
|
|
|
|
|
|
class IdmSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(IdmTypedGroup.__table__,),
|
|
)
|
|
self.session = Session(self.engine)
|
|
self.session.add_all(
|
|
(
|
|
IdmTypedGroup(
|
|
id="group-1",
|
|
tenant_id="tenant-1",
|
|
key="permit-holder",
|
|
name="Permit holders",
|
|
group_type="business_status",
|
|
),
|
|
IdmTypedGroup(
|
|
id="group-other",
|
|
tenant_id="tenant-2",
|
|
key="other",
|
|
name="Other tenant",
|
|
group_type="business_status",
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
self.source = IdmSearchSource()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_typed_group_search_is_tenant_and_scope_bounded(self) -> None:
|
|
page = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type="typed_group",
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(("group-1",), tuple(doc.resource_id for doc in page.documents))
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="idm",
|
|
resource_type="typed_group",
|
|
resource_id="group-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({"idm:relationship:read"}),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal(set()),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
|
|
def test_typed_group_event_is_translated_to_an_upsert(self) -> None:
|
|
event = PlatformEvent(
|
|
type="idm.typed_group.updated",
|
|
module_id="idm",
|
|
tenant=EventTenantRef(id="tenant-1"),
|
|
resource=EventObjectRef(type="typed_group", id="group-1"),
|
|
)
|
|
change = self.source.index_changes_for_event(
|
|
self.session,
|
|
event=event,
|
|
delivery_key="delivery-1",
|
|
)[0]
|
|
self.assertEqual("upsert", change.kind)
|
|
self.assertEqual("group-1", change.reference.resource_id)
|
|
|
|
|
|
def _principal(scopes: set[str]) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="user-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="user-1"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|