144 lines
4.8 KiB
Python
144 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
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_forms_runtime.backend.db.models import (
|
|
FormInstanceIdentity,
|
|
FormInstanceRevision,
|
|
)
|
|
from govoplan_forms_runtime.backend.search_source import (
|
|
FormsRuntimeSearchSource,
|
|
PROVIDER_ID,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 8, 6, 12, 0, tzinfo=UTC)
|
|
|
|
|
|
class FormsRuntimeSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
FormInstanceIdentity.__table__.create(self.engine)
|
|
FormInstanceRevision.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
identity = FormInstanceIdentity(
|
|
id="identity-row-1",
|
|
tenant_id="tenant-1",
|
|
instance_id="instance-1",
|
|
definition_id="permit-form",
|
|
definition_revision="4",
|
|
created_by="account-1",
|
|
)
|
|
revision = FormInstanceRevision(
|
|
tenant_id="tenant-1",
|
|
instance_id="instance-1",
|
|
identity_id="identity-row-1",
|
|
revision=3,
|
|
status="submitted",
|
|
recorded_at=NOW,
|
|
changed_by="account-1",
|
|
snapshot={
|
|
"definition_ref": {
|
|
"object_id": "permit-form",
|
|
"version": "4",
|
|
"label": "Permit application",
|
|
},
|
|
"values": {"protected_field": "PROTECTED-VALUE"},
|
|
"receipt_id": "receipt-1",
|
|
"service_ref": {"object_id": "permit", "version": "2"},
|
|
},
|
|
)
|
|
self.session.add_all((identity, revision))
|
|
self.session.commit()
|
|
self.source = FormsRuntimeSearchSource()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_search_projection_excludes_submitted_values_and_rechecks_access(self) -> None:
|
|
page = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type="form_submission",
|
|
rebuild_id="rebuild-1",
|
|
),
|
|
)
|
|
document = page.documents[0]
|
|
self.assertNotIn("PROTECTED-VALUE", repr(document))
|
|
self.assertFalse(document.metadata["protected_values_indexed"])
|
|
self.assertEqual("Permit application", document.title)
|
|
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="forms_runtime",
|
|
resource_type="form_submission",
|
|
resource_id="instance-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="3")
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({"forms_runtime:submission:participate"}),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({"forms_runtime:workspace:read"}, account_id="other"),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({"forms_runtime:submission:participate"}, account_id="other"),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
|
|
changes = self.source.index_changes_for_event(
|
|
self.session,
|
|
event=PlatformEvent(
|
|
type="forms_runtime.instance.submitted",
|
|
module_id="forms_runtime",
|
|
tenant=EventTenantRef(id="tenant-1"),
|
|
resource=EventObjectRef(type="form_submission", id="instance-1"),
|
|
),
|
|
delivery_key="delivery-1",
|
|
)
|
|
self.assertEqual("upsert", changes[0].kind)
|
|
|
|
|
|
def _principal(scopes: set[str], *, account_id: str = "account-1") -> ApiPrincipal:
|
|
return ApiPrincipal(
|
|
principal=PrincipalRef(
|
|
account_id=account_id,
|
|
membership_id=f"membership-{account_id}",
|
|
tenant_id="tenant-1",
|
|
scopes=frozenset(scopes),
|
|
),
|
|
account=SimpleNamespace(id=account_id),
|
|
user=SimpleNamespace(id=f"membership-{account_id}"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|