255 lines
8.3 KiB
Python
255 lines
8.3 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.search import (
|
|
SearchAuthorizationRequest,
|
|
SearchBackfillRequest,
|
|
SearchResourceReference,
|
|
)
|
|
from govoplan_records.backend.db.models import (
|
|
RecordAccessGrantRevision,
|
|
RecordIdentity,
|
|
RecordRevision,
|
|
)
|
|
from govoplan_records.backend.search_source import (
|
|
ADMIN_SCOPE,
|
|
PROVIDER_ID,
|
|
READ_SCOPE,
|
|
RESOURCE_TYPE,
|
|
RecordsSearchSource,
|
|
)
|
|
|
|
|
|
NOW = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
|
|
|
|
|
|
class RecordsSearchSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
RecordIdentity.__table__.create(self.engine)
|
|
RecordRevision.__table__.create(self.engine)
|
|
RecordAccessGrantRevision.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
identity = RecordIdentity(
|
|
id="identity-1",
|
|
tenant_id="tenant-1",
|
|
record_id="record-1",
|
|
record_number="EA-2026-0001",
|
|
)
|
|
self.session.add_all(
|
|
(
|
|
identity,
|
|
RecordRevision(
|
|
tenant_id="tenant-1",
|
|
record_id="record-1",
|
|
identity_id=identity.id,
|
|
revision=1,
|
|
class_id="class-1",
|
|
file_plan_node_id="plan-1",
|
|
title="Permit decision",
|
|
description="Decision record",
|
|
state="open",
|
|
source_authority_mode="native_authoritative",
|
|
access_mode="tenant",
|
|
purpose="case-work",
|
|
institutional_context={"organization_unit_id": "unit-1"},
|
|
external_reference={},
|
|
search_text="ea-2026-0001 permit decision",
|
|
valid_from=NOW,
|
|
recorded_at=NOW,
|
|
snapshot={},
|
|
),
|
|
RecordIdentity(
|
|
id="identity-2",
|
|
tenant_id="tenant-2",
|
|
record_id="record-2",
|
|
record_number="EA-OTHER",
|
|
),
|
|
)
|
|
)
|
|
self.session.flush()
|
|
self.session.add(
|
|
RecordRevision(
|
|
tenant_id="tenant-2",
|
|
record_id="record-2",
|
|
identity_id="identity-2",
|
|
revision=1,
|
|
class_id="class-1",
|
|
file_plan_node_id="plan-1",
|
|
title="Other tenant",
|
|
state="open",
|
|
source_authority_mode="native_authoritative",
|
|
access_mode="tenant",
|
|
purpose="case-work",
|
|
institutional_context={},
|
|
external_reference={},
|
|
search_text="other tenant",
|
|
valid_from=NOW,
|
|
recorded_at=NOW,
|
|
snapshot={},
|
|
)
|
|
)
|
|
self.session.commit()
|
|
self.source = RecordsSearchSource()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_backfill_and_live_authorization_are_tenant_scoped(self) -> None:
|
|
page = self.source.backfill(
|
|
self.session,
|
|
request=SearchBackfillRequest(
|
|
tenant_id="tenant-1",
|
|
provider_id=PROVIDER_ID,
|
|
resource_type=RESOURCE_TYPE,
|
|
rebuild_id="records-rebuild-1",
|
|
),
|
|
)
|
|
self.assertEqual(
|
|
("record-1",), tuple(document.resource_id for document in page.documents)
|
|
)
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="records",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="record-1",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({READ_SCOPE}),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({ADMIN_SCOPE}),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal(set()),
|
|
requests=(request,),
|
|
)[reference.key]
|
|
)
|
|
other_tenant_reference = SearchResourceReference(
|
|
tenant_id="tenant-2",
|
|
module_id="records",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="record-2",
|
|
)
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session,
|
|
_principal({READ_SCOPE}),
|
|
requests=(
|
|
SearchAuthorizationRequest(
|
|
reference=other_tenant_reference,
|
|
source_revision="1",
|
|
),
|
|
),
|
|
)[other_tenant_reference.key]
|
|
)
|
|
|
|
def test_restricted_search_requires_an_explicit_search_purpose_grant(self) -> None:
|
|
self.session.add_all(
|
|
(
|
|
RecordIdentity(
|
|
id="identity-restricted",
|
|
tenant_id="tenant-1",
|
|
record_id="record-restricted",
|
|
record_number="EA-RESTRICTED",
|
|
),
|
|
RecordRevision(
|
|
tenant_id="tenant-1",
|
|
record_id="record-restricted",
|
|
identity_id="identity-restricted",
|
|
revision=1,
|
|
class_id="class-1",
|
|
file_plan_node_id="plan-1",
|
|
title="Restricted decision",
|
|
state="open",
|
|
source_authority_mode="native_authoritative",
|
|
access_mode="restricted",
|
|
purpose="case-work",
|
|
institutional_context={},
|
|
external_reference={},
|
|
search_text="restricted decision",
|
|
valid_from=NOW,
|
|
recorded_at=NOW,
|
|
snapshot={},
|
|
),
|
|
)
|
|
)
|
|
self.session.commit()
|
|
reference = SearchResourceReference(
|
|
tenant_id="tenant-1",
|
|
module_id="records",
|
|
resource_type=RESOURCE_TYPE,
|
|
resource_id="record-restricted",
|
|
)
|
|
request = SearchAuthorizationRequest(reference=reference, source_revision="1")
|
|
self.assertFalse(
|
|
self.source.authorize(
|
|
self.session, _principal({READ_SCOPE}), requests=(request,)
|
|
)[reference.key]
|
|
)
|
|
|
|
self.session.add(
|
|
RecordAccessGrantRevision(
|
|
tenant_id="tenant-1",
|
|
grant_id="search-grant-1",
|
|
record_id="record-restricted",
|
|
revision=1,
|
|
status="active",
|
|
subject_type="account",
|
|
subject_id="account-1",
|
|
actions=["read"],
|
|
allowed_purposes=["records.search"],
|
|
reason="Search is required for assigned work.",
|
|
valid_from=NOW,
|
|
recorded_at=NOW,
|
|
changed_by="account-1",
|
|
institutional_context={},
|
|
idempotency_key="search-grant-create-1",
|
|
request_sha256="a" * 64,
|
|
)
|
|
)
|
|
self.session.commit()
|
|
self.assertTrue(
|
|
self.source.authorize(
|
|
self.session, _principal({READ_SCOPE}), requests=(request,)
|
|
)[reference.key]
|
|
)
|
|
|
|
|
|
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="user-1"),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|