from pathlib import Path import httpx import pytest from qonto_assistant.app import create_app from qonto_assistant.audit import AuditLogger from qonto_assistant.config import Settings from qonto_assistant.contracts import ActorClaims from qonto_assistant.credentials import EnvironmentCredentialProvider from qonto_assistant.errors import PolicyDeniedError from qonto_assistant.policy import PolicyEngine from qonto_assistant.qonto_client import QontoClient from qonto_assistant.rate_limits import ConcurrencyLimiter, RateLimiter from qonto_assistant.service import CapabilityService POLICY_FILE = ( Path(__file__).resolve().parents[1] / "src" / "qonto_assistant" / "policy" / "qonto-v1.yaml" ) def _settings() -> Settings: return Settings( service_name="qonto-assistant", default_tenant_id="binky", default_actor_lane="green", required_scope="finance.qonto.read", enforce_scope=False, policy_file=POLICY_FILE, qonto_base_url="https://example.test", qonto_fixture_dir=None, qonto_auth_mode="legacy_api_key", qonto_organization_path="/v2/organization", qonto_transactions_path="/v2/transactions", qonto_timeout_seconds=1, qonto_max_retries=0, qonto_secret_ttl_seconds=60, rate_limit_requests=20, rate_limit_window_seconds=60, max_concurrency=4, deny_escalation_enabled=True, deny_escalation_threshold=3, deny_escalation_window_seconds=60, deny_escalation_lockout_seconds=300, audit_heartbeat_interval_seconds=86400, key_cape_jwks_url=None, key_cape_issuer="https://key-cape.netkingdom", key_cape_audience="qonto-assistant", key_cape_required=False, key_cape_cache_seconds=300, key_cape_timeout_seconds=5, flex_auth_base_url=None, flex_auth_timeout_seconds=3, tenant_engine_base_url=None, tenant_engine_timeout_seconds=3, tenant_engine_required_roles=frozenset({"VEN", "CUS"}), credential_source="env", openbao_path="tenants/binky/qonto-api", openbao_command="bao", openbao_timeout_seconds=5, mcp_auth_token=None, host="127.0.0.1", port=8080, ) def _claims() -> ActorClaims: return ActorClaims(actor_id="codex", tenant_id="binky", lane="green") def _service(monkeypatch) -> tuple[CapabilityService, list[dict[str, object]]]: monkeypatch.setenv("API_USER", "binky-user") monkeypatch.setenv("API_KEY", "top-secret") events: list[dict[str, object]] = [] settings = _settings() organization_payload = { "organization": { "name": "Binky Hedgehog GmbH", "legal_name": "Binky Hedgehog GmbH", "slug": "binky-hedgehog-gmbh-6923", "legal_country": "DE", "legal_registration_date": "2019-03-15", "bank_accounts": [ { "name": "Hauptkonto", "slug": "main-account", "currency": "EUR", "balance": 2185.94, "authorized_balance": 2185.94, "iban": "DE02100100101234566810", "main": True, "status": "active", }, { "name": "Kickstart Business", "slug": "secondary-account", "currency": "EUR", "balance": 0, "authorized_balance": 0, "iban": "DE02100100101234567038", "main": False, "status": "active", }, ], } } transactions_payload = { "transactions": [ { "id": "tx-qonto", "settled_at": "2026-07-01T08:00:00Z", "label": "Qonto", "side": "debit", "amount": 70.8, "currency": "EUR", "category": "subscription", "operation_type": "qonto_fee", "status": "completed", }, { "id": "tx-hub31-1", "settled_at": "2026-06-02T08:00:00Z", "label": "HUB31", "side": "debit", "amount": 297.5, "currency": "EUR", "category": "other_expense", "operation_type": "transfer", "status": "completed", }, { "id": "tx-hub31-2", "settled_at": "2026-05-02T08:00:00Z", "label": "HUB31", "side": "debit", "amount": 297.5, "currency": "EUR", "category": "other_expense", "operation_type": "transfer", "status": "completed", }, { "id": "tx-stripe", "settled_at": "2026-06-29T08:00:00Z", "label": "Stripe", "side": "credit", "amount": 8.55, "currency": "EUR", "category": "other_income", "operation_type": "income", "status": "completed", }, ] } def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/v2/organization": return httpx.Response(200, json=organization_payload) if request.url.path == "/v2/transactions": return httpx.Response(200, json=transactions_payload) return httpx.Response(404, json={"error": "not_found"}) client = QontoClient( base_url=settings.qonto_base_url, organization_path=settings.qonto_organization_path, transactions_path=settings.qonto_transactions_path, auth_mode=settings.qonto_auth_mode, timeout_seconds=settings.qonto_timeout_seconds, max_retries=settings.qonto_max_retries, credential_provider=EnvironmentCredentialProvider(), transport=httpx.MockTransport(handler), ) policy = PolicyEngine.from_file( settings.policy_file, required_scope=settings.required_scope, enforce_scope=settings.enforce_scope, ) service = CapabilityService( client=client, policy=policy, audit_logger=AuditLogger(sink=events.append), rate_limiter=RateLimiter(limit=20, window_seconds=60), concurrency_limiter=ConcurrencyLimiter(limit=4), ) return service, events def test_accounts_contract_returns_redacted_summary(monkeypatch) -> None: service, _ = _service(monkeypatch) payload = service.get_accounts(claims=_claims(), request_id="req-accounts") assert payload["organization"]["name"] == "Binky Hedgehog GmbH" assert payload["accounts"][0]["iban_last4"] == "6810" assert "iban" not in payload["accounts"][0] def test_transactions_contract_denies_oversized_page_size(monkeypatch) -> None: service, events = _service(monkeypatch) try: service.list_transactions( claims=_claims(), request_id="req-deny", account_slug=None, page=1, page_size=101, window_days=31, status="completed", side=None, ) except PolicyDeniedError as exc: assert exc.error_code == "arg_constraint" else: raise AssertionError("Expected policy denial") assert events[-1]["decision"] == "deny" assert events[-1]["deny_reason"] == "arg_constraint" def test_snapshot_contract_returns_cost_run_rate_hints_for_90_day_window(monkeypatch) -> None: service, events = _service(monkeypatch) payload = service.get_snapshot( claims=_claims(), request_id="req-snapshot", window_days=90, page_size=50, ) assert payload["summary"]["total_balance"] == 2185.94 assert payload["cost_run_rate_hints"]["recurring_debits"][0]["label"] == "HUB31" assert any(event["capability"] == "snapshot_bundle" for event in events) async def test_app_lifecycle_and_reconciliation_use_the_request_audit_stream(monkeypatch) -> None: service, events = _service(monkeypatch) app = create_app(settings=_settings(), service=service) async with app.router.lifespan_context(app): assert events[-1]["event_class"] == "audit.heartbeat" assert events[-1]["reason"] == "startup" async with httpx.AsyncClient( transport=httpx.ASGITransport(app=app), base_url="http://test" ) as client: response = await client.get( "/v1/audit/reconciliation", headers={"X-Actor-ID": "observer", "X-Tenant-ID": "binky"}, ) assert response.status_code == 200 assert response.json()["last_stream_sequence"] == 1 assert response.json()["source_transition_counts"] == { "audit.allow": 0, "audit.deny": 0, } assert len(events) == 1 assert events[-1]["event_class"] == "audit.heartbeat" assert events[-1]["reason"] == "shutdown" assert events[-1]["stream_sequence"] == 2 async def test_client_closes_when_shutdown_heartbeat_fails(monkeypatch) -> None: service, _ = _service(monkeypatch) closed = [] monkeypatch.setattr(service.client, "close", lambda: closed.append(True)) def failing_shutdown(payload): if payload.get("reason") == "shutdown": raise RuntimeError("sink unavailable") service.audit_logger.sink = failing_shutdown app = create_app(settings=_settings(), service=service) with pytest.raises(ExceptionGroup) as caught: async with app.router.lifespan_context(app): pass assert caught.group_contains(RuntimeError, match="sink unavailable") assert closed == [True] def test_app_rejects_split_audit_streams(monkeypatch) -> None: service, _ = _service(monkeypatch) with pytest.raises(ValueError, match="share one audit_logger"): create_app(settings=_settings(), service=service, audit_logger=AuditLogger())