feat: add search and external integration contracts

This commit is contained in:
2026-07-29 15:50:08 +02:00
parent 68328f3d8e
commit d9b5708df0
14 changed files with 607 additions and 3 deletions

View File

@@ -0,0 +1,36 @@
from __future__ import annotations
import unittest
from govoplan_core.core.external_references import (
ExternalObjectReference,
ExternalReferenceValidationError,
)
class ExternalReferenceContractTests(unittest.TestCase):
def test_reference_has_stable_identity_and_ordered_maturity(self) -> None:
reference = ExternalObjectReference(
system="openproject-main",
object_type="work_package",
object_id="42",
maturity="synchronize",
canonical_url="https://projects.example.test/work_packages/42",
)
self.assertEqual("openproject-main:work_package:42", reference.identity_key)
self.assertTrue(reference.supports("read"))
self.assertFalse(reference.supports("replace"))
def test_reference_rejects_credentials_in_urls(self) -> None:
with self.assertRaises(ExternalReferenceValidationError):
ExternalObjectReference(
system="wiki",
object_type="page",
object_id="Main_Page",
canonical_url="https://user:secret@example.test/wiki/Main_Page",
)
if __name__ == "__main__":
unittest.main()

View File

@@ -0,0 +1,82 @@
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 (
SearchDocument,
SearchProvider,
SearchProviderRegistration,
SearchQuery,
SearchResult,
)
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 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",
)
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",),
),
),
)
)
registry.configure_capability_context(
ModuleContext(registry=registry, settings=object())
)
providers = registry.search_providers()
self.assertEqual(1, len(providers))
self.assertIsInstance(providers[0][1], SearchProvider)
if __name__ == "__main__":
unittest.main()