Changeset - 3620d97f9d16
[Not reviewed]
0 1 2
Joar Wandborg - 10 years ago 2013-12-26 12:22:27
joar@wandborg.se
[tests] Added tests and updated storage.ledgercli

- [ledgercli] Log info messages for operations such as add, delete, update
transactions
3 files changed with 284 insertions and 0 deletions:
0 comments (0 inline, 0 general)
accounting/storage/ledgercli.py
Show inline comments
...
 
@@ -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)
 

	
accounting/tests/__init__.py
Show inline comments
 
new file 100644
accounting/tests/test_transactions.py
Show inline comments
 
new file 100644
 
'''
 
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()
0 comments (0 inline, 0 general)