[tests] Added tests and updated storage.ledgercli
- [ledgercli] Log info messages for operations such as add, delete, update transactions
This commit is contained in:
Joar Wandborg
parent
0108548b4b
commit
3620d97f9d
3 changed files with 284 additions and 0 deletions
|
|
@ -211,6 +211,8 @@ class Ledger(Storage):
|
|||
with open(self.ledger_file, 'ab') as f:
|
||||
f.write(output)
|
||||
|
||||
_log.info('Added transaction %s', transaction.id)
|
||||
|
||||
_log.debug('written to file: %s', output)
|
||||
|
||||
return transaction.id
|
||||
|
|
@ -419,6 +421,11 @@ class Ledger(Storage):
|
|||
# Delete the preceding line to make the file
|
||||
del_start -= 1
|
||||
|
||||
_log.info('Removing transaction with ID: %s (lines %d-%d)',
|
||||
transaction_id,
|
||||
del_start,
|
||||
semantic_lines['next_transaction_or_eof'])
|
||||
|
||||
del lines[del_start:semantic_lines['next_transaction_or_eof']]
|
||||
|
||||
with open(self.ledger_file, 'w') as f:
|
||||
|
|
@ -444,6 +451,7 @@ class Ledger(Storage):
|
|||
|
||||
self.add_transaction(transaction)
|
||||
|
||||
_log.info('Updated transaction %s', transaction.id)
|
||||
_log.debug('Updated transaction from: %s to: %s', old_transaction,
|
||||
transaction)
|
||||
|
||||
|
|
|
|||
0
accounting/tests/__init__.py
Normal file
0
accounting/tests/__init__.py
Normal file
276
accounting/tests/test_transactions.py
Normal file
276
accounting/tests/test_transactions.py
Normal file
|
|
@ -0,0 +1,276 @@
|
|||
'''
|
||||
Tests for accounting-api
|
||||
'''
|
||||
import os
|
||||
import unittest
|
||||
import tempfile
|
||||
import logging
|
||||
import copy
|
||||
import uuid
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from flask import json
|
||||
|
||||
from accounting.web import app, init_ledger
|
||||
|
||||
from accounting.transport import AccountingEncoder, AccountingDecoder
|
||||
from accounting.models import Transaction, Posting, Amount
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
|
||||
class TransactionTestCase(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.app = app.test_client()
|
||||
self.fd, app.config['LEDGER_FILE'] = tempfile.mkstemp()
|
||||
init_ledger()
|
||||
self.simple_transaction = Transaction(
|
||||
date=datetime.today(),
|
||||
payee='Joar',
|
||||
postings=[
|
||||
Posting('Assets:Checking', Amount('-133.7', 'USD')),
|
||||
Posting('Expenses:Foo', Amount('133.7', 'USD'))
|
||||
]
|
||||
)
|
||||
|
||||
def tearDown(self):
|
||||
os.close(self.fd)
|
||||
os.unlink(app.config['LEDGER_FILE'])
|
||||
|
||||
def test_get_transactions(self):
|
||||
open(app.config['LEDGER_FILE'], 'w').write(
|
||||
'1400-12-21 Old stuff\n'
|
||||
' ;Id: foo\n'
|
||||
' Assets:Checking -100 USD\n'
|
||||
' Expenses:Tax 100 USD\n')
|
||||
rv = self.app.get('/transaction')
|
||||
|
||||
json_transaction = (
|
||||
b'{\n'
|
||||
b' "transactions": [\n'
|
||||
b' {\n'
|
||||
b' "__type__": "Transaction", \n'
|
||||
b' "date": "1400-12-21", \n'
|
||||
b' "id": "foo", \n'
|
||||
b' "metadata": {}, \n'
|
||||
b' "payee": "Old stuff", \n'
|
||||
b' "postings": [\n'
|
||||
b' {\n'
|
||||
b' "__type__": "Posting", \n'
|
||||
b' "account": "Assets:Checking", \n'
|
||||
b' "amount": {\n'
|
||||
b' "__type__": "Amount", \n'
|
||||
b' "amount": "-100", \n'
|
||||
b' "symbol": "USD"\n'
|
||||
b' }, \n'
|
||||
b' "metadata": {}\n'
|
||||
b' }, \n'
|
||||
b' {\n'
|
||||
b' "__type__": "Posting", \n'
|
||||
b' "account": "Expenses:Tax", \n'
|
||||
b' "amount": {\n'
|
||||
b' "__type__": "Amount", \n'
|
||||
b' "amount": "100", \n'
|
||||
b' "symbol": "USD"\n'
|
||||
b' }, \n'
|
||||
b' "metadata": {}\n'
|
||||
b' }\n'
|
||||
b' ]\n'
|
||||
b' }\n'
|
||||
b' ]\n'
|
||||
b'}')
|
||||
|
||||
self.assertEqual(rv.get_data(), json_transaction)
|
||||
|
||||
def _post_json(self, path, data, expect=200, **kw):
|
||||
response = self.app.post(
|
||||
path,
|
||||
content_type='application/json',
|
||||
data=json.dumps(data, cls=AccountingEncoder),
|
||||
**kw
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, expect)
|
||||
|
||||
return self._decode_response(response)
|
||||
|
||||
def _decode_response(self, response):
|
||||
return json.loads(response.data, cls=AccountingDecoder)
|
||||
|
||||
def _get_json(self, path, expect=200, **kw):
|
||||
response = self.app.get(path, **kw)
|
||||
|
||||
self.assertEqual(response.status_code, expect)
|
||||
|
||||
return self._decode_response(response)
|
||||
|
||||
def _open_json(self, method, path, expect=200, **kw):
|
||||
response = self.app.open(
|
||||
path,
|
||||
method=method.upper(),
|
||||
**kw
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, expect)
|
||||
|
||||
return self._decode_response(response)
|
||||
|
||||
def _add_simple_transaction(self, transaction_id=None):
|
||||
if transaction_id is None:
|
||||
transaction_id = str(uuid.uuid4())
|
||||
|
||||
transaction = copy.deepcopy(self.simple_transaction)
|
||||
transaction.id = transaction_id
|
||||
|
||||
response = self._post_json('/transaction', transaction)
|
||||
|
||||
self.assertEqual(len(response['transaction_ids']), 1)
|
||||
self.assertEqual(response['status'], 'OK')
|
||||
|
||||
response = self._get_json('/transaction/' + transaction.id)
|
||||
|
||||
self.assertEqual(transaction_id, response['transaction'].id)
|
||||
|
||||
self.assertEqual(response['transaction'], transaction)
|
||||
|
||||
return transaction
|
||||
|
||||
def test_post_transaction_without_id(self):
|
||||
transaction = copy.deepcopy(self.simple_transaction)
|
||||
|
||||
response = self._post_json('/transaction', transaction)
|
||||
|
||||
self.assertEqual(len(response['transaction_ids']), 1)
|
||||
self.assertEqual(response['status'], 'OK')
|
||||
|
||||
transaction.id = response['transaction_ids'][0]
|
||||
|
||||
response = self._get_json('/transaction/' + transaction.id)
|
||||
|
||||
self.assertEqual(response['transaction'], transaction)
|
||||
|
||||
def test_delete_transaction(self):
|
||||
transaction = copy.deepcopy(self.simple_transaction)
|
||||
|
||||
response = self._post_json('/transaction', transaction)
|
||||
|
||||
transaction_id = response['transaction_ids'][0]
|
||||
|
||||
self.assertIsNotNone(transaction_id)
|
||||
|
||||
response = self._open_json('DELETE',
|
||||
'/transaction/' + transaction_id)
|
||||
|
||||
self.assertEqual(response['status'], 'OK')
|
||||
|
||||
with self.assertRaises(ValueError):
|
||||
# ValueError thrown because the response does not contain any JSON
|
||||
response = self._get_json('/transaction/' + transaction_id, 404)
|
||||
|
||||
def test_post_multiple_transactions(self):
|
||||
transactions = [
|
||||
Transaction(
|
||||
date=datetime.today(),
|
||||
payee='Rent',
|
||||
postings=[
|
||||
Posting(
|
||||
account='Assets:Checking',
|
||||
amount=Amount(amount='-4600.00', symbol='SEK')
|
||||
),
|
||||
Posting(
|
||||
account='Expenses:Rent',
|
||||
amount=Amount(amount='4600.00', symbol='SEK')
|
||||
)
|
||||
]
|
||||
),
|
||||
Transaction(
|
||||
date=datetime.today(),
|
||||
payee='Hosting',
|
||||
postings=[
|
||||
Posting(
|
||||
account='Assets:Checking',
|
||||
amount=Amount(amount='-700.00', symbol='SEK')
|
||||
),
|
||||
Posting(
|
||||
account='Expenses:Hosting',
|
||||
amount=Amount(amount='700.00', symbol='SEK')
|
||||
)
|
||||
]
|
||||
)
|
||||
]
|
||||
|
||||
response = self._post_json('/transaction',
|
||||
{'transactions': transactions})
|
||||
|
||||
self.assertEqual(len(response['transaction_ids']), 2)
|
||||
|
||||
transactions[0].id = response['transaction_ids'][0]
|
||||
transactions[1].id = response['transaction_ids'][1]
|
||||
|
||||
response = self._get_json('/transaction/' + transactions[0].id)
|
||||
|
||||
self.assertEqual(transactions[0], response['transaction'])
|
||||
|
||||
response = self._get_json('/transaction/' + transactions[1].id)
|
||||
|
||||
self.assertEqual(transactions[1], response['transaction'])
|
||||
|
||||
def test_update_transaction_payee(self):
|
||||
transaction = self._add_simple_transaction()
|
||||
|
||||
transaction.payee = 'not Joar'
|
||||
|
||||
response = self._post_json('/transaction/' + transaction.id,
|
||||
{'transaction': transaction})
|
||||
|
||||
self.assertEqual(response['status'], 'OK')
|
||||
|
||||
response = self._get_json('/transaction/'+ transaction.id)
|
||||
|
||||
self.assertEqual(response['transaction'], transaction)
|
||||
|
||||
def test_update_transaction_postings(self):
|
||||
transaction = self._add_simple_transaction()
|
||||
|
||||
postings = [
|
||||
Posting(account='Assets:Checking',
|
||||
amount=Amount(amount='-733.10', symbol='SEK')),
|
||||
Posting(account='Expenses:Bar',
|
||||
amount=Amount(amount='733.10', symbol='SEK'))
|
||||
]
|
||||
|
||||
transaction.postings = postings
|
||||
|
||||
response = self._post_json('/transaction/' + transaction.id,
|
||||
{'transaction': transaction})
|
||||
|
||||
self.assertEqual(response['status'], 'OK')
|
||||
|
||||
response = self._get_json('/transaction/' + transaction.id)
|
||||
|
||||
self.assertEqual(response['transaction'], transaction)
|
||||
|
||||
def test_post_unbalanced_transaction(self):
|
||||
transaction = Transaction(
|
||||
date=datetime.today(),
|
||||
payee='Unbalanced Transaction',
|
||||
postings=[
|
||||
Posting(account='Assets:Checking',
|
||||
amount=Amount(amount='100.00', symbol='USD')),
|
||||
Posting(account='Income:Foo',
|
||||
amount=Amount(amount='-100.01', symbol='USD'))
|
||||
]
|
||||
)
|
||||
|
||||
self._post_json('/transaction', transaction)
|
||||
|
||||
response = self._get_json('/transaction')
|
||||
|
||||
import pdb; pdb.set_trace()
|
||||
|
||||
def test_update_transaction_amounts(self): pass
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue