from __future__ import annotations import json import time import uuid from datetime import UTC, datetime from decimal import Decimal from pathlib import Path import pytest from commercial_runtime_testkit import seed_meter from sqlalchemy import create_engine, func, select from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool import app.models # noqa: F401 - 注册完整 metadata from app.api.deps import CurrentUserContext from app.db.base_class import Base from app.models.commercial import UsageMeterEvent from app.models.commercial_runtime import CommercialRuntimeReservation from app.models.financial_connector import FinancialConnectorConfig, FinancialConnectorEvent from app.models.financial_record import ExpenseClaim, ExpenseClaimItem from app.schemas.financial_connector import FinancialEventEnvelope from app.services.expense_claim_attachment_commercial import ( ExpenseClaimAttachmentCommercialAccessDenied, stage_attachment_deletion, stage_attachment_replacement, stage_claim_attachment_deletion, ) from app.services.expense_claim_attachment_storage import ExpenseClaimAttachmentStorage from app.services.financial_connector_auth import ( FinancialConnectorAuthError, FinancialConnectorSecretResolver, sign_financial_event, ) from app.services.financial_connector_commercial import ( FinancialConnectorCommercialAccessDenied, ) from app.services.financial_connector_ingestion import ( FinancialConnectorConflictError, FinancialConnectorIngestionService, ) @pytest.fixture() def factory() -> sessionmaker[Session]: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(engine) result = sessionmaker(bind=engine, expire_on_commit=False) try: yield result finally: Base.metadata.drop_all(engine) engine.dispose() def test_connector_only_meters_new_authenticated_committed_event( factory: sessionmaker[Session], ) -> None: now = datetime.now(UTC) timestamp = int(now.timestamp()) with factory() as db: claim = _seed_connector_boundary(db, now=now, hard_limit=Decimal("10")) db.commit() envelope = _envelope(claim, "SENSITIVE-EXTERNAL-EVENT-001", now) first = _ingest(db, envelope, timestamp) assert first.replayed is False assert db.scalar(select(func.count(UsageMeterEvent.id))) == 0 db.commit() replay = _ingest(db, envelope, timestamp) db.commit() assert replay.replayed is True conflicting = envelope.model_copy( update={"payload": {**envelope.payload, "amount": "66.01"}} ) with pytest.raises(FinancialConnectorConflictError): _ingest(db, conflicting, timestamp) db.rollback() with pytest.raises(FinancialConnectorAuthError): _ingest(db, envelope, timestamp, signature="sha256=" + "0" * 64) db.rollback() usage = db.scalars(select(UsageMeterEvent)).one() reservations = list(db.scalars(select(CommercialRuntimeReservation)).all()) assert Decimal(usage.quantity) == Decimal("1") assert len(reservations) == 1 and reservations[0].status == "committed" serialized = json.dumps(usage.metadata_json, ensure_ascii=False) assert envelope.external_event_id not in serialized assert envelope.correlation_id not in serialized assert claim.claim_no not in serialized def test_connector_rollback_releases_reservation_and_writes_no_usage( factory: sessionmaker[Session], ) -> None: now = datetime.now(UTC) with factory() as db: claim = _seed_connector_boundary(db, now=now, hard_limit=Decimal("10")) db.commit() envelope = _envelope(claim, "rollback-event", now) _ingest(db, envelope, int(now.timestamp())) db.rollback() reservation = db.scalars(select(CommercialRuntimeReservation)).one() assert reservation.status == "released" assert db.scalar(select(func.count(UsageMeterEvent.id))) == 0 assert db.scalar(select(func.count(FinancialConnectorEvent.id))) == 0 retry = _ingest(db, envelope, int(now.timestamp())) db.commit() assert retry.replayed is False assert db.scalar(select(func.count(FinancialConnectorEvent.id))) == 1 assert db.scalar(select(func.count(UsageMeterEvent.id))) == 1 db.refresh(reservation) assert reservation.status == "committed" def test_connector_quota_denial_happens_before_second_event_persistence( factory: sessionmaker[Session], ) -> None: now = datetime.now(UTC) with factory() as db: claim = _seed_connector_boundary(db, now=now, hard_limit=Decimal("1")) db.commit() _ingest(db, _envelope(claim, "quota-first", now), int(now.timestamp())) db.commit() with pytest.raises(FinancialConnectorCommercialAccessDenied): _ingest(db, _envelope(claim, "quota-second", now), int(now.timestamp())) db.rollback() assert db.scalar(select(func.count(FinancialConnectorEvent.id))) == 1 assert db.scalar(select(func.count(UsageMeterEvent.id))) == 1 def test_attachment_bytes_commit_and_rollback_follow_business_transaction( factory: sessionmaker[Session], tmp_path: Path, ) -> None: now = datetime.now(UTC) storage = _Storage(tmp_path / "expense-claims") item = ExpenseClaimItem(id="item-a", claim_id="claim-a", invoice_id=None) current_user = _user() old_path = storage.build_item_dir("claim-a", "item-a") / "old.txt" old_path.parent.mkdir(parents=True) old_path.write_bytes(b"old") item.invoice_id = storage.to_storage_key(old_path) with factory() as db: seed_meter( db, "tenant-a", now, basis="bytes", tool_type="storage", tool_name="attachment.upload", hard_limit=Decimal("100"), preflight_quantity=Decimal("100"), ) db.commit() stage_attachment_replacement( db, storage=storage, item=item, current_user=current_user, claim_id="claim-a", content=b"new-content", request_id="upload-rollback", ) new_path = storage.build_item_dir("claim-a", "item-a") / "new.txt" new_path.parent.mkdir(parents=True) new_path.write_bytes(b"new-content") db.rollback() assert old_path.read_bytes() == b"old" assert not new_path.exists() assert db.scalar(select(func.count(UsageMeterEvent.id))) == 0 assert db.scalars(select(CommercialRuntimeReservation)).one().status == "released" stage_attachment_replacement( db, storage=storage, item=item, current_user=current_user, claim_id="claim-a", content=b"accepted", request_id="upload-commit", ) accepted_path = storage.build_item_dir("claim-a", "item-a") / "accepted.txt" accepted_path.parent.mkdir(parents=True) accepted_path.write_bytes(b"accepted") db.commit() assert accepted_path.read_bytes() == b"accepted" assert not old_path.exists() usage = db.scalars(select(UsageMeterEvent)).one() assert Decimal(usage.quantity) == Decimal(len(b"accepted")) assert usage.metadata_json["quantity_basis"] == "bytes" @pytest.mark.parametrize( ("basis", "hard_limit"), [("bytes", Decimal("2")), ("objects", Decimal("100"))], ) def test_attachment_quota_or_wrong_basis_denies_before_existing_file_moves( factory: sessionmaker[Session], tmp_path: Path, basis: str, hard_limit: Decimal, ) -> None: now = datetime.now(UTC) storage = _Storage(tmp_path / "expense-claims") item = ExpenseClaimItem(id="item-a", claim_id="claim-a", invoice_id=None) old_path = storage.build_item_dir("claim-a", "item-a") / "old.txt" old_path.parent.mkdir(parents=True) old_path.write_bytes(b"keep-me") item.invoice_id = storage.to_storage_key(old_path) with factory() as db: seed_meter( db, "tenant-a", now, basis=basis, tool_type="storage", tool_name="attachment.upload", hard_limit=hard_limit, preflight_quantity=Decimal("100"), ) db.commit() with pytest.raises(ExpenseClaimAttachmentCommercialAccessDenied): stage_attachment_replacement( db, storage=storage, item=item, current_user=_user(), claim_id="claim-a", content=b"three", request_id="quota-denied", ) assert old_path.read_bytes() == b"keep-me" assert db.scalar(select(func.count(CommercialRuntimeReservation.id))) == 0 assert db.scalar(select(func.count(UsageMeterEvent.id))) == 0 def test_attachment_delete_restores_on_rollback_and_finalizes_on_commit( factory: sessionmaker[Session], tmp_path: Path, ) -> None: storage = _Storage(tmp_path / "expense-claims") item = ExpenseClaimItem(id="item-a", claim_id="claim-a", invoice_id=None) file_path = storage.build_item_dir("claim-a", "item-a") / "invoice.pdf" file_path.parent.mkdir(parents=True) file_path.write_bytes(b"invoice") item.invoice_id = storage.to_storage_key(file_path) with factory() as db: stage_attachment_deletion(db, storage=storage, item=item) assert not file_path.exists() db.rollback() assert file_path.read_bytes() == b"invoice" stage_attachment_deletion(db, storage=storage, item=item) db.commit() assert not file_path.exists() def test_claim_attachment_tree_delete_follows_business_transaction( factory: sessionmaker[Session], tmp_path: Path, ) -> None: storage = _Storage(tmp_path / "expense-claims") file_path = storage.build_item_dir("claim-a", "item-a") / "invoice.pdf" file_path.parent.mkdir(parents=True) file_path.write_bytes(b"invoice") with factory() as db: stage_claim_attachment_deletion(db, storage=storage, claim_id="claim-a") assert not file_path.exists() db.rollback() assert file_path.read_bytes() == b"invoice" stage_claim_attachment_deletion(db, storage=storage, claim_id="claim-a") db.commit() assert not file_path.exists() def test_multiple_attachment_replacements_restore_in_reverse_order( factory: sessionmaker[Session], tmp_path: Path, ) -> None: storage = _Storage(tmp_path / "expense-claims") item = ExpenseClaimItem(id="item-a", claim_id="claim-a", invoice_id=None) original_path = storage.build_item_dir("claim-a", "item-a") / "original.txt" original_path.parent.mkdir(parents=True) original_path.write_bytes(b"original") item.invoice_id = storage.to_storage_key(original_path) with factory() as db: stage_attachment_replacement( db, storage=storage, item=item, current_user=_user(), claim_id="claim-a", content=b"middle", request_id="replacement-one", ) middle_path = storage.build_item_dir("claim-a", "item-a") / "middle.txt" middle_path.parent.mkdir(parents=True) middle_path.write_bytes(b"middle") item.invoice_id = storage.to_storage_key(middle_path) stage_attachment_replacement( db, storage=storage, item=item, current_user=_user(), claim_id="claim-a", content=b"latest", request_id="replacement-two", ) latest_path = storage.build_item_dir("claim-a", "item-a") / "latest.txt" latest_path.parent.mkdir(parents=True) latest_path.write_bytes(b"latest") db.rollback() assert original_path.read_bytes() == b"original" assert not middle_path.exists() assert not latest_path.exists() class _Storage(ExpenseClaimAttachmentStorage): def __init__(self, root: Path) -> None: self._root = root def root(self) -> Path: return self._root.resolve() def _seed_connector_boundary( db: Session, *, now: datetime, hard_limit: Decimal, ) -> ExpenseClaim: seed_meter( db, "tenant-a", now, basis="events", tool_type="connector", tool_name="financial.ingest", hard_limit=hard_limit, preflight_quantity=Decimal("1"), ) db.add( FinancialConnectorConfig( id=str(uuid.uuid4()), tenant_id="tenant-a", provider="metered-bank", environment="mock", key_version="v1", secret_ref="connector/metered-test", allowed_event_types_json=["payment_settled"], clock_skew_seconds=300, status="active", created_by="platform-admin", ) ) claim = ExpenseClaim( id=str(uuid.uuid4()), tenant_id="tenant-a", claim_no=f"BX-METER-{uuid.uuid4().hex[:8]}", employee_name="计量测试员工", department_name="财务部", expense_type="travel", reason="连接器资源边界测试", location="上海", amount=Decimal("66.00"), currency="CNY", invoice_count=1, occurred_at=now, submitted_at=now, status="pending_payment", approval_stage="待付款", risk_flags_json=[], ) db.add(claim) db.flush() return claim def _envelope(claim: ExpenseClaim, event_id: str, now: datetime) -> FinancialEventEnvelope: return FinancialEventEnvelope( tenant_id="tenant-a", external_event_id=event_id, event_type="payment_settled", occurred_at=now, correlation_id=f"sensitive-correlation-{event_id}"[:64], payload={ "claim_id": claim.id, "claim_reference": claim.claim_no, "amount": str(claim.amount), "currency": claim.currency, "external_payment_reference": f"SENSITIVE-PAYMENT-{event_id}", }, ) def _ingest( db: Session, envelope: FinancialEventEnvelope, timestamp: int, *, signature: str = "", ): resolved_signature = signature or sign_financial_event( envelope, timestamp=timestamp, secret="metered-server-secret", tenant_id="tenant-a", provider="metered-bank", key_version="v1", ) return FinancialConnectorIngestionService( db, secrets=FinancialConnectorSecretResolver( {"connector/metered-test": "metered-server-secret"} ), now_epoch=timestamp, ).ingest( envelope, tenant_header="tenant-a", provider_header="metered-bank", key_version_header="v1", timestamp_header=str(timestamp), signature_header=resolved_signature, ) def _user() -> CurrentUserContext: return CurrentUserContext( username="employee-a", name="员工甲", role_codes=["employee"], is_admin=False, tenant_id="tenant-a", auth_session_id=f"session-{time.time_ns()}", )