"""Response-only parity/query regression with isolated synthetic SQLite data. This does not benchmark imports, mutate storage, or replace route authorization. The isolated schema uses string columns for external module identities instead of importing optional modules. All response, share-state and Campaign-use queries execute through the real Files ORM. """ from __future__ import annotations from datetime import UTC, datetime, timedelta import unittest from sqlalchemy import Column, MetaData, String, Table, create_engine, event, inspect from sqlalchemy.orm import Session from sqlalchemy.types import NullType from govoplan_files.backend.db.models import ( CampaignAttachmentUse, FileAsset, FileBlob, FileShare, FileVersion, ) from govoplan_files.backend.route_support import _asset_list_response, _asset_response class UploadResponseBatchingTests(unittest.TestCase): def setUp(self): self.engine = create_engine("sqlite://") self.addCleanup(self.engine.dispose) metadata = MetaData() for model in ( FileBlob, FileAsset, FileVersion, FileShare, CampaignAttachmentUse, ): Table( model.__tablename__, metadata, *( Column( column.name, String(36) if isinstance(column.type, NullType) else column.type, primary_key=column.primary_key, nullable=column.nullable, default=column.default.arg if column.default is not None else None, server_default=column.server_default.arg if column.server_default is not None else None, ) for column in model.__table__.columns ), ) metadata.create_all(self.engine) now = datetime.now(UTC) with self.engine.begin() as connection: def seed(row): # Core inserts into separate fixture metadata avoid resolving # foreign identities or changing global ORM declarations. values = { attribute.columns[0].name: getattr(row, attribute.key) for attribute in inspect(type(row)).column_attrs if attribute.key in row.__dict__ } connection.execute(metadata.tables[row.__tablename__].insert(), values) for index in range(64): asset_id = f"asset-{index:03}" version_id = f"version-{index:03}" blob_id = f"blob-{index:03}" checksum = f"{index:064x}" owner_type = "group" if index % 2 else "user" seed( FileBlob( id=blob_id, tenant_id="tenant-fixture", storage_backend="local", storage_key=f"synthetic/{blob_id}", checksum_sha256=checksum, size_bytes=index + 10, content_type="text/plain", ref_count=1, ) ) seed( FileAsset( id=asset_id, tenant_id="tenant-fixture", owner_type=owner_type, owner_user_id="user-fixture" if owner_type == "user" else None, owner_group_id="group-fixture" if owner_type == "group" else None, current_version_id=version_id, display_path=f"imported/{index:03}.txt", filename=f"{index:03}.txt", description=f"Synthetic response {index}", retained_until=now + timedelta(days=365) if index % 2 else None, legal_hold=bool(index % 3), lifecycle_revision=index + 2, lifecycle_reason=f"Retention decision {index}", deleted_at=now - timedelta(days=1) if index % 7 == 0 else None, metadata_={ "fixture": index, "source_provenance": { "source_type": "archive", "revision": f"source-{index}", "metadata": { "archive_source_file_id": "synthetic-archive" }, }, "source_revision": f"source-{index}", }, ) ) seed( FileVersion( id=version_id, tenant_id="tenant-fixture", file_asset_id=asset_id, blob_id=blob_id, version_number=1, filename_at_upload=f"{index:03}.txt", display_path_at_upload=f"imported/{index:03}.txt", size_bytes=index + 10, checksum_sha256=checksum, content_type="text/plain", ) ) for rank, state in enumerate(("active", "expired", "revoked")): seed( FileShare( id=f"share-{index:03}-{state}", tenant_id="tenant-fixture", file_asset_id=asset_id, target_type="group" if rank == 0 else "user", target_id=f"target-{state}", permission="write" if rank == 0 else "read", created_by_user_id="user-fixture", created_at=now - timedelta(hours=rank + 1), expires_at=now - timedelta(days=1) if state == "expired" else None, revoked_at=now - timedelta(days=1) if state == "revoked" else None, revoked_by_user_id="user-fixture" if state == "revoked" else None, ) ) # Duplicate sent evidence must still produce one response; # built-only usage and missing usage are not audit-relevant. stages = ( ("sent", "sent") if index % 3 == 0 else ("built",) if index % 3 == 1 else () ) for use_index, stage in enumerate(stages): seed( CampaignAttachmentUse( id=f"use-{index:03}-{use_index}", tenant_id="tenant-fixture", campaign_id="campaign-fixture", campaign_version_id="campaign-version-fixture", campaign_job_id=f"job-{use_index}", file_asset_id=asset_id, file_version_id=version_id, file_blob_id=blob_id, filename_used=f"{index:03}.txt", checksum_sha256=checksum, size_bytes=index + 10, content_type="text/plain", use_stage=stage, ) ) def render(self, *, batched: bool, include_shares: bool): statements = [] def record_select( connection, cursor, statement, parameters, context, executemany ): if statement.lstrip().upper().startswith("SELECT"): statements.append(statement) # A fresh session ensures neither path receives preloaded versions or # blobs. Loading the already-authorized asset list is deliberately # outside the measurement; this measures response construction only. with Session(self.engine) as session: assets = session.query(FileAsset).order_by(FileAsset.id).all() event.listen(self.engine, "before_cursor_execute", record_select) try: responses = ( _asset_list_response(session, assets, include_shares=include_shares) if batched else [ _asset_response(session, asset, include_shares=include_shares) for asset in assets ] ) result = [response.model_dump(mode="json") for response in responses] finally: event.remove(self.engine, "before_cursor_execute", record_select) return result, statements def test_batched_response_matches_all_metadata_lifecycle_shares_and_audit_flags( self, ): for include_shares in (False, True): with self.subTest(include_shares=include_shares): batched, _ = self.render(batched=True, include_shares=include_shares) individual, _ = self.render( batched=False, include_shares=include_shares ) self.assertEqual(individual, batched) self.assertEqual(64, len(batched)) for index, item in enumerate(batched): self.assertEqual(f"asset-{index:03}", item["id"]) self.assertEqual(index + 2, item["lifecycle_revision"]) self.assertEqual( f"Retention decision {index}", item["lifecycle_reason"] ) self.assertEqual( bool(index % 2), item["retained_until"] is not None ) self.assertEqual(bool(index % 3), item["legal_hold"]) self.assertEqual(index % 3 == 0, item["audit_relevant"]) self.assertEqual(index % 7 == 0, item["deleted_at"] is not None) self.assertEqual(f"source-{index}", item["source_revision"]) if include_shares: self.assertEqual( [ f"share-{index:03}-{state}" for state in ("active", "expired", "revoked") ], [share["id"] for share in item["shares"]], ) self.assertEqual( [True, False, False], [share["active"] for share in item["shares"]], ) else: self.assertEqual([], item["shares"]) def test_64_asset_response_batches_real_selects_instead_of_querying_per_item(self): for include_shares, per_item_queries, batch_queries in ( (False, 192, 2), (True, 256, 3), ): with self.subTest(include_shares=include_shares): individual, individual_queries = self.render( batched=False, include_shares=include_shares ) batched, batched_queries = self.render( batched=True, include_shares=include_shares ) self.assertEqual(individual, batched) self.assertEqual(per_item_queries, len(individual_queries)) self.assertEqual(batch_queries, len(batched_queries)) if __name__ == "__main__": unittest.main()