133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
import unittest
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.institutional import InstitutionalReference
|
|
from govoplan_core.core.records import RecordContractError, RecordSourceLocator
|
|
from govoplan_forms_runtime.backend.db.models import (
|
|
FormInstanceIdentity,
|
|
FormInstanceRevision,
|
|
)
|
|
from govoplan_forms_runtime.backend.domain import FormInstance
|
|
from govoplan_forms_runtime.backend.record_source import FormsRuntimeRecordSource
|
|
|
|
|
|
NOW = datetime(2026, 8, 6, 10, 0, tzinfo=UTC)
|
|
|
|
|
|
@dataclass
|
|
class Principal:
|
|
account_id: str = "account-1"
|
|
tenant_id: str = "tenant-1"
|
|
scopes: tuple[str, ...] = ("forms_runtime:submission:participate",)
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope in self.scopes
|
|
|
|
|
|
class FormsRuntimeRecordSourceTests(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)
|
|
instance = FormInstance(
|
|
tenant_id="tenant-1",
|
|
instance_id="submission-1",
|
|
revision=2,
|
|
status="submitted",
|
|
definition_ref=InstitutionalReference(
|
|
kind="form",
|
|
owner_module="forms",
|
|
object_id="permit-form",
|
|
tenant_id="tenant-1",
|
|
version="3",
|
|
),
|
|
values={"name": "Ada"},
|
|
validation_results=(),
|
|
receipt_id="receipt-1",
|
|
recorded_at=NOW,
|
|
change_reason="Submitted.",
|
|
created_by="account-1",
|
|
changed_by="account-1",
|
|
)
|
|
self.session.add_all(
|
|
(
|
|
FormInstanceIdentity(
|
|
id="identity-1",
|
|
tenant_id="tenant-1",
|
|
instance_id="submission-1",
|
|
definition_id="permit-form",
|
|
definition_revision="3",
|
|
created_by="account-1",
|
|
),
|
|
FormInstanceRevision(
|
|
id="revision-2",
|
|
tenant_id="tenant-1",
|
|
instance_id="submission-1",
|
|
identity_id="identity-1",
|
|
revision=2,
|
|
status="submitted",
|
|
recorded_at=NOW,
|
|
snapshot=instance.to_dict(),
|
|
changed_by="account-1",
|
|
),
|
|
)
|
|
)
|
|
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="forms_runtime",
|
|
resource_type="form_submission_revision",
|
|
resource_id="submission-1",
|
|
source_revision="2",
|
|
)
|
|
|
|
def test_resolves_exact_authorized_immutable_submission(self) -> None:
|
|
result = FormsRuntimeRecordSource().resolve(
|
|
self.session,
|
|
Principal(),
|
|
locator=self.locator(),
|
|
purpose="file permit request",
|
|
)
|
|
|
|
self.assertEqual("Form submission receipt-1", result.label)
|
|
self.assertEqual(64, len(result.content_sha256 or ""))
|
|
self.assertEqual("submitted", result.metadata["status"])
|
|
self.assertNotIn("values", result.metadata)
|
|
|
|
def test_current_access_is_rechecked(self) -> None:
|
|
with self.assertRaisesRegex(RecordContractError, "cannot read"):
|
|
FormsRuntimeRecordSource().resolve(
|
|
self.session,
|
|
Principal(account_id="account-2"),
|
|
locator=self.locator(),
|
|
purpose="file permit request",
|
|
)
|
|
|
|
result = FormsRuntimeRecordSource().resolve(
|
|
self.session,
|
|
Principal(
|
|
account_id="account-2",
|
|
scopes=("forms_runtime:workspace:read",),
|
|
),
|
|
locator=self.locator(),
|
|
purpose="records administration",
|
|
)
|
|
self.assertEqual("receipt-1", result.metadata["receipt_id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|