tegwick
ca12843013
Implements QONTO-WP-0002 (policy-gated Qonto REST service with audit logging, rate limiting, and credential handling) and the ADHOC-2026-07-21 follow-up (fixture-backed local smoke mode, repo classification metadata). Marks QONTO-WP-0001/0002 and the ad-hoc workplan finished, and regenerates WORK-RECORDS.md and the ADHOC workplan's state_hub_workstream_id via fix-consistency. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
import httpx
|
|
|
|
from qonto_assistant.contracts import QontoCredentials
|
|
from qonto_assistant.qonto_client import QontoClient
|
|
|
|
|
|
class StubCredentialProvider:
|
|
def __init__(self) -> None:
|
|
self.invalidated = False
|
|
|
|
def get_credentials(self) -> QontoCredentials:
|
|
return QontoCredentials(api_user="binky-user", api_key="top-secret")
|
|
|
|
def invalidate(self) -> None:
|
|
self.invalidated = True
|
|
|
|
|
|
def test_client_sends_legacy_api_key_header_and_params() -> None:
|
|
provider = StubCredentialProvider()
|
|
seen: dict[str, object] = {}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen["auth"] = request.headers["Authorization"]
|
|
seen["path"] = request.url.path
|
|
seen["params"] = dict(request.url.params)
|
|
return httpx.Response(200, json={"transactions": []})
|
|
|
|
client = QontoClient(
|
|
base_url="https://example.test",
|
|
organization_path="/v2/organization",
|
|
transactions_path="/v2/transactions",
|
|
auth_mode="legacy_api_key",
|
|
timeout_seconds=1,
|
|
max_retries=0,
|
|
credential_provider=provider,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
client.list_transactions(
|
|
iban="DE1234567890",
|
|
page=2,
|
|
page_size=25,
|
|
window_days=14,
|
|
status="completed",
|
|
side="debit",
|
|
)
|
|
|
|
assert seen["auth"] == "binky-user:top-secret"
|
|
assert seen["path"] == "/v2/transactions"
|
|
assert seen["params"] == {
|
|
"iban": "DE1234567890",
|
|
"current_page": "2",
|
|
"per_page": "25",
|
|
"window_days": "14",
|
|
"status": "completed",
|
|
"side": "debit",
|
|
}
|
|
|
|
|
|
def test_client_invalidates_cached_credentials_on_auth_failure() -> None:
|
|
provider = StubCredentialProvider()
|
|
|
|
def handler(_: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(401, json={"error": "unauthorized"})
|
|
|
|
client = QontoClient(
|
|
base_url="https://example.test",
|
|
organization_path="/v2/organization",
|
|
transactions_path="/v2/transactions",
|
|
auth_mode="legacy_api_key",
|
|
timeout_seconds=1,
|
|
max_retries=0,
|
|
credential_provider=provider,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
|
|
try:
|
|
client.get_organization()
|
|
except Exception as exc: # noqa: BLE001
|
|
assert getattr(exc, "error_code", None) == "qonto_auth_failed"
|
|
else:
|
|
raise AssertionError("Expected Qonto auth failure")
|
|
|
|
assert provider.invalidated is True
|