192 lines
9.6 KiB
Python
192 lines
9.6 KiB
Python
import csv
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
import io
|
|
import os
|
|
from pathlib import Path
|
|
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 openpyxl import Workbook
|
|
from database import connect
|
|
from main import app
|
|
from models import cents, percent
|
|
from services.excel_import import import_excel
|
|
from services.income_service import dashboard
|
|
|
|
|
|
class FinanceTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.path = Path(self.temp.name) / 'finance.db'
|
|
self.env = patch.dict(os.environ, {'FINANCE_DB_PATH': str(self.path)})
|
|
self.env.start()
|
|
self.client = TestClient(app)
|
|
self.client.__enter__()
|
|
with connect() as db:
|
|
self.asset = db.execute("SELECT id FROM assets WHERE name='Main Street Capital'").fetchone()[0]
|
|
|
|
def tearDown(self):
|
|
self.client.__exit__(None, None, None)
|
|
self.env.stop()
|
|
self.temp.cleanup()
|
|
|
|
def payment(self, **changes):
|
|
data = dict(date='2026-09-02', asset_id=str(self.asset), category='dividend', amount='0,04', note='', received='1')
|
|
data.update(changes)
|
|
response = self.client.post('/income/new', data=data, follow_redirects=False)
|
|
self.assertEqual(response.status_code, 303, response.text)
|
|
return response
|
|
|
|
def test_health_and_empty_dashboard(self):
|
|
self.assertEqual(self.client.get('/health').json(), {'status': 'ok'})
|
|
response = self.client.get('/')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('Noch keine Zahlungen', response.text)
|
|
self.assertIn('0,00 €', response.text)
|
|
|
|
def test_create_edit_delete_and_validation(self):
|
|
self.payment(note='<script>alert(1)</script>')
|
|
self.assertIn('<script>', self.client.get('/income').text)
|
|
with connect() as db:
|
|
row = db.execute('SELECT * FROM income_entries').fetchone()
|
|
self.assertEqual(row['amount'], 4)
|
|
entry_id = row['id']
|
|
self.assertEqual(self.client.get(f'/income/{entry_id}/edit').status_code, 200)
|
|
response = self.client.post(f'/income/{entry_id}/edit', data=dict(date='2026-09-03', asset_id=self.asset, category='interest', amount='28.00', received='1'))
|
|
self.assertEqual(response.status_code, 200)
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT amount FROM income_entries').fetchone()[0], 2800)
|
|
with self.assertRaises(sqlite3.IntegrityError):
|
|
db.execute('DELETE FROM assets WHERE id=?', (self.asset,))
|
|
for changes in [dict(amount='NaN'), dict(amount='1.234'), dict(date='2026-02-30'), dict(asset_id='99999'), dict(asset_id='9'*30), dict(category='invalid'), dict(amount='999999999999999')]:
|
|
data = dict(date='2026-09-02', asset_id=self.asset, category='dividend', amount='0,04', received='1')
|
|
data.update(changes)
|
|
self.assertEqual(self.client.post('/income/new', data=data).status_code, 422)
|
|
self.assertEqual(self.client.post(f'/income/{entry_id}/delete', follow_redirects=False).status_code, 303)
|
|
self.assertEqual(self.client.get(f'/income/{entry_id}/edit').status_code, 404)
|
|
|
|
def test_month_year_and_comparison(self):
|
|
self.payment(date='2025-09-02', amount='10')
|
|
self.payment(date='2025-01-02', amount='5')
|
|
self.payment(amount='12')
|
|
self.payment(amount='3')
|
|
self.payment(date='2026-01-02', amount='15')
|
|
self.payment(amount='28', expected='1', received='')
|
|
stats = dashboard(date(2026,9,9))
|
|
self.assertEqual(stats['month'], 1500)
|
|
self.assertEqual(stats['prior_month'], 1000)
|
|
self.assertEqual(stats['year'], 3000)
|
|
self.assertEqual(stats['prior_year'], 1500)
|
|
self.assertEqual(stats['month_change'], Decimal('50.00'))
|
|
self.assertEqual(stats['year_change'], Decimal('100.00'))
|
|
self.assertEqual(stats['all_time'], 4500)
|
|
self.assertEqual(stats['count'], 3)
|
|
self.assertEqual(stats['pending']['amount'], 2800)
|
|
self.assertEqual(self.client.get('/').status_code, 200)
|
|
|
|
def test_zero_and_decimal_and_dynamic_years(self):
|
|
self.assertIsNone(percent(100, 0))
|
|
self.assertEqual(percent(50,100), Decimal('-50.00'))
|
|
for value in ['0.04', '0,04']:
|
|
self.assertEqual(cents(value),4)
|
|
self.payment(date='2028-01-01')
|
|
stats = dashboard(date(2026,9,9))
|
|
self.assertIn(2028, stats['years'])
|
|
self.assertIn('2028 vs. 2027', self.client.get('/').text)
|
|
self.assertIsNone(stats['year_change'])
|
|
|
|
def test_csv_filters_and_security(self):
|
|
self.payment(note='=HYPERLINK("bad")')
|
|
self.payment(date='2025-03-01', category='interest', received='', expected='1')
|
|
response = self.client.get('/export/income.csv')
|
|
self.assertEqual(response.status_code,200)
|
|
self.assertTrue(response.content.startswith(b'\xef\xbb\xbf'))
|
|
rows = list(csv.reader(io.StringIO(response.content.decode('utf-8-sig')), delimiter=';'))
|
|
self.assertEqual(rows[0], ['Datum','Position','Kategorie','Betrag','Notiz','Erwartet','Erhalten'])
|
|
self.assertEqual(rows[1][3], '0,04')
|
|
self.assertTrue(rows[1][4].startswith("'="))
|
|
self.assertEqual(rows[2][-2:], ['Ja','Nein'])
|
|
self.assertEqual(self.client.get('/income?year=&month=&asset_id=&category=').status_code,200)
|
|
self.assertEqual(self.client.get('/income?month=13').status_code,422)
|
|
response = self.client.get('/income?year=2026&month=9&category=dividend')
|
|
self.assertIn('02.09.2026',response.text)
|
|
self.assertNotIn('01.03.2025',response.text)
|
|
self.assertEqual(self.client.post('/income/1/delete', headers={'Origin':'https://evil.example'}).status_code,403)
|
|
|
|
def workbook(self, invalid=False):
|
|
book = Workbook()
|
|
sheet = book.active
|
|
sheet.append(['Dashboard', 'Gesamt', '=SUM(C4:C9)'])
|
|
sheet.append(['Datum','Art des Ertrags','Betrag (€)','Kategorie','Erwartet','Erhalten'])
|
|
sheet.append([date(2026,9,2),'MSC',0.04,'Dividenden / Ausschüttungen',None,None])
|
|
sheet.append([date(2026,9,2),'Main Street Capital',0.04,'dividend',None,None])
|
|
sheet.append([date(2026,8,1),'Air Baltic',28,'Zinsen',True,False])
|
|
sheet.append([date(2026,9,1),'Neue Position',1.25,'Sonstiges',None,None])
|
|
if invalid:
|
|
sheet.append(['unbekannt','MSC',4,'dividend'])
|
|
book.create_sheet('Ignorieren').append([date(2026,1,1),'MSC',999,'dividend'])
|
|
filename = Path(self.temp.name) / 'history.xlsx'
|
|
book.save(filename)
|
|
book.close()
|
|
return filename
|
|
|
|
def test_import_duplicate_occurrences_aliases_and_tombstones(self):
|
|
filename = self.workbook()
|
|
self.payment() # manual match must be reused
|
|
result = import_excel(filename)
|
|
self.assertEqual(result['rows'],4)
|
|
self.assertEqual(result['added'],3)
|
|
self.assertEqual(import_excel(filename)['added'],0)
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM income_entries').fetchone()[0],4)
|
|
self.assertEqual(db.execute("SELECT COUNT(*) FROM assets WHERE normalized_name='mainstreetcapital'").fetchone()[0],1)
|
|
self.assertEqual(db.execute('SELECT SUM(amount) FROM income_entries WHERE received=1').fetchone()[0],133)
|
|
entry_id = db.execute('SELECT id FROM income_entries ORDER BY id LIMIT 1').fetchone()[0]
|
|
self.client.post(f'/income/{entry_id}/delete')
|
|
self.assertEqual(import_excel(filename)['added'],0)
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM income_entries').fetchone()[0],3)
|
|
|
|
def test_import_atomic_and_dry_run(self):
|
|
filename = self.workbook()
|
|
self.assertEqual(import_excel(filename, dry_run=True)['added'],4)
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM income_entries').fetchone()[0],0)
|
|
filename = self.workbook(invalid=True)
|
|
with self.assertRaises(ValueError):
|
|
import_excel(filename)
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM income_entries').fetchone()[0],0)
|
|
|
|
def test_sqlite_lock_returns_retryable_error(self):
|
|
locker = sqlite3.connect(self.path)
|
|
try:
|
|
locker.execute('BEGIN IMMEDIATE')
|
|
response = self.client.post('/income/new', data=dict(date='2026-09-01', asset_id=self.asset,
|
|
category='dividend', amount='1', received='1'))
|
|
self.assertEqual(response.status_code, 503)
|
|
self.assertEqual(response.headers['retry-after'], '5')
|
|
finally:
|
|
locker.rollback()
|
|
locker.close()
|
|
with connect() as db:
|
|
self.assertEqual(db.execute('SELECT COUNT(*) FROM income_entries').fetchone()[0], 0)
|
|
|
|
def test_asset_normalization_and_inactive_asset(self):
|
|
response = self.client.post('/assets/new', data={'name':'MSC','asset_type':'stock'}, follow_redirects=False)
|
|
self.assertEqual(response.status_code,303)
|
|
self.assertEqual(response.headers['location'],f'/income/new?asset_id={self.asset}')
|
|
with connect() as db:
|
|
db.execute('UPDATE assets SET active=0 WHERE id=?',(self.asset,))
|
|
self.assertEqual(self.client.post('/income/new',data=dict(date='2026-09-01',asset_id=self.asset,category='dividend',amount='1')).status_code,422)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|