170 lines
5.1 KiB
Python
170 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.search import SearchDocument, SearchQuery
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_search.backend.db.models import (
|
|
SearchIndexAclToken,
|
|
SearchIndexDocument,
|
|
)
|
|
from govoplan_search.backend.service import SearchIndexService
|
|
|
|
|
|
class _Registry:
|
|
def manifests(self):
|
|
return (
|
|
SimpleNamespace(id="search"),
|
|
SimpleNamespace(id="cases"),
|
|
)
|
|
|
|
|
|
class SearchServiceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite://")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=(
|
|
SearchIndexDocument.__table__,
|
|
SearchIndexAclToken.__table__,
|
|
),
|
|
)
|
|
self.session = Session(self.engine)
|
|
self.service = SearchIndexService(_Registry())
|
|
self.principal = SimpleNamespace(
|
|
tenant_id="tenant-1",
|
|
account_id="account-1",
|
|
membership_id="membership-1",
|
|
identity_id=None,
|
|
group_ids=frozenset({"group-1"}),
|
|
role_ids=frozenset(),
|
|
function_assignment_ids=frozenset(),
|
|
scopes=frozenset({"cases:case:read"}),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def test_index_search_and_acl_filtering(self) -> None:
|
|
self.service.upsert_document(
|
|
self.session,
|
|
self.principal,
|
|
document=SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Resident parking permit",
|
|
summary="Monthly permit review",
|
|
url="/cases/case-1",
|
|
acl_tokens=("group:group-1",),
|
|
),
|
|
)
|
|
self.service.upsert_document(
|
|
self.session,
|
|
self.principal,
|
|
document=SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-2",
|
|
title="Restricted permit",
|
|
url="/cases/case-2",
|
|
acl_tokens=("account:someone-else",),
|
|
),
|
|
)
|
|
self.session.flush()
|
|
|
|
results = self.service.search(
|
|
self.session,
|
|
self.principal,
|
|
query=SearchQuery(text="permit", tenant_id="tenant-1"),
|
|
)
|
|
|
|
self.assertEqual(["case-1"], [result.resource_id for result in results])
|
|
|
|
def test_upsert_replaces_acl_tokens_and_delete_is_idempotent(self) -> None:
|
|
document = SearchDocument(
|
|
tenant_id="tenant-1",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Permit",
|
|
url="/cases/case-1",
|
|
acl_tokens=("group:group-1",),
|
|
)
|
|
self.service.upsert_document(
|
|
self.session, self.principal, document=document
|
|
)
|
|
self.session.flush()
|
|
self.service.upsert_document(
|
|
self.session,
|
|
self.principal,
|
|
document=SearchDocument(
|
|
**{
|
|
**{
|
|
field: getattr(document, field)
|
|
for field in (
|
|
"tenant_id",
|
|
"module_id",
|
|
"resource_type",
|
|
"resource_id",
|
|
"title",
|
|
"url",
|
|
)
|
|
},
|
|
"acl_tokens": ("account:account-1",),
|
|
}
|
|
),
|
|
)
|
|
self.session.flush()
|
|
|
|
self.assertEqual(
|
|
["account:account-1"],
|
|
[
|
|
token.token
|
|
for token in self.session.query(SearchIndexAclToken).all()
|
|
],
|
|
)
|
|
arguments = {
|
|
"tenant_id": "tenant-1",
|
|
"module_id": "cases",
|
|
"resource_type": "case",
|
|
"resource_id": "case-1",
|
|
}
|
|
self.assertTrue(
|
|
self.service.delete_document(
|
|
self.session, self.principal, **arguments
|
|
)
|
|
)
|
|
self.assertFalse(
|
|
self.service.delete_document(
|
|
self.session, self.principal, **arguments
|
|
)
|
|
)
|
|
|
|
def test_index_rejects_cross_tenant_write(self) -> None:
|
|
with self.assertRaises(PermissionError):
|
|
self.service.upsert_document(
|
|
self.session,
|
|
self.principal,
|
|
document=SearchDocument(
|
|
tenant_id="tenant-2",
|
|
module_id="cases",
|
|
resource_type="case",
|
|
resource_id="case-1",
|
|
title="Permit",
|
|
url="/cases/case-1",
|
|
visibility="tenant",
|
|
),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|