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()