162 lines
9.7 KiB
Python
162 lines
9.7 KiB
Python
import tempfile
|
|
import unittest
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from datetime import date
|
|
from pathlib import Path
|
|
from fastapi.testclient import TestClient
|
|
from server.main import create_app
|
|
|
|
|
|
class MoneyApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp=tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.path=Path(self.temp.name)/'test.db'
|
|
self.app=create_app(self.path,start_worker=False)
|
|
self.client=TestClient(self.app)
|
|
self.addCleanup(self.client.close)
|
|
response=self.client.post('/api/auth/register',json={'name':'Money Test','email':'money@example.com','password':'long-password-123'})
|
|
self.headers={'X-CSRF-Token':response.json()['csrfToken']}
|
|
|
|
def payload(self):
|
|
return {'name':'Money SIM','provider':'Example','type':'SIM','country':'US','activatedAt':'2026-01-01','balances':[{'currency':'USD','amount':'0.30'},{'currency':'JPY','amount':'1000'}],'rules':[{'action':'sms','interval':90,'unit':'days','anchor':'completion','dueDate':date.today().isoformat(),'cost':{'currency':'USD','amount':'0.10'}}]}
|
|
|
|
def create_card(self):
|
|
response=self.client.post('/api/cards',headers=self.headers,json=self.payload())
|
|
self.assertEqual(response.status_code,201,response.text)
|
|
return response.json()['workspace']['cards'][0]
|
|
|
|
def complete(self,card,request_id='money-operation',**extra):
|
|
rule=card['rules'][0]
|
|
payload={'completedAt':date.today().isoformat(),'note':'','requestId':request_id,'version':rule['version'],**extra}
|
|
return self.client.post(f"/api/rules/{rule['id']}/completions",headers=self.headers,json=payload)
|
|
|
|
def test_multiple_balances_and_rule_cost_persist_after_restart(self):
|
|
card=self.create_card()
|
|
with TestClient(create_app(self.path,start_worker=False)) as client:
|
|
client.post('/api/auth/login',json={'email':'money@example.com','password':'long-password-123'})
|
|
persisted=client.get('/api/workspace').json()['workspace']['cards'][0]
|
|
self.assertEqual(persisted['balances'],card['balances'])
|
|
self.assertEqual(persisted['rules'][0]['cost'],{'currency':'USD','amount':'0.10'})
|
|
|
|
def test_explicit_deduction_and_duplicate_submit_are_atomic(self):
|
|
card=self.create_card()
|
|
payload={'balanceAction':'deduct','cardVersion':card['version']}
|
|
first=self.complete(card,**payload)
|
|
self.assertEqual(first.status_code,200,first.text)
|
|
retry=self.complete(card,**payload)
|
|
self.assertEqual(retry.status_code,200,retry.text)
|
|
workspace=retry.json()['workspace']
|
|
self.assertEqual(workspace['cards'][0]['balances'],[{'currency':'USD','amount':'0.20'},{'currency':'JPY','amount':'1000'}])
|
|
self.assertEqual(len(workspace['events']),1)
|
|
event=workspace['events'][0]
|
|
self.assertEqual(event['cost'],{'currency':'USD','amount':'0.10'})
|
|
self.assertEqual(event['balanceBefore']['amount'],'0.30')
|
|
self.assertEqual(event['balanceAfter']['amount'],'0.20')
|
|
|
|
def test_record_only_keeps_balance_and_historical_actual_cost(self):
|
|
card=self.create_card()
|
|
response=self.complete(card,cost={'currency':'GBP','amount':'2.5'})
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
workspace=response.json()['workspace']
|
|
self.assertEqual(workspace['cards'][0]['balances'],card['balances'])
|
|
self.assertEqual(workspace['events'][0]['cost'],{'currency':'GBP','amount':'2.50'})
|
|
rule=workspace['cards'][0]['rules'][0]
|
|
edited={k:rule[k] for k in ['action','interval','unit','anchor','dueDate','instructions','version']}
|
|
edited['cost']={'currency':'USD','amount':'5.00'}
|
|
updated=self.client.put(f"/api/rules/{rule['id']}",headers=self.headers,json=edited)
|
|
self.assertEqual(updated.status_code,200,updated.text)
|
|
self.assertEqual(updated.json()['workspace']['events'][0]['cost'],{'currency':'GBP','amount':'2.50'})
|
|
|
|
def test_changed_retry_cannot_silently_discard_cost_or_balance_action(self):
|
|
card=self.create_card()
|
|
first=self.complete(card,cost={'currency':'USD','amount':'0.10'})
|
|
self.assertEqual(first.status_code,200,first.text)
|
|
for changed in [
|
|
{'cost':{'currency':'USD','amount':'0.20'}},
|
|
{'cost':{'currency':'USD','amount':'0.10'},'balanceAction':'deduct','cardVersion':card['version']},
|
|
{'cost':None},
|
|
{'note':'changed after a lost response'},
|
|
]:
|
|
with self.subTest(changed=changed):
|
|
retry=self.complete(card,**changed)
|
|
self.assertEqual(retry.status_code,409,retry.text)
|
|
retry=self.complete(card,cost={'currency':'USD','amount':'0.100'})
|
|
self.assertEqual(retry.status_code,200,retry.text)
|
|
workspace=retry.json()['workspace']
|
|
self.assertEqual(workspace['cards'][0]['balances'],card['balances'])
|
|
self.assertEqual(len(workspace['events']),1)
|
|
self.assertEqual(workspace['events'][0]['cost'],{'currency':'USD','amount':'0.10'})
|
|
|
|
def test_insufficient_balance_rejects_entire_completion(self):
|
|
card=self.create_card()
|
|
response=self.complete(card,cost={'currency':'USD','amount':'0.31'},balanceAction='deduct',cardVersion=card['version'])
|
|
self.assertEqual(response.status_code,422,response.text)
|
|
workspace=self.client.get('/api/workspace').json()['workspace']
|
|
self.assertEqual(workspace['cards'][0],card)
|
|
self.assertEqual(workspace['events'],[])
|
|
|
|
def test_stale_card_balance_version_rejects_deduction(self):
|
|
card=self.create_card()
|
|
changed={**self.payload(),'rules':[],'version':card['version'],'balances':[{'currency':'USD','amount':'8.00'}]}
|
|
response=self.client.put(f"/api/cards/{card['id']}",headers=self.headers,json=changed)
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
self.assertEqual(self.complete(card,balanceAction='deduct',cardVersion=card['version']).status_code,409)
|
|
|
|
def test_duplicate_currency_and_invalid_precision_are_rejected(self):
|
|
payload=self.payload()
|
|
payload['balances'].append({'currency':'USD','amount':'9.00'})
|
|
self.assertEqual(self.client.post('/api/cards',headers=self.headers,json=payload).status_code,422)
|
|
payload=self.payload()
|
|
payload['rules'][0]['cost']['amount']='0.001'
|
|
self.assertEqual(self.client.post('/api/cards',headers=self.headers,json=payload).status_code,422)
|
|
|
|
def test_legacy_card_edits_preserve_existing_money(self):
|
|
card=self.create_card()
|
|
changed={k:v for k,v in self.payload().items() if k not in ('balances','rules')}
|
|
changed.update(name='Renamed',version=card['version'])
|
|
response=self.client.put(f"/api/cards/{card['id']}",headers=self.headers,json=changed)
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
updated=response.json()['workspace']['cards'][0]
|
|
self.assertEqual(updated['balances'],card['balances'])
|
|
rule=updated['rules'][0]
|
|
fields={k:rule[k] for k in ['action','interval','unit','anchor','dueDate','instructions','version']}
|
|
response=self.client.put(f"/api/rules/{rule['id']}",headers=self.headers,json=fields)
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
self.assertEqual(response.json()['workspace']['cards'][0]['rules'][0]['cost'],card['rules'][0]['cost'])
|
|
|
|
def test_topup_creates_new_currency_and_saves_balance_snapshot(self):
|
|
payload=self.payload()
|
|
payload['rules'][0].update(action='topup',cost={'currency':'HKD','amount':'10.50'})
|
|
response=self.client.post('/api/cards',headers=self.headers,json=payload)
|
|
self.assertEqual(response.status_code,201,response.text)
|
|
card=response.json()['workspace']['cards'][0]
|
|
response=self.complete(card,balanceAction='credit',cardVersion=card['version'])
|
|
self.assertEqual(response.status_code,200,response.text)
|
|
workspace=response.json()['workspace']
|
|
self.assertEqual(workspace['cards'][0]['balances'][-1],{'currency':'HKD','amount':'10.50'})
|
|
self.assertEqual(workspace['events'][0]['balanceBefore'],{'currency':'HKD','amount':'0.00'})
|
|
|
|
def test_concurrent_requests_cannot_deduct_twice(self):
|
|
card=self.create_card()
|
|
def submit(request_id):
|
|
return self.complete(card,request_id,balanceAction='deduct',cardVersion=card['version']).status_code
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
statuses=list(pool.map(submit,['money-concurrent-one','money-concurrent-two']))
|
|
self.assertEqual(sorted(statuses),[200,409])
|
|
workspace=self.client.get('/api/workspace').json()['workspace']
|
|
self.assertEqual(workspace['cards'][0]['balances'][0]['amount'],'0.20')
|
|
self.assertEqual(len(workspace['events']),1)
|
|
|
|
def test_other_account_cannot_change_or_deduct_balance(self):
|
|
card=self.create_card()
|
|
with TestClient(self.app) as other:
|
|
response=other.post('/api/auth/register',json={'name':'Other','email':'other-money@example.com','password':'long-password-123'})
|
|
headers={'X-CSRF-Token':response.json()['csrfToken']}
|
|
payload={**self.payload(),'rules':[],'version':card['version']}
|
|
self.assertEqual(other.put(f"/api/cards/{card['id']}",headers=headers,json=payload).status_code,404)
|
|
rule=card['rules'][0]
|
|
response=other.post(f"/api/rules/{rule['id']}/completions",headers=headers,json={'completedAt':date.today().isoformat(),'requestId':'foreign-deduction','version':rule['version'],'balanceAction':'deduct','cardVersion':card['version']})
|
|
self.assertEqual(response.status_code,404)
|
|
self.assertEqual(self.client.get('/api/workspace').json()['workspace']['cards'][0],card)
|