164 lines
4.9 KiB
Python
164 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
from govoplan_core.core.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillPage,
|
|
SearchBackfillRequest,
|
|
SearchDocument,
|
|
SearchProvider,
|
|
SearchProviderRegistration,
|
|
SearchQuery,
|
|
SearchResourceReference,
|
|
SearchResourceType,
|
|
SearchResult,
|
|
SearchSourceProvider,
|
|
SearchSourceProviderRegistration,
|
|
)
|
|
|
|
|
|
class _Provider:
|
|
def search(self, session, principal, *, query):
|
|
del session, principal
|
|
return (
|
|
SearchResult(
|
|
provider_id="test",
|
|
module_id="test",
|
|
resource_type="record",
|
|
resource_id="1",
|
|
title=query.text,
|
|
url="/test/1",
|
|
),
|
|
)
|
|
|
|
|
|
class _Source:
|
|
def resource_types(self):
|
|
return (
|
|
SearchResourceType(
|
|
provider_id="test.records",
|
|
module_id="test",
|
|
resource_type="record",
|
|
label="Records",
|
|
),
|
|
)
|
|
|
|
def backfill(self, session, *, request):
|
|
del session
|
|
return SearchBackfillPage(
|
|
documents=(),
|
|
next_cursor=None,
|
|
complete=True,
|
|
high_watermark=request.cursor,
|
|
)
|
|
|
|
def authorize(self, session, principal, *, requests):
|
|
del session, principal
|
|
return {
|
|
request.reference.key: True
|
|
for request in requests
|
|
}
|
|
|
|
|
|
class SearchContractTests(unittest.TestCase):
|
|
def test_query_and_restricted_document_are_bounded(self) -> None:
|
|
query = SearchQuery(text=" permit ", tenant_id="tenant-1", limit=20)
|
|
document = SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Permit",
|
|
url="/cases/case-1",
|
|
acl_tokens=("account:account-1",),
|
|
)
|
|
|
|
self.assertEqual("permit", query.text)
|
|
self.assertEqual("restricted", document.visibility)
|
|
with self.assertRaises(ValueError):
|
|
SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Permit",
|
|
url="/cases/case-1",
|
|
)
|
|
with self.assertRaisesRegex(ValueError, "body"):
|
|
SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Permit",
|
|
url="/cases/case-1",
|
|
body="x" * 200_001,
|
|
acl_tokens=("account:account-1",),
|
|
)
|
|
|
|
def test_manifest_provider_registration_resolves_lazily(self) -> None:
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="test",
|
|
name="Test",
|
|
version="1",
|
|
search_providers=(
|
|
SearchProviderRegistration(
|
|
id="test.records",
|
|
factory=lambda context: _Provider(),
|
|
resource_types=("record",),
|
|
),
|
|
),
|
|
search_sources=(
|
|
SearchSourceProviderRegistration(
|
|
id="test.records",
|
|
factory=lambda context: _Source(),
|
|
),
|
|
),
|
|
)
|
|
)
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=object())
|
|
)
|
|
|
|
providers = registry.search_providers()
|
|
self.assertEqual(1, len(providers))
|
|
self.assertIsInstance(providers[0][1], SearchProvider)
|
|
sources = registry.search_sources()
|
|
self.assertEqual(1, len(sources))
|
|
self.assertIsInstance(sources[0][1], SearchSourceProvider)
|
|
request = SearchAuthorizationRequest(
|
|
reference=SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="test",
|
|
resource_type="record",
|
|
resource_id="1",
|
|
),
|
|
source_revision="1",
|
|
)
|
|
self.assertTrue(
|
|
sources[0][1].authorize(
|
|
object(),
|
|
object(),
|
|
requests=(request,),
|
|
)[request.reference.key]
|
|
)
|
|
page = sources[0][1].backfill(
|
|
object(),
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id="test.records",
|
|
resource_type="record",
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
self.assertTrue(page.complete)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|