108 lines
3.3 KiB
Python
108 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.records import RecordContractError, RecordSourceLocator
|
|
from govoplan_cases.backend.db.models import CaseIdentity, CaseRecordRevision
|
|
from govoplan_cases.backend.record_source import CasesRecordSource
|
|
|
|
|
|
NOW = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
tenant_id: str = "tenant-1"
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope == "cases:case:read"
|
|
|
|
|
|
class CasesRecordSourceTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
|
CaseIdentity.__table__.create(self.engine)
|
|
CaseRecordRevision.__table__.create(self.engine)
|
|
self.session = Session(self.engine)
|
|
identity = CaseIdentity(
|
|
id="identity-1",
|
|
tenant_id="tenant-1",
|
|
case_id="case-1",
|
|
case_number="2026/C-1",
|
|
created_by="account-1",
|
|
)
|
|
revision = CaseRecordRevision(
|
|
id="revision-1",
|
|
tenant_id="tenant-1",
|
|
case_id="case-1",
|
|
identity_id="identity-1",
|
|
revision=3,
|
|
case_type_key="permit",
|
|
status_key="review",
|
|
title="Permit application",
|
|
access_mode="restricted",
|
|
search_text="permit application",
|
|
opened_at=NOW,
|
|
recorded_at=NOW,
|
|
snapshot={
|
|
"case_id": "case-1",
|
|
"revision": 3,
|
|
"title": "Permit application",
|
|
},
|
|
)
|
|
self.session.add_all((identity, revision))
|
|
self.session.flush()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
self.engine.dispose()
|
|
|
|
def locator(self) -> RecordSourceLocator:
|
|
return RecordSourceLocator(
|
|
tenant_id="tenant-1",
|
|
source_module="cases",
|
|
resource_type="case_revision",
|
|
resource_id="case-1",
|
|
source_revision="3",
|
|
)
|
|
|
|
def test_resolves_exact_currently_authorized_case_revision(self) -> None:
|
|
with patch(
|
|
"govoplan_cases.backend.record_source.can_access_case", return_value=True
|
|
):
|
|
result = CasesRecordSource().resolve(
|
|
self.session,
|
|
Principal(),
|
|
locator=self.locator(),
|
|
purpose="preserve decision basis",
|
|
)
|
|
|
|
self.assertEqual("2026/C-1 - Permit application", result.label)
|
|
self.assertEqual(64, len(result.content_sha256 or ""))
|
|
self.assertEqual("review", result.metadata["status_key"])
|
|
|
|
def test_object_access_is_rechecked_and_fails_closed(self) -> None:
|
|
with (
|
|
patch(
|
|
"govoplan_cases.backend.record_source.can_access_case",
|
|
return_value=False,
|
|
),
|
|
self.assertRaisesRegex(RecordContractError, "cannot read"),
|
|
):
|
|
CasesRecordSource().resolve(
|
|
self.session,
|
|
Principal(),
|
|
locator=self.locator(),
|
|
purpose="preserve decision basis",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|