180 lines
5.1 KiB
Python
180 lines
5.1 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.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_dataflow.backend.db.models import DataflowPipeline
|
|
from govoplan_dataflow.backend.manifest import (
|
|
ADMIN_SCOPE,
|
|
READ_SCOPE,
|
|
get_manifest,
|
|
)
|
|
from govoplan_dataflow.backend.search_source import (
|
|
DataflowSearchSource,
|
|
PROVIDER_ID,
|
|
RESOURCE_TYPE,
|
|
)
|
|
|
|
|
|
class DataflowSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(DataflowPipeline.__table__,),
|
|
)
|
|
self.session = Session(self.engine)
|
|
self.session.add_all(
|
|
(
|
|
_pipeline(
|
|
"00000000-0000-0000-0000-000000000001",
|
|
"tenant-1",
|
|
"Monthly reconciliation",
|
|
),
|
|
_pipeline(
|
|
"00000000-0000-0000-0000-000000000002",
|
|
"tenant-1",
|
|
"Sanctions screening",
|
|
),
|
|
_pipeline(
|
|
"00000000-0000-0000-0000-000000000003",
|
|
"tenant-2",
|
|
"Other tenant",
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
self.source = DataflowSearchSource(registry=None)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_manifest_announces_optional_search_source(self) -> None:
|
|
manifest = get_manifest()
|
|
|
|
self.assertIn("search", manifest.optional_dependencies)
|
|
self.assertEqual(
|
|
(PROVIDER_ID,),
|
|
tuple(item.id for item in manifest.search_sources),
|
|
)
|
|
|
|
def test_backfill_is_tenant_bounded_and_stably_paged(self) -> None:
|
|
first = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type=RESOURCE_TYPE,
|
|
rebuild_id="rebuild-1",
|
|
limit=1,
|
|
),
|
|
)
|
|
second = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type=RESOURCE_TYPE,
|
|
rebuild_id="rebuild-1",
|
|
cursor=first.next_cursor,
|
|
limit=1,
|
|
),
|
|
)
|
|
|
|
self.assertFalse(first.complete)
|
|
self.assertTrue(second.complete)
|
|
self.assertEqual(
|
|
{"Monthly reconciliation", "Sanctions screening"},
|
|
{
|
|
first.documents[0].title,
|
|
second.documents[0].title,
|
|
},
|
|
)
|
|
self.assertTrue(
|
|
first.documents[0].url.startswith(
|
|
"/dataflow?pipelineId="
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
first.documents[0].requires_authorization_recheck
|
|
)
|
|
|
|
def test_authorization_rechecks_tenant_scope_and_permission(self) -> None:
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="dataflow",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="00000000-0000-0000-0000-000000000001",
|
|
)
|
|
request = SearchAuthorizationRequest(
|
|
reference=reference,
|
|
source_revision="1",
|
|
)
|
|
|
|
allowed = self.source.authorize(
|
|
self.session,
|
|
_principal({READ_SCOPE}),
|
|
requests=(request,),
|
|
)
|
|
denied = self.source.authorize(
|
|
self.session,
|
|
_principal(set()),
|
|
requests=(request,),
|
|
)
|
|
administrative = self.source.authorize(
|
|
self.session,
|
|
_principal({ADMIN_SCOPE}),
|
|
requests=(request,),
|
|
)
|
|
|
|
self.assertTrue(allowed[reference.key])
|
|
self.assertFalse(denied[reference.key])
|
|
self.assertTrue(administrative[reference.key])
|
|
|
|
|
|
def _pipeline(
|
|
pipeline_id: str,
|
|
tenant_id: str,
|
|
name: str,
|
|
) -> DataflowPipeline:
|
|
return DataflowPipeline(
|
|
id=pipeline_id,
|
|
tenant_id=tenant_id,
|
|
scope_type="tenant",
|
|
scope_id=tenant_id,
|
|
definition_kind="flow",
|
|
name=name,
|
|
description=f"{name} description",
|
|
status="active",
|
|
current_revision=1,
|
|
)
|
|
|
|
|
|
def _principal(scopes: set[str]) -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id="account-1"),
|
|
user=SimpleNamespace(id="membership-1"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|