Update test_teller.py
This commit is contained in:
+66
-14
@@ -31,7 +31,7 @@ import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import date, timedelta
|
||||
from datetime import date, datetime, timedelta
|
||||
|
||||
# ── Load .env from project root ───────────────────────────────────────────────
|
||||
|
||||
@@ -52,6 +52,25 @@ if os.path.exists(_env):
|
||||
TELLER_BASE = 'https://api.teller.io'
|
||||
TELLER_VERSION = '2020-10-12'
|
||||
|
||||
# ── Logger ────────────────────────────────────────────────────────────────────
|
||||
|
||||
_log_file = None # set in main() when --log-file is given
|
||||
|
||||
def _log(label, data, status_code=None, url=None):
|
||||
"""Append a structured JSON entry to the log file if one is configured."""
|
||||
if _log_file is None:
|
||||
return
|
||||
entry = {
|
||||
'timestamp': datetime.now().isoformat(timespec='seconds'),
|
||||
'label': label,
|
||||
'url': url,
|
||||
'status': status_code,
|
||||
'response': data,
|
||||
}
|
||||
with open(_log_file, 'a', encoding='utf-8') as f:
|
||||
f.write(json.dumps(entry, default=str) + '\n')
|
||||
|
||||
|
||||
# ── Console helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def _ok(msg): print(f' \033[32m✓\033[0m {msg}')
|
||||
@@ -164,7 +183,7 @@ function startConnect() {{
|
||||
"""
|
||||
|
||||
|
||||
def connect_bank(cert_path, key_path, port=8765, days=30):
|
||||
def connect_bank(cert_path, key_path, port=8765, days=30, log_path=None):
|
||||
"""
|
||||
Spin up a local HTTP server, open Teller Connect in the browser,
|
||||
capture the access token, then run the full test suite.
|
||||
@@ -256,6 +275,11 @@ def connect_bank(cert_path, key_path, port=8765, days=30):
|
||||
_ok(f'Bank connected: {institution}')
|
||||
_info(f'Enrollment ID : {enrollment_id}')
|
||||
_info(f'Access token : {access_token[:8]}…{"*" * (len(access_token) - 8)}')
|
||||
_log('teller_connect_enrollment', {
|
||||
'enrollment_id': enrollment_id,
|
||||
'institution': institution,
|
||||
'access_token': f'{access_token[:8]}…(redacted)',
|
||||
})
|
||||
|
||||
# Save token to a local file for reuse
|
||||
token_file = os.path.join(_root, '.teller_token')
|
||||
@@ -269,6 +293,10 @@ def connect_bank(cert_path, key_path, port=8765, days=30):
|
||||
_info(f'Token saved to .teller_token (gitignored) for reuse with --token')
|
||||
|
||||
# Run full test suite with the captured token
|
||||
global _log_file
|
||||
if log_path:
|
||||
_log_file = log_path
|
||||
|
||||
print()
|
||||
try:
|
||||
session = _build_session(access_token, cert_path, key_path)
|
||||
@@ -294,38 +322,47 @@ def connect_bank(cert_path, key_path, port=8765, days=30):
|
||||
|
||||
def test_connectivity(session):
|
||||
_head('1. Connectivity — GET /accounts')
|
||||
url = f'{TELLER_BASE}/accounts'
|
||||
try:
|
||||
resp = session.get(f'{TELLER_BASE}/accounts', timeout=15)
|
||||
resp = session.get(url, timeout=15)
|
||||
if resp.status_code == 200:
|
||||
accounts = resp.json()
|
||||
_log('get_accounts', accounts, resp.status_code, url)
|
||||
_ok(f'Connected — {len(accounts)} account(s) returned')
|
||||
for a in accounts:
|
||||
_info(f'{a["id"]} {a.get("institution", {}).get("name", "?")} '
|
||||
f'{a.get("name", "")} ({a.get("type", "")} / {a.get("subtype", "")})')
|
||||
return accounts
|
||||
elif resp.status_code == 401:
|
||||
_log('get_accounts', resp.text, resp.status_code, url)
|
||||
_fail('Unauthorized (401) — check access token and mTLS certs')
|
||||
_info(f'Response: {resp.text[:500]}')
|
||||
else:
|
||||
_log('get_accounts', resp.text, resp.status_code, url)
|
||||
_fail(f'HTTP {resp.status_code}')
|
||||
_info(f'Response: {resp.text[:500]}')
|
||||
except Exception as e:
|
||||
_log('get_accounts', str(e), None, url)
|
||||
_fail(f'Request failed: {e}')
|
||||
return []
|
||||
|
||||
|
||||
def test_balances(session, account_id):
|
||||
_head(f'2. Balance — GET /accounts/{account_id}/balances')
|
||||
url = f'{TELLER_BASE}/accounts/{account_id}/balances'
|
||||
try:
|
||||
resp = session.get(f'{TELLER_BASE}/accounts/{account_id}/balances', timeout=15)
|
||||
resp = session.get(url, timeout=15)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
_log('get_balances', data, resp.status_code, url)
|
||||
_ok(f'Ledger: {data.get("ledger", "n/a")} | Available: {data.get("available", "n/a")}')
|
||||
_info(_pretty(data))
|
||||
return data
|
||||
else:
|
||||
_log('get_balances', resp.text, resp.status_code, url)
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_log('get_balances', str(e), None, url)
|
||||
_fail(f'Request failed: {e}')
|
||||
return None
|
||||
|
||||
@@ -334,14 +371,12 @@ def test_transactions(session, account_id, days=30):
|
||||
_head(f'3. Transactions — GET /accounts/{account_id}/transactions (last {days} days)')
|
||||
start = (date.today() - timedelta(days=days)).isoformat()
|
||||
end = date.today().isoformat()
|
||||
url = f'{TELLER_BASE}/accounts/{account_id}/transactions'
|
||||
try:
|
||||
resp = session.get(
|
||||
f'{TELLER_BASE}/accounts/{account_id}/transactions',
|
||||
params={'start_date': start, 'end_date': end},
|
||||
timeout=30,
|
||||
)
|
||||
resp = session.get(url, params={'start_date': start, 'end_date': end}, timeout=30)
|
||||
if resp.status_code == 200:
|
||||
txns = resp.json()
|
||||
_log('get_transactions', txns, resp.status_code, url)
|
||||
_ok(f'{len(txns)} transaction(s) from {start} to {end}')
|
||||
for txn in txns[:5]:
|
||||
amount = float(txn.get('amount', 0))
|
||||
@@ -352,27 +387,30 @@ def test_transactions(session, account_id, days=30):
|
||||
_info(f'… and {len(txns) - 5} more')
|
||||
return txns
|
||||
else:
|
||||
_log('get_transactions', resp.text, resp.status_code, url)
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_log('get_transactions', str(e), None, url)
|
||||
_fail(f'Request failed: {e}')
|
||||
return []
|
||||
|
||||
|
||||
def test_transaction_detail(session, account_id, txn_id):
|
||||
_head(f'4. Transaction detail — GET /accounts/{account_id}/transactions/{txn_id}')
|
||||
url = f'{TELLER_BASE}/accounts/{account_id}/transactions/{txn_id}'
|
||||
try:
|
||||
resp = session.get(
|
||||
f'{TELLER_BASE}/accounts/{account_id}/transactions/{txn_id}',
|
||||
timeout=15,
|
||||
)
|
||||
resp = session.get(url, timeout=15)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
_log('get_transaction_detail', data, resp.status_code, url)
|
||||
_ok('Retrieved successfully')
|
||||
_info(_pretty(data))
|
||||
return data
|
||||
else:
|
||||
_log('get_transaction_detail', resp.text, resp.status_code, url)
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_log('get_transaction_detail', str(e), None, url)
|
||||
_fail(f'Request failed: {e}')
|
||||
return None
|
||||
|
||||
@@ -433,11 +471,25 @@ def main():
|
||||
help='Days of transaction history to fetch (default: 30)')
|
||||
parser.add_argument('--port', type=int, default=8765,
|
||||
help='Local server port for --connect mode (default: 8765)')
|
||||
parser.add_argument('--log-file', metavar='PATH', default=None,
|
||||
help='Append all API responses as JSON lines to this file '
|
||||
'(default: logs/teller_test_YYYYMMDD_HHMMSS.log)')
|
||||
args = parser.parse_args()
|
||||
|
||||
cert = args.cert
|
||||
key = args.key
|
||||
|
||||
# ── Set up log file ───────────────────────────────────────────────────────
|
||||
global _log_file
|
||||
log_path = args.log_file
|
||||
if log_path is None:
|
||||
log_dir = os.path.join(_root, 'logs')
|
||||
os.makedirs(log_dir, exist_ok=True)
|
||||
stamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
log_path = os.path.join(log_dir, f'teller_test_{stamp}.log')
|
||||
_log_file = log_path
|
||||
_info(f'Logging responses to: {_log_file}')
|
||||
|
||||
if not cert or not key:
|
||||
_fail('Cert/key paths not found.\n'
|
||||
' Set TELLER_CERT_PATH / TELLER_KEY_PATH in .env, '
|
||||
@@ -445,7 +497,7 @@ def main():
|
||||
sys.exit(1)
|
||||
|
||||
if args.connect:
|
||||
connect_bank(cert, key, port=args.port, days=args.days)
|
||||
connect_bank(cert, key, port=args.port, days=args.days, log_path=log_path)
|
||||
print('\n\033[1mDone.\033[0m\n')
|
||||
return
|
||||
|
||||
|
||||
Reference in New Issue
Block a user