Files @ ef12c232ada2
Branch filter:

Location: NPO-Accounting/experimental-accounting-api/accounting/tests/test_transactions.py

Joar Wandborg
[doc] Updated restapi documentation
'''
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()