334 lines
11 KiB
Python
334 lines
11 KiB
Python
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()
|