Add purpose-bound form evidence storage
This commit is contained in:
@@ -0,0 +1,333 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from io import BytesIO
|
||||
from types import SimpleNamespace
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import UploadFile
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
from starlette.datastructures import Headers
|
||||
|
||||
from govoplan_access.backend.db.models import Account, User
|
||||
from govoplan_core.core.access import (
|
||||
CAPABILITY_ACCESS_DIRECTORY,
|
||||
AccessSubjectRef,
|
||||
UserRef,
|
||||
)
|
||||
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
||||
from govoplan_core.core.form_evidence import (
|
||||
FormEvidenceContractError,
|
||||
FormEvidenceGrantRequest,
|
||||
FormEvidenceInspectionRequest,
|
||||
)
|
||||
from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference
|
||||
from govoplan_core.db.base import Base
|
||||
from govoplan_files.backend.db.models import (
|
||||
FileAsset,
|
||||
FileBlob,
|
||||
FileFormEvidenceGrant,
|
||||
FileVersion,
|
||||
)
|
||||
from govoplan_files.backend.form_evidence import FilesFormEvidenceProvider
|
||||
from govoplan_files.backend.routes.form_evidence import upload_form_evidence
|
||||
|
||||
|
||||
class _Directory:
|
||||
def __init__(self, user: UserRef) -> None:
|
||||
self._user = user
|
||||
|
||||
def get_account(self, account_id: str):
|
||||
del account_id
|
||||
return None
|
||||
|
||||
def get_user(self, user_id: str):
|
||||
return self._user if user_id == self._user.id else None
|
||||
|
||||
def get_users(self, user_ids):
|
||||
return {user_id: self._user for user_id in user_ids if user_id == self._user.id}
|
||||
|
||||
def users_for_tenant(self, tenant_id: str):
|
||||
return (self._user,) if tenant_id == self._user.tenant_id else ()
|
||||
|
||||
def get_group(self, group_id: str):
|
||||
del group_id
|
||||
return None
|
||||
|
||||
def get_groups(self, group_ids):
|
||||
del group_ids
|
||||
return {}
|
||||
|
||||
def groups_for_tenant(self, tenant_id: str):
|
||||
del tenant_id
|
||||
return ()
|
||||
|
||||
def groups_for_user(self, user_id: str, *, tenant_id: str):
|
||||
del user_id, tenant_id
|
||||
return ()
|
||||
|
||||
def display_label(self, subject: AccessSubjectRef):
|
||||
del subject
|
||||
return self._user.display_name
|
||||
|
||||
|
||||
class _Registry:
|
||||
def __init__(self, directory: _Directory) -> None:
|
||||
self._directory = directory
|
||||
|
||||
def has_capability(self, name: str) -> bool:
|
||||
return name == CAPABILITY_ACCESS_DIRECTORY
|
||||
|
||||
def require_capability(self, name: str):
|
||||
if not self.has_capability(name):
|
||||
raise KeyError(name)
|
||||
return self._directory
|
||||
|
||||
|
||||
class FilesFormEvidenceTests(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.engine = create_engine("sqlite+pysqlite:///:memory:")
|
||||
Base.metadata.create_all(
|
||||
self.engine,
|
||||
tables=[
|
||||
Account.__table__,
|
||||
User.__table__,
|
||||
FileBlob.__table__,
|
||||
FileAsset.__table__,
|
||||
FileVersion.__table__,
|
||||
FileFormEvidenceGrant.__table__,
|
||||
ChangeSequenceEntry.__table__,
|
||||
],
|
||||
)
|
||||
self.session = Session(self.engine)
|
||||
user_ref = UserRef(
|
||||
id="user-1",
|
||||
account_id="account-1",
|
||||
tenant_id="tenant-1",
|
||||
email="user@example.test",
|
||||
display_name="Evidence Custodian",
|
||||
)
|
||||
self.provider = FilesFormEvidenceProvider(
|
||||
_Registry(_Directory(user_ref)),
|
||||
SimpleNamespace(file_upload_max_bytes=2_000_000),
|
||||
)
|
||||
self.principal = SimpleNamespace(tenant_id="tenant-1")
|
||||
self.definition_ref = InstitutionalReference(
|
||||
kind="form",
|
||||
owner_module="forms",
|
||||
object_id="form-1",
|
||||
tenant_id="tenant-1",
|
||||
version="3",
|
||||
)
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.session.close()
|
||||
self.engine.dispose()
|
||||
|
||||
def request(self, **overrides: object) -> FormEvidenceGrantRequest:
|
||||
values: dict[str, object] = {
|
||||
"tenant_id": "tenant-1",
|
||||
"instance_id": "instance-1",
|
||||
"definition_ref": self.definition_ref,
|
||||
"evidence_kind": "document",
|
||||
"purpose": "supporting document",
|
||||
"idempotency_key": "grant-1",
|
||||
"expires_at": datetime.now(UTC) + timedelta(hours=1),
|
||||
"custodian_ref": "user:user-1",
|
||||
"max_size_bytes": 1_000_000,
|
||||
"allowed_content_types": ("application/pdf",),
|
||||
}
|
||||
values.update(overrides)
|
||||
return FormEvidenceGrantRequest(**values) # type: ignore[arg-type]
|
||||
|
||||
def test_grant_persists_only_token_hash_and_replays_without_secret(self) -> None:
|
||||
request = self.request()
|
||||
issued = self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=request,
|
||||
)
|
||||
row = self.session.query(FileFormEvidenceGrant).one()
|
||||
|
||||
self.assertIsNotNone(issued.upload_token)
|
||||
self.assertNotEqual(issued.upload_token, row.token_sha256)
|
||||
self.assertLessEqual(
|
||||
issued.expires_at,
|
||||
datetime.now(UTC) + timedelta(minutes=15, seconds=1),
|
||||
)
|
||||
replay = self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=request,
|
||||
)
|
||||
self.assertTrue(replay.replayed)
|
||||
self.assertIsNone(replay.upload_token)
|
||||
self.assertEqual(issued.grant_id, replay.grant_id)
|
||||
|
||||
def test_grant_capacity_counts_outstanding_uploads_but_allows_replay(self) -> None:
|
||||
request = self.request(metadata={"remaining_attachments": 1})
|
||||
self.provider.create_upload_grant(self.session, self.principal, request=request)
|
||||
|
||||
replay = self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=request,
|
||||
)
|
||||
self.assertTrue(replay.replayed)
|
||||
with self.assertRaisesRegex(
|
||||
FormEvidenceContractError,
|
||||
"maximum number of active attachment uploads",
|
||||
):
|
||||
self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=self.request(
|
||||
idempotency_key="grant-2",
|
||||
metadata={"remaining_attachments": 1},
|
||||
),
|
||||
)
|
||||
|
||||
def test_idempotency_key_cannot_be_reused_for_another_request(self) -> None:
|
||||
self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=self.request(),
|
||||
)
|
||||
with self.assertRaisesRegex(
|
||||
FormEvidenceContractError,
|
||||
"idempotency conflict",
|
||||
):
|
||||
self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=self.request(purpose="another purpose"),
|
||||
)
|
||||
|
||||
def test_inspection_accepts_only_exact_verified_managed_version(self) -> None:
|
||||
issued = self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=self.request(),
|
||||
)
|
||||
grant = self.session.get(FileFormEvidenceGrant, issued.grant_id)
|
||||
assert grant is not None
|
||||
blob = FileBlob(
|
||||
id="blob-1",
|
||||
tenant_id="tenant-1",
|
||||
storage_backend="local",
|
||||
storage_key="tenant-1/blob-1",
|
||||
checksum_sha256="a" * 64,
|
||||
size_bytes=100,
|
||||
integrity_status="verified",
|
||||
)
|
||||
asset = FileAsset(
|
||||
id="asset-1",
|
||||
tenant_id="tenant-1",
|
||||
owner_type="user",
|
||||
owner_user_id="user-1",
|
||||
current_version_id="version-1",
|
||||
display_path="Form submissions/instance-1/evidence.pdf",
|
||||
filename="evidence.pdf",
|
||||
)
|
||||
version = FileVersion(
|
||||
id="version-1",
|
||||
tenant_id="tenant-1",
|
||||
file_asset_id="asset-1",
|
||||
blob_id="blob-1",
|
||||
version_number=1,
|
||||
filename_at_upload="evidence.pdf",
|
||||
display_path_at_upload=asset.display_path,
|
||||
content_type="application/pdf",
|
||||
size_bytes=100,
|
||||
checksum_sha256="a" * 64,
|
||||
)
|
||||
self.session.add_all((blob, asset, version))
|
||||
grant.status = "uploaded"
|
||||
grant.file_asset_id = asset.id
|
||||
grant.file_version_id = version.id
|
||||
self.session.flush()
|
||||
reference = EvidenceReference(
|
||||
kind="document",
|
||||
owner_module="files",
|
||||
evidence_id=asset.id,
|
||||
tenant_id="tenant-1",
|
||||
version=version.id,
|
||||
checksum="a" * 64,
|
||||
)
|
||||
|
||||
accepted = self.provider.inspect_evidence(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=FormEvidenceInspectionRequest(
|
||||
tenant_id="tenant-1",
|
||||
instance_id="instance-1",
|
||||
definition_ref=self.definition_ref,
|
||||
evidence=reference,
|
||||
purpose="final submission",
|
||||
final=True,
|
||||
),
|
||||
)
|
||||
self.assertTrue(accepted.accepted)
|
||||
|
||||
rejected = self.provider.inspect_evidence(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=FormEvidenceInspectionRequest(
|
||||
tenant_id="tenant-1",
|
||||
instance_id="another-instance",
|
||||
definition_ref=self.definition_ref,
|
||||
evidence=reference,
|
||||
purpose="final submission",
|
||||
final=True,
|
||||
),
|
||||
)
|
||||
self.assertEqual("rejected", rejected.state)
|
||||
|
||||
def test_public_upload_consumes_grant_and_returns_exact_evidence(self) -> None:
|
||||
issued = self.provider.create_upload_grant(
|
||||
self.session,
|
||||
self.principal,
|
||||
request=self.request(),
|
||||
)
|
||||
assert issued.upload_token is not None
|
||||
stored = SimpleNamespace(
|
||||
asset=SimpleNamespace(id="asset-uploaded"),
|
||||
version=SimpleNamespace(
|
||||
id="version-uploaded",
|
||||
checksum_sha256="c" * 64,
|
||||
size_bytes=8,
|
||||
),
|
||||
)
|
||||
upload = UploadFile(
|
||||
filename="evidence.pdf",
|
||||
file=BytesIO(b"evidence"),
|
||||
headers=Headers({"content-type": "application/pdf"}),
|
||||
)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"govoplan_files.backend.routes.form_evidence.create_file_asset",
|
||||
return_value=stored,
|
||||
) as create,
|
||||
patch("govoplan_files.backend.routes.form_evidence.emit_platform_event"),
|
||||
):
|
||||
response = upload_form_evidence(
|
||||
file=upload,
|
||||
x_form_evidence_token=issued.upload_token,
|
||||
session=self.session,
|
||||
)
|
||||
|
||||
grant = self.session.get(FileFormEvidenceGrant, issued.grant_id)
|
||||
assert grant is not None
|
||||
self.assertEqual("uploaded", grant.status)
|
||||
self.assertEqual("asset-uploaded", grant.file_asset_id)
|
||||
self.assertEqual("version-uploaded", grant.file_version_id)
|
||||
self.assertEqual("asset-uploaded", response["evidence"]["evidence_id"])
|
||||
self.assertEqual(b"evidence", create.call_args.kwargs["data"])
|
||||
self.assertEqual("user-1", create.call_args.kwargs["owner_id"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user