06/01 Add teller test tool
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
# Environment
|
||||
.env
|
||||
.teller_token
|
||||
|
||||
# Python
|
||||
__pycache__/
|
||||
|
||||
@@ -0,0 +1,485 @@
|
||||
"""
|
||||
Teller API Test Script
|
||||
======================
|
||||
Standalone script to verify Teller credentials, API connectivity, and bank
|
||||
enrollment — no Flask app required.
|
||||
|
||||
Modes
|
||||
-----
|
||||
1. Test with an existing access token:
|
||||
python scripts/test_teller.py --token <access_token>
|
||||
|
||||
2. Connect a new bank via Teller Connect (opens browser, captures token):
|
||||
python scripts/test_teller.py --connect
|
||||
python scripts/test_teller.py --connect --port 8765 # custom port
|
||||
|
||||
After completing the bank login in the browser, the script captures the
|
||||
access token and enrollment ID, then automatically runs all API tests.
|
||||
|
||||
3. Target a specific account:
|
||||
python scripts/test_teller.py --token <tok> --account <account_id>
|
||||
|
||||
4. Fetch more history:
|
||||
python scripts/test_teller.py --token <tok> --days 90
|
||||
|
||||
Reads TELLER_APP_ID, TELLER_ENV, TELLER_CERT_PATH, TELLER_KEY_PATH, and
|
||||
TELLER_WEBHOOK_SECRET from .env automatically.
|
||||
Override cert paths with --cert / --key flags if needed.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import date, timedelta
|
||||
|
||||
# ── Load .env from project root ───────────────────────────────────────────────
|
||||
|
||||
_here = os.path.dirname(os.path.abspath(__file__))
|
||||
_root = os.path.dirname(_here)
|
||||
_env = os.path.join(_root, '.env')
|
||||
|
||||
if os.path.exists(_env):
|
||||
with open(_env) as _f:
|
||||
for _line in _f:
|
||||
_line = _line.strip()
|
||||
if _line and not _line.startswith('#') and '=' in _line:
|
||||
_k, _v = _line.split('=', 1)
|
||||
os.environ.setdefault(_k.strip(), _v.strip().strip('"').strip("'"))
|
||||
|
||||
# ── Teller constants ──────────────────────────────────────────────────────────
|
||||
|
||||
TELLER_BASE = 'https://api.teller.io'
|
||||
TELLER_VERSION = '2020-10-12'
|
||||
|
||||
# ── Console helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
def _ok(msg): print(f' \033[32m✓\033[0m {msg}')
|
||||
def _fail(msg): print(f' \033[31m✗\033[0m {msg}')
|
||||
def _info(msg): print(f' \033[34m→\033[0m {msg}')
|
||||
def _warn(msg): print(f' \033[33m!\033[0m {msg}')
|
||||
def _head(msg): print(f'\n\033[1m{msg}\033[0m')
|
||||
|
||||
|
||||
def _pretty(data, indent=4):
|
||||
return json.dumps(data, indent=indent, default=str)
|
||||
|
||||
|
||||
# ── mTLS session ──────────────────────────────────────────────────────────────
|
||||
|
||||
def _build_session(access_token, cert_path, key_path):
|
||||
import requests
|
||||
if not os.path.exists(cert_path):
|
||||
raise FileNotFoundError(f'Cert not found: {cert_path}')
|
||||
if not os.path.exists(key_path):
|
||||
raise FileNotFoundError(f'Key not found: {key_path}')
|
||||
|
||||
s = requests.Session()
|
||||
s.cert = (cert_path, key_path)
|
||||
s.auth = (access_token, '')
|
||||
s.headers.update({'Teller-Version': TELLER_VERSION, 'Accept': 'application/json'})
|
||||
return s
|
||||
|
||||
|
||||
# ── Bank connection via Teller Connect ───────────────────────────────────────
|
||||
|
||||
_CONNECT_HTML = """<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<title>Teller Connect — PFM Test</title>
|
||||
<script src="https://cdn.teller.io/connect/connect.js"></script>
|
||||
<style>
|
||||
body {{ font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', sans-serif;
|
||||
background: #0f172a; color: #f1f5f9; display: flex;
|
||||
align-items: center; justify-content: center; height: 100vh; margin: 0; }}
|
||||
.card {{ background: #1e293b; border-radius: 12px; padding: 40px; text-align: center;
|
||||
max-width: 400px; width: 100%; box-shadow: 0 8px 32px rgba(0,0,0,.4); }}
|
||||
h2 {{ margin: 0 0 8px; font-size: 20px; }}
|
||||
p {{ color: #94a3b8; font-size: 14px; margin: 0 0 28px; }}
|
||||
button {{ background: #10b981; color: #fff; border: none; border-radius: 8px;
|
||||
padding: 12px 28px; font-size: 15px; font-weight: 600; cursor: pointer;
|
||||
width: 100%; transition: background .2s; }}
|
||||
button:hover {{ background: #059669; }}
|
||||
button:disabled {{ background: #334155; color: #64748b; cursor: not-allowed; }}
|
||||
#status {{ margin-top: 16px; font-size: 13px; color: #94a3b8; min-height: 20px; }}
|
||||
.success {{ color: #10b981; }} .error {{ color: #ef4444; }}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="card">
|
||||
<h2>🏦 Connect Your Bank</h2>
|
||||
<p>Sign in to your bank via Teller Connect.<br>
|
||||
Your credentials go directly to your bank — never to this script.</p>
|
||||
<button id="connectBtn" onclick="startConnect()">Connect a Bank Account</button>
|
||||
<div id="status"></div>
|
||||
</div>
|
||||
<script>
|
||||
function setStatus(msg, cls) {{
|
||||
const el = document.getElementById('status');
|
||||
el.textContent = msg;
|
||||
el.className = cls || '';
|
||||
}}
|
||||
|
||||
function startConnect() {{
|
||||
document.getElementById('connectBtn').disabled = true;
|
||||
setStatus('Opening Teller Connect…');
|
||||
|
||||
const teller = TellerConnect.setup({{
|
||||
applicationId: '{app_id}',
|
||||
environment: '{env}',
|
||||
products: ['transactions'],
|
||||
onSuccess: function(enrollment) {{
|
||||
setStatus('✓ Connected! Sending token to test script…', 'success');
|
||||
fetch('/capture', {{
|
||||
method: 'POST',
|
||||
headers: {{'Content-Type': 'application/json'}},
|
||||
body: JSON.stringify({{
|
||||
accessToken: enrollment.accessToken,
|
||||
enrollmentId: enrollment.enrollment.id,
|
||||
institution: enrollment.enrollment.institution.name,
|
||||
}}),
|
||||
}}).then(() => {{
|
||||
setStatus('✓ Token captured. You can close this tab.', 'success');
|
||||
}}).catch(() => {{
|
||||
setStatus('Capture failed — check terminal output.', 'error');
|
||||
}});
|
||||
}},
|
||||
onFailure: function(failure) {{
|
||||
setStatus('Connection failed: ' + (failure.message || JSON.stringify(failure)), 'error');
|
||||
document.getElementById('connectBtn').disabled = false;
|
||||
}},
|
||||
onExit: function() {{
|
||||
if (!document.getElementById('status').classList.contains('success')) {{
|
||||
setStatus('Cancelled.');
|
||||
document.getElementById('connectBtn').disabled = false;
|
||||
}}
|
||||
}},
|
||||
}});
|
||||
teller.open();
|
||||
}}
|
||||
</script>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
|
||||
def connect_bank(cert_path, key_path, port=8765, days=30):
|
||||
"""
|
||||
Spin up a local HTTP server, open Teller Connect in the browser,
|
||||
capture the access token, then run the full test suite.
|
||||
"""
|
||||
import threading
|
||||
import webbrowser
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
app_id = os.environ.get('TELLER_APP_ID', '')
|
||||
env = os.environ.get('TELLER_ENV', 'development')
|
||||
|
||||
if not app_id:
|
||||
_fail('TELLER_APP_ID not set in .env — cannot launch Teller Connect')
|
||||
sys.exit(1)
|
||||
|
||||
captured = {}
|
||||
server_ready = threading.Event()
|
||||
capture_done = threading.Event()
|
||||
|
||||
html = _CONNECT_HTML.format(app_id=app_id, env=env)
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, fmt, *args):
|
||||
pass # silence access log
|
||||
|
||||
def do_GET(self):
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'text/html; charset=utf-8')
|
||||
self.end_headers()
|
||||
self.wfile.write(html.encode())
|
||||
|
||||
def do_POST(self):
|
||||
if self.path == '/capture':
|
||||
length = int(self.headers.get('Content-Length', 0))
|
||||
body = self.rfile.read(length)
|
||||
try:
|
||||
data = json.loads(body)
|
||||
captured.update(data)
|
||||
self.send_response(200)
|
||||
self.send_header('Content-Type', 'application/json')
|
||||
self.end_headers()
|
||||
self.wfile.write(b'{"status":"ok"}')
|
||||
capture_done.set()
|
||||
except Exception:
|
||||
self.send_response(400)
|
||||
self.end_headers()
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
server = HTTPServer(('127.0.0.1', port), Handler)
|
||||
url = f'http://127.0.0.1:{port}'
|
||||
|
||||
def _serve():
|
||||
server_ready.set()
|
||||
server.serve_forever()
|
||||
|
||||
t = threading.Thread(target=_serve, daemon=True)
|
||||
t.start()
|
||||
server_ready.wait()
|
||||
|
||||
_head('Bank Connection — Teller Connect')
|
||||
_info(f'Local server: {url}')
|
||||
_info(f'App ID : {app_id}')
|
||||
_info(f'Environment : {env}')
|
||||
print()
|
||||
print(' Opening browser. Complete the bank login, then return here.')
|
||||
print(' Press Ctrl+C to cancel.\n')
|
||||
|
||||
webbrowser.open(url)
|
||||
|
||||
try:
|
||||
capture_done.wait(timeout=300) # 5-minute window
|
||||
except KeyboardInterrupt:
|
||||
print('\n Cancelled.')
|
||||
sys.exit(0)
|
||||
finally:
|
||||
server.shutdown()
|
||||
|
||||
if not captured.get('accessToken'):
|
||||
_fail('No token received — did you complete the bank login?')
|
||||
sys.exit(1)
|
||||
|
||||
access_token = captured['accessToken']
|
||||
enrollment_id = captured.get('enrollmentId', 'unknown')
|
||||
institution = captured.get('institution', 'unknown')
|
||||
|
||||
print()
|
||||
_ok(f'Bank connected: {institution}')
|
||||
_info(f'Enrollment ID : {enrollment_id}')
|
||||
_info(f'Access token : {access_token[:8]}…{"*" * (len(access_token) - 8)}')
|
||||
|
||||
# Save token to a local file for reuse
|
||||
token_file = os.path.join(_root, '.teller_token')
|
||||
with open(token_file, 'w') as f:
|
||||
json.dump({
|
||||
'access_token': access_token,
|
||||
'enrollment_id': enrollment_id,
|
||||
'institution': institution,
|
||||
'captured_at': date.today().isoformat(),
|
||||
}, f, indent=2)
|
||||
_info(f'Token saved to .teller_token (gitignored) for reuse with --token')
|
||||
|
||||
# Run full test suite with the captured token
|
||||
print()
|
||||
try:
|
||||
session = _build_session(access_token, cert_path, key_path)
|
||||
except FileNotFoundError as e:
|
||||
_fail(str(e))
|
||||
sys.exit(1)
|
||||
|
||||
accounts = test_connectivity(session)
|
||||
account_id = accounts[0]['id'] if accounts else None
|
||||
|
||||
if account_id:
|
||||
test_balances(session, account_id)
|
||||
txns = test_transactions(session, account_id, days=days)
|
||||
if txns:
|
||||
test_transaction_detail(session, account_id, txns[0]['id'])
|
||||
else:
|
||||
_warn('No accounts returned — skipping balance/transaction tests')
|
||||
|
||||
test_webhook_signature(os.environ.get('TELLER_WEBHOOK_SECRET', ''))
|
||||
|
||||
|
||||
# ── Individual API tests ──────────────────────────────────────────────────────
|
||||
|
||||
def test_connectivity(session):
|
||||
_head('1. Connectivity — GET /accounts')
|
||||
try:
|
||||
resp = session.get(f'{TELLER_BASE}/accounts', timeout=15)
|
||||
if resp.status_code == 200:
|
||||
accounts = resp.json()
|
||||
_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:
|
||||
_fail('Unauthorized (401) — check access token and mTLS certs')
|
||||
_info(f'Response: {resp.text[:500]}')
|
||||
else:
|
||||
_fail(f'HTTP {resp.status_code}')
|
||||
_info(f'Response: {resp.text[:500]}')
|
||||
except Exception as e:
|
||||
_fail(f'Request failed: {e}')
|
||||
return []
|
||||
|
||||
|
||||
def test_balances(session, account_id):
|
||||
_head(f'2. Balance — GET /accounts/{account_id}/balances')
|
||||
try:
|
||||
resp = session.get(f'{TELLER_BASE}/accounts/{account_id}/balances', timeout=15)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
_ok(f'Ledger: {data.get("ledger", "n/a")} | Available: {data.get("available", "n/a")}')
|
||||
_info(_pretty(data))
|
||||
return data
|
||||
else:
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_fail(f'Request failed: {e}')
|
||||
return None
|
||||
|
||||
|
||||
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()
|
||||
try:
|
||||
resp = session.get(
|
||||
f'{TELLER_BASE}/accounts/{account_id}/transactions',
|
||||
params={'start_date': start, 'end_date': end},
|
||||
timeout=30,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
txns = resp.json()
|
||||
_ok(f'{len(txns)} transaction(s) from {start} to {end}')
|
||||
for txn in txns[:5]:
|
||||
amount = float(txn.get('amount', 0))
|
||||
sign = '-' if amount > 0 else '+'
|
||||
_info(f'{txn.get("date", "")} {sign}${abs(amount):.2f}'
|
||||
f' [{txn.get("status", "")}] {txn.get("description", "")}')
|
||||
if len(txns) > 5:
|
||||
_info(f'… and {len(txns) - 5} more')
|
||||
return txns
|
||||
else:
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_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}')
|
||||
try:
|
||||
resp = session.get(
|
||||
f'{TELLER_BASE}/accounts/{account_id}/transactions/{txn_id}',
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
_ok('Retrieved successfully')
|
||||
_info(_pretty(data))
|
||||
return data
|
||||
else:
|
||||
_fail(f'HTTP {resp.status_code} — {resp.text[:300]}')
|
||||
except Exception as e:
|
||||
_fail(f'Request failed: {e}')
|
||||
return None
|
||||
|
||||
|
||||
def test_webhook_signature(webhook_secret):
|
||||
_head('5. Webhook HMAC verification (local simulation)')
|
||||
if not webhook_secret:
|
||||
_warn('TELLER_WEBHOOK_SECRET not set — skipping')
|
||||
return
|
||||
|
||||
import hmac
|
||||
import hashlib
|
||||
import time
|
||||
|
||||
body = b'{"type":"transactions.processed","enrollment_id":"test_enroll_123"}'
|
||||
ts = str(int(time.time()))
|
||||
payload = f'{ts}.'.encode() + body
|
||||
sig = hmac.new(webhook_secret.encode(), payload, hashlib.sha256).hexdigest()
|
||||
header = f't={ts},v1={sig}'
|
||||
|
||||
parts = dict(p.split('=', 1) for p in header.split(',') if '=' in p)
|
||||
sigs = [v for p in header.split(',') if p.startswith('v1=') for _, v in [p.split('=', 1)]]
|
||||
recomputed = hmac.new(
|
||||
webhook_secret.encode(),
|
||||
f'{parts["t"]}.'.encode() + body,
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
|
||||
if any(hmac.compare_digest(recomputed, s) for s in sigs):
|
||||
_ok('HMAC-SHA256 signature round-trip passed')
|
||||
_info(f'Example Teller-Signature header: {header}')
|
||||
else:
|
||||
_fail('Signature mismatch — check TELLER_WEBHOOK_SECRET')
|
||||
|
||||
|
||||
# ── Entry point ───────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description='Test Teller API — use --connect to enroll a bank, '
|
||||
'or --token to test an existing enrollment.',
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
)
|
||||
|
||||
mode = parser.add_mutually_exclusive_group(required=True)
|
||||
mode.add_argument('--connect', action='store_true',
|
||||
help='Open Teller Connect in browser to enroll a bank, then run tests')
|
||||
mode.add_argument('--token', metavar='ACCESS_TOKEN',
|
||||
help='Test using an existing Teller access token')
|
||||
|
||||
parser.add_argument('--account', default=None,
|
||||
help='Teller account ID (optional; defaults to first account)')
|
||||
parser.add_argument('--cert', default=os.environ.get('TELLER_CERT_PATH', ''),
|
||||
help='Path to certificate.pem (default: TELLER_CERT_PATH from .env)')
|
||||
parser.add_argument('--key', default=os.environ.get('TELLER_KEY_PATH', ''),
|
||||
help='Path to private_key.pem (default: TELLER_KEY_PATH from .env)')
|
||||
parser.add_argument('--days', type=int, default=30,
|
||||
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)')
|
||||
args = parser.parse_args()
|
||||
|
||||
cert = args.cert
|
||||
key = args.key
|
||||
|
||||
if not cert or not key:
|
||||
_fail('Cert/key paths not found.\n'
|
||||
' Set TELLER_CERT_PATH / TELLER_KEY_PATH in .env, '
|
||||
'or pass --cert / --key flags.')
|
||||
sys.exit(1)
|
||||
|
||||
if args.connect:
|
||||
connect_bank(cert, key, port=args.port, days=args.days)
|
||||
print('\n\033[1mDone.\033[0m\n')
|
||||
return
|
||||
|
||||
# ── --token mode ──────────────────────────────────────────────────────────
|
||||
print(f'\n\033[1mTeller API Test\033[0m')
|
||||
print(f' Base URL : {TELLER_BASE}')
|
||||
print(f' Cert : {cert}')
|
||||
print(f' Key : {key}')
|
||||
print(f' Token : {args.token[:8]}…{"*" * max(0, len(args.token) - 8)}')
|
||||
|
||||
try:
|
||||
session = _build_session(args.token, cert, key)
|
||||
except FileNotFoundError as e:
|
||||
_fail(str(e))
|
||||
sys.exit(1)
|
||||
|
||||
accounts = test_connectivity(session)
|
||||
account_id = args.account
|
||||
if not account_id and accounts:
|
||||
account_id = accounts[0]['id']
|
||||
_info(f'No --account given; using first account: {account_id}')
|
||||
|
||||
if account_id:
|
||||
test_balances(session, account_id)
|
||||
txns = test_transactions(session, account_id, days=args.days)
|
||||
if txns:
|
||||
test_transaction_detail(session, account_id, txns[0]['id'])
|
||||
else:
|
||||
_warn('No account ID available — skipping balance/transaction tests')
|
||||
|
||||
test_webhook_signature(os.environ.get('TELLER_WEBHOOK_SECRET', ''))
|
||||
|
||||
print('\n\033[1mDone.\033[0m\n')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user