qonto-assistant/tests/test_api.py

from pathlib import Path

import httpx

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,
        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)