212 lines
12 KiB
Python
212 lines
12 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'app'))
|
|
from fastapi.testclient import TestClient
|
|
from database import connect, initialize
|
|
from main import app
|
|
from models import decimal_string
|
|
from services.income_service import dashboard
|
|
|
|
|
|
class APITests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.token = secrets.token_urlsafe(32)
|
|
self.env = patch.dict(os.environ, {'FINANCE_DB_PATH': str(Path(self.temp.name) / 'finance.db'),
|
|
'FINANCE_API_TOKEN': self.token})
|
|
self.env.start()
|
|
self.client = TestClient(app)
|
|
self.client.__enter__()
|
|
self.headers = {'Authorization': f'Bearer {self.token}'}
|
|
with connect() as db:
|
|
self.asset_id = db.execute("SELECT id FROM assets WHERE name='Enbridge'").fetchone()[0]
|
|
|
|
def tearDown(self):
|
|
self.client.__exit__(None, None, None)
|
|
self.env.stop()
|
|
self.temp.cleanup()
|
|
|
|
def request(self, method, path, **kwargs):
|
|
return self.client.request(method, '/api/v1' + path, headers=self.headers, **kwargs)
|
|
|
|
def create_income(self, **changes):
|
|
data = dict(date=date.today().isoformat(), asset_id=self.asset_id, category='dividend',
|
|
amount='0.04', note='Enbridge', expected=False, received=True)
|
|
data.update(changes)
|
|
response = self.request('POST', '/income', json=data)
|
|
self.assertEqual(response.status_code, 201, response.text)
|
|
return response.json()
|
|
|
|
def test_all_routes_require_auth(self):
|
|
routes = [('GET','/assets'), ('POST','/assets'), ('GET','/assets/1'), ('PATCH','/assets/1'), ('DELETE','/assets/1'),
|
|
('GET','/income'), ('POST','/income'), ('GET','/income/1'), ('PATCH','/income/1'), ('DELETE','/income/1'),
|
|
('GET','/stats/summary'), ('GET','/stats/monthly'), ('GET','/stats/by-asset'), ('GET','/stats/by-category'), ('GET','/meta')]
|
|
for method, path in routes:
|
|
for headers in [{}, {'Authorization':'Bearer incorrect'}, {'Authorization':'Basic incorrect'}]:
|
|
response = self.client.request(method, '/api/v1'+path, headers=headers)
|
|
self.assertEqual(response.status_code, 401, (method,path,response.text))
|
|
self.assertEqual(response.headers['www-authenticate'], 'Bearer')
|
|
self.assertEqual(self.request('GET', '/assets').status_code, 200)
|
|
self.assertEqual(self.client.get('/api/v1/assets?token='+self.token).status_code, 401)
|
|
|
|
def test_unconfigured_api_keeps_web_available(self):
|
|
for value in ['', ' ', 'change-me']:
|
|
with patch.dict(os.environ, {'FINANCE_API_TOKEN':value}):
|
|
self.assertEqual(self.request('GET','/assets').status_code,503)
|
|
self.assertEqual(self.client.get('/').status_code,200)
|
|
self.assertEqual(self.client.get('/health').json(), {'status':'ok'})
|
|
with patch.dict(os.environ):
|
|
del os.environ['FINANCE_API_TOKEN']
|
|
self.assertEqual(self.client.get('/api/v1/meta').status_code,503)
|
|
|
|
def test_assets_crud_duplicate_and_partial_updates(self):
|
|
data = dict(name='Test Position', ticker='TP', asset_type='etf', active=True)
|
|
response = self.request('POST','/assets',json=data)
|
|
self.assertEqual(response.status_code,201)
|
|
asset = response.json()
|
|
self.assertNotIn('normalized_name',asset)
|
|
self.assertEqual(self.request('GET',f"/assets/{asset['id']}").json(),asset)
|
|
self.assertEqual(self.request('POST','/assets',json={**data,'name':' TEST POSITION '}).status_code,409)
|
|
self.assertEqual(self.request('POST','/assets',json={**data,'name':'MSC'}).status_code,409)
|
|
result = self.request('PATCH',f"/assets/{asset['id']}",json={'ticker':None,'active':False}).json()
|
|
self.assertEqual(result['name'],asset['name'])
|
|
self.assertIsNone(result['ticker'])
|
|
self.assertFalse(result['active'])
|
|
self.assertIn(result,self.request('GET','/assets?active=false').json())
|
|
self.assertNotIn(result,self.request('GET','/assets?active=true').json())
|
|
self.assertEqual(self.request('PATCH',f"/assets/{asset['id']}",json={'name':'Enbridge'}).status_code,409)
|
|
self.assertEqual(self.request('DELETE',f"/assets/{asset['id']}").status_code,204)
|
|
|
|
def test_asset_soft_delete_preserves_history_and_renames(self):
|
|
entry = self.create_income()
|
|
self.assertEqual(self.request('DELETE',f'/assets/{self.asset_id}').content,b'')
|
|
self.assertFalse(self.request('GET',f'/assets/{self.asset_id}').json()['active'])
|
|
self.assertEqual(self.request('GET',f"/income/{entry['id']}").status_code,200)
|
|
self.assertEqual(self.request('PATCH',f"/income/{entry['id']}",json={'note':'Korrektur'}).status_code,200)
|
|
self.assertEqual(self.request('POST','/income',json=dict(date='2026-09-02',asset_id=self.asset_id,category='dividend',amount='1')).status_code,422)
|
|
self.request('PATCH',f'/assets/{self.asset_id}',json={'name':'Enbridge umbenannt'})
|
|
initialize()
|
|
names = [asset['name'] for asset in self.request('GET','/assets').json()]
|
|
self.assertIn('Enbridge umbenannt',names)
|
|
self.assertNotIn('Enbridge',names)
|
|
|
|
def test_income_crud_partial_and_decimal(self):
|
|
entry = self.create_income(amount='0,04')
|
|
self.assertEqual(entry['amount'],'0.04')
|
|
self.assertEqual(entry['asset'],'Enbridge')
|
|
self.assertEqual(self.request('GET',f"/income/{entry['id']}").json(),entry)
|
|
changed = self.request('PATCH',f"/income/{entry['id']}",json={'amount':'28.00','note':None}).json()
|
|
self.assertEqual(changed['date'],entry['date'])
|
|
self.assertEqual(changed['amount'],'28.00')
|
|
self.assertTrue(changed['received'])
|
|
self.assertFalse(changed['expected'])
|
|
self.assertIsNone(changed['note'])
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT amount FROM income_entries WHERE id=?',(entry['id'],)).fetchone()[0],2800)
|
|
response = self.request('DELETE',f"/income/{entry['id']}")
|
|
self.assertEqual(response.status_code,204)
|
|
self.assertEqual(response.content,b'')
|
|
self.assertEqual(self.request('GET',f"/income/{entry['id']}").status_code,404)
|
|
|
|
def test_filters_and_pagination(self):
|
|
first = self.create_income(date='2025-09-02')
|
|
second = self.create_income(date='2026-09-02',received=False,expected=True,category='interest')
|
|
self.assertEqual(self.request('GET','/income?limit=1&offset=1').json()[0]['id'],first['id'])
|
|
result = self.request('GET',f'/income?year=2026&month=9&asset_id={self.asset_id}&category=interest&received=false&expected=true').json()
|
|
self.assertEqual([row['id'] for row in result],[second['id']])
|
|
self.assertEqual(len(self.request('GET','/income?received=true&expected=false').json()),1)
|
|
for query in ['limit=1001','limit=0','offset=-1','month=13','year=0','received=unknown','category=invalid']:
|
|
self.assertEqual(self.request('GET','/income?'+query).status_code,422)
|
|
|
|
def test_stats_share_dashboard_rules(self):
|
|
today = date.today()
|
|
self.create_income(amount='0.10')
|
|
self.create_income(amount='0.20',category='distribution')
|
|
self.create_income(amount='0.30',category='interest')
|
|
self.create_income(amount='0.40',category='other')
|
|
self.create_income(date=f'{today.year-1}-{today.month:02}-01',amount='0.50')
|
|
self.create_income(amount='28.00',expected=True,received=False)
|
|
stats = dashboard()
|
|
result = self.request('GET','/stats/summary').json()
|
|
self.assertEqual(result['current_month'],'1.00')
|
|
self.assertEqual(result['current_year'],'1.00')
|
|
self.assertEqual(result['previous_year'],'0.50')
|
|
self.assertEqual(result['current_month_yoy_percent'],'100.00')
|
|
self.assertEqual(result['current_year_yoy_percent'],'100.00')
|
|
self.assertEqual(result['all_time'],decimal_string(stats['all_time']))
|
|
self.assertEqual(result['current_year_payment_count'],4)
|
|
monthly = self.request('GET',f'/stats/monthly?year={today.year}').json()[0]
|
|
self.assertEqual(len(monthly['months']),12)
|
|
self.assertEqual(monthly['months'][today.month-1]['amount'],'1.00')
|
|
shares = self.request('GET','/stats/by-asset').json()
|
|
self.assertEqual(shares[0]['amount'],'1.50')
|
|
self.assertEqual(shares[0]['percentage'],'100.00')
|
|
kinds = self.request('GET','/stats/by-category').json()
|
|
self.assertEqual({row['category'] for row in kinds},{'dividend','interest','distribution','other'})
|
|
self.assertEqual(sum(Decimal(row['amount']) for row in kinds),Decimal('1.50'))
|
|
|
|
def test_empty_stats_and_zero_denominator(self):
|
|
result = self.request('GET','/stats/summary').json()
|
|
self.assertEqual(result['all_time'],'0.00')
|
|
self.assertIsNone(result['current_year_yoy_percent'])
|
|
self.assertEqual(self.request('GET','/stats/by-asset').json(),[])
|
|
self.assertTrue(all(row['percentage'] is None for row in self.request('GET','/stats/by-category').json()))
|
|
self.assertEqual(self.request('GET','/stats/monthly?year=2024').json()[0]['total'],'0.00')
|
|
|
|
def test_validation_and_unknown_ids(self):
|
|
base = dict(date='2026-09-02',asset_id=self.asset_id,category='dividend',amount='0.04')
|
|
for change in [{'amount':0.04},{'amount':'0.001'},{'amount':'NaN'},{'amount':'Infinity'},{'date':'2026-02-30'},
|
|
{'asset_id':99999},{'asset_id':True},{'category':'invalid'},{'received':None},{'received':'false'}, {'surprise':'field'}]:
|
|
response = self.request('POST','/income',json={**base,**change})
|
|
self.assertEqual(response.status_code,422,response.text)
|
|
for change in [{'name':' '},{'name':'!!!'},{'name':None},{'asset_type':'invalid'}]:
|
|
self.assertEqual(self.request('POST','/assets',json={'name':'Valid','asset_type':'stock',**change}).status_code,422)
|
|
entry = self.create_income()
|
|
for change in [{'amount':None},{'date':None},{'received':None}]:
|
|
self.assertEqual(self.request('PATCH',f"/income/{entry['id']}",json=change).status_code,422)
|
|
for resource in ['assets','income']:
|
|
for method in ['GET','PATCH','DELETE']:
|
|
kwargs = {'json':{}} if method=='PATCH' else {}
|
|
self.assertEqual(self.request(method,f'/{resource}/999999',**kwargs).status_code,404)
|
|
|
|
def test_web_and_api_use_same_entries(self):
|
|
response = self.client.post('/income/new',data=dict(date='2026-09-02',asset_id=self.asset_id,
|
|
category='dividend',amount='0,04',received='1'),follow_redirects=False)
|
|
self.assertEqual(response.status_code,303)
|
|
entry = self.request('GET','/income').json()[0]
|
|
self.request('PATCH',f"/income/{entry['id']}",json={'amount':'12.34'})
|
|
self.assertIn('12,34 €',self.client.get('/income').text)
|
|
self.client.post(f"/income/{entry['id']}/delete")
|
|
self.assertEqual(self.request('GET','/income').json(),[])
|
|
|
|
def test_openapi_meta_errors_and_no_cors(self):
|
|
schema = self.client.get('/openapi.json').json()
|
|
for path, methods in schema['paths'].items():
|
|
if path.startswith('/api/v1/'):
|
|
for operation in methods.values():
|
|
self.assertEqual(operation['security'],[{'FinanceAPIToken':[]}])
|
|
self.assertEqual(self.request('GET','/meta').json(),{'name':'Finance Dashboard','api_version':'v1','database':'ok'})
|
|
response = self.client.get('/api/v1/assets',headers={**self.headers,'Origin':'https://example.com'})
|
|
self.assertNotIn('access-control-allow-origin',response.headers)
|
|
with patch('api.routes.asset_service.list_assets',side_effect=RuntimeError('private-internal-information')):
|
|
response = self.request('GET','/assets')
|
|
self.assertEqual(response.status_code,500)
|
|
self.assertNotIn('private-internal-information',response.text)
|
|
with patch('api.routes.asset_service.list_assets',side_effect=sqlite3.OperationalError('secret-path')):
|
|
response = self.request('GET','/assets')
|
|
self.assertEqual(response.status_code,503)
|
|
self.assertNotIn('secret-path',response.text)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|