diff --git a/accounting/tests/test_transactions.py b/accounting/tests/test_transactions.py new file mode 100644 index 0000000000000000000000000000000000000000..03ea2fcbf19c5d0883f63ea7e814cb96e965054c --- /dev/null +++ b/accounting/tests/test_transactions.py @@ -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()