From 88cf1253fdc85d7dc1e49e20c6bdfac724934767 Mon Sep 17 00:00:00 2001 From: Nguyen Ngo Date: Mon, 1 Jun 2026 20:59:59 -0400 Subject: [PATCH] Update test_teller.py --- scripts/test_teller.py | 80 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 66 insertions(+), 14 deletions(-) diff --git a/scripts/test_teller.py b/scripts/test_teller.py index 7975e6d..0be89de 100644 --- a/scripts/test_teller.py +++ b/scripts/test_teller.py @@ -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