61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""One transaction, one insights call, assert 200s -- phase-0-base-project-spec.md §12."""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import urllib.request
|
||
|
|
|
||
|
|
import psycopg
|
||
|
|
|
||
|
|
GATEWAY_URL = os.environ.get("GATEWAY_URL", "http://localhost:8000")
|
||
|
|
DSN = os.environ.get("POSTGRES_DSN", "postgresql://ledgerline:ledgerline@localhost:5432/ledgerline")
|
||
|
|
|
||
|
|
|
||
|
|
def _get(url):
|
||
|
|
with urllib.request.urlopen(url, timeout=10) as resp:
|
||
|
|
return resp.status, json.loads(resp.read())
|
||
|
|
|
||
|
|
|
||
|
|
def _post(url, payload):
|
||
|
|
data = json.dumps(payload).encode()
|
||
|
|
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST")
|
||
|
|
try:
|
||
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||
|
|
return resp.status, json.loads(resp.read())
|
||
|
|
except urllib.error.HTTPError as exc:
|
||
|
|
return exc.code, json.loads(exc.read())
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
with psycopg.connect(DSN, autocommit=True) as conn:
|
||
|
|
with conn.cursor() as cur:
|
||
|
|
cur.execute("SELECT data FROM seed_manifest LIMIT 1")
|
||
|
|
row = cur.fetchone()
|
||
|
|
if not row:
|
||
|
|
print("smoke: no seed manifest found -- run `make seed` first", file=sys.stderr)
|
||
|
|
sys.exit(1)
|
||
|
|
account_id = row[0]["clean_account_id"]
|
||
|
|
|
||
|
|
status, body = _post(
|
||
|
|
f"{GATEWAY_URL}/api/transactions",
|
||
|
|
{
|
||
|
|
"account_id": account_id,
|
||
|
|
"kind": "deposit",
|
||
|
|
"amount_minor": 500,
|
||
|
|
"currency": "EUR",
|
||
|
|
"merchant_name": "Smoke Test Merchant",
|
||
|
|
},
|
||
|
|
)
|
||
|
|
assert status == 201, f"transaction failed: {status} {body}"
|
||
|
|
print(f"transaction ok: {body['transaction_id']} status={body['status']}")
|
||
|
|
|
||
|
|
status, body = _get(f"{GATEWAY_URL}/api/accounts/{account_id}/insights")
|
||
|
|
assert status == 200, f"insights failed: {status} {body}"
|
||
|
|
print(f"insights ok: {body}")
|
||
|
|
|
||
|
|
print("smoke test passed")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|