path: root/talerbank/app/tests.py

#  This file is part of TALER
#  (C) 2014, 2015, 2016 Taler Systems SA
#
#  TALER is free software; you can redistribute it and/or modify
#  it under the terms of the GNU Affero General Public License as
#  published by the Free Software Foundation; either version 3,
#  or (at your option) any later version.
#
#  TALER is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty
#  of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#  See the GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public
#  License along with TALER; see the file COPYING.  If not,
#  see <http://www.gnu.org/licenses/>.
#
#  @author Marcello Stanisci

import json
import time
import zlib
import timeit
import logging
import unittest
import base64
from urllib.parse import unquote
from django.db import connection
from django.test import TestCase, Client
from django.urls import reverse
from django.conf import settings
from django.contrib.auth.models import User
from mock import patch, MagicMock
from .models import BankAccount, BankTransaction
from . import urls
from .views import wire_transfer
from taler.util.amount import (
    Amount,
    SignedAmount,
    CurrencyMismatchError,
    AmountFormatError,
)

LOGGER = logging.getLogger()
LOGGER.setLevel(logging.DEBUG)

# logging.disable(logging.CRITICAL)
# reenable: logging.disable(logging.NOTSET)
logging.disable(logging.NOTSET)


def make_auth_line(username, password):
    credentials = "%s:%s" % (username, password)
    b64enc = base64.b64encode(bytes(credentials, "utf-8"))
    header_line = "Basic %s" % b64enc.decode()
    return header_line


def clear_db():
    User.objects.all().delete()
    BankAccount.objects.all().delete()
    BankTransaction.objects.all().delete()
    with connection.cursor() as cursor:
        cursor.execute("ALTER SEQUENCE app_bankaccount_account_no_seq" " RESTART")
        cursor.execute("ALTER SEQUENCE app_banktransaction_id_seq RESTART")


# Bank used to crash when malformed JSON was sent
# to some endpoint that needs to authenticate the
# user.
class MalformedLoginTestCase(TestCase):
    def test_malformed_login(self):
        self.client.generic("POST", reverse("add-incoming", urlconf=urls), "malformed")


class PublicAccountsTestCase(TestCase):
    def setUp(self):
        clear_db()
        self.user = User.objects.create_user(username="Bank", password="Bank")
        self.user.save()

        self.user_bank_account = BankAccount(
            account_no=100, is_public=True, user=self.user
        )

        self.user_bank_account.save()

    def test_public_accounts(self):
        self.assertTrue(User.objects.get(username="Bank"))

        response = self.client.get(reverse("public-accounts", urlconf=urls))


class WithdrawTestCase(TestCase):
    def setUp(self):
        self.user_bank_account = BankAccount(
            user=User.objects.create_user(
                username="test_user", password="test_password"
            ),
            account_no=100,
        )
        self.user_bank_account.save()

        self.exchange_bank_account = BankAccount(
            user=User.objects.create_user(username="test_exchange", password=""),
            account_no=99,
        )
        self.exchange_bank_account.save()
        self.client = Client()

    @patch("talerbank.app.views.wire_transfer")
    @patch("hashlib.new")
    @patch("time.time")
    @unittest.skip("skip outdated test case")
    def test_withdraw(self, mocked_time, mocked_hashlib, mocked_wire_transfer):
        amount = Amount(settings.TALER_CURRENCY, 0, 1)
        params = {
            "amount_value": str(amount.value),
            "amount_fraction": str(amount.fraction),
            "amount_currency": amount.currency,
            "reserve_pub": "UVZ789",
            "exchange": "https://exchange.example.com/",
            "exchange_wire_details": "payto://x-taler-bank/bank.example/99",
        }
        self.client.login(username="test_user", password="test_password")

        response = self.client.get(reverse("pin-question", urlconf=urls), params)
        self.assertEqual(response.status_code, 200)
        # We mock hashlib in order to fake the CAPTCHA.
        hasher = MagicMock()
        hasher.hexdigest = MagicMock()
        hasher.hexdigest.return_value = "0"
        mocked_hashlib.return_value = hasher
        mocked_time.return_value = 0

        response = self.client.post(
            reverse("withdraw-confirm", urlconf=urls), {"pin_1": "0"}
        )

        args, kwargs = mocked_wire_transfer.call_args
        del kwargs
        self.assertTrue(
            args[0].dump() == amount.dump()
            and self.user_bank_account in args
            and "UVZ789" in args
            and self.exchange_bank_account in args
        )

    def tearDown(self):
        clear_db()


class InternalWireTransferTestCase(TestCase):
    def setUp(self):
        BankAccount(
            user=User.objects.create_user(username="give_money", password="gm")
        ).save()
        self.take_money = BankAccount(
            user=User.objects.create_user(username="take_money"), account_no=4
        )
        self.take_money.save()

    def tearDown(self):
        clear_db()

    def test_internal_wire_transfer(self):
        client = Client()
        client.login(username="give_money", password="gm")
        response = client.post(
            reverse("profile", urlconf=urls),
            {
                "amount": 3.0,
                "receiver": self.take_money.account_no,
                "subject": "charity",
            },
        )
        take_money = BankAccount.objects.get(account_no=4)
        r = SignedAmount.parse(f"{settings.TALER_CURRENCY}:3.0")
        self.assertEqual(take_money.balance, r)
        self.assertEqual(302, response.status_code)


class RegisterTestCase(TestCase):
    """User registration"""

    def setUp(self):
        clear_db()
        BankAccount(user=User.objects.create_user(username="Bank")).save()

    def tearDown(self):
        clear_db()

    def test_register(self):
        client = Client()
        response = client.post(
            reverse("register", urlconf=urls),
            {"username": "test_register", "password": "test_register"},
            follow=True,
        )
        self.assertIn(("/profile", 302), response.redirect_chain)
        # this assertion tests "/profile""s view
        self.assertEqual(200, response.status_code)

    def test_register_headless(self):
        client = Client()

        # Normal case.
        response = client.post(
            reverse("register-headless", urlconf=urls),
            {"username": "test_register_headless", "password": "password*+#@"},
        )
        self.assertEqual(200, response.status_code)

        # Double-check account creation.
        self.assertTrue(
            self.client.login(
                username="test_register_headless", password="password*+#@"
            )
        )

        # Try registering unavailable username.
        response = client.post(
            reverse("register-headless", urlconf=urls),
            {"username": "test_register_headless", "password": "password"},
        )
        self.assertEqual(409, response.status_code)

        # NOTE: Django 2.2.2 allows ANY character!  Is this normal?
        response = client.post(
            reverse("register-headless", urlconf=urls),
            {"username": "'''+++;;;'''", "password": "password2"},
        )
        self.assertEqual(200, response.status_code)


class LoginTestCase(TestCase):
    """User login"""

    def setUp(self):
        BankAccount(
            user=User.objects.create_user(
                username="test_user", password="test_password"
            )
        ).save()
        self.client = Client()

    def tearDown(self):
        clear_db()

    def test_login(self):
        self.assertTrue(
            self.client.login(username="test_user", password="test_password")
        )
        self.assertFalse(
            self.client.login(username="test_user", password="test_passwordii")
        )

    def test_failing_login(self):
        response = self.client.get(
            reverse("history", urlconf=urls),
            {"auth": "basic"},
            HTTP_AUTHORIZATION=make_auth_line("Wrong", "Credentials"),
        )
        data = response.content.decode("utf-8")
        self.assertEqual(401, response.status_code)


class WithdrawHeadlessTestCase(TestCase):
    def setUp(self):
        BankAccount(
            user=User.objects.create_user(
                username="headless_wallet", password="headless_password"
            ),
            balance=SignedAmount(True, Amount(settings.TALER_CURRENCY, 10, 0)),
        ).save()
        # Gets account #2, in line with config.
        BankAccount(
            user=User.objects.create_user(
                username="normal_exchange", password="normal_password"
            ),
            account_no=2,
        ).save()

    def test_withdraw_headless(self):
        client = Client()

        # Use default exchange.
        data = (
            '{"auth": {"type": "basic"}, \
                 "reserve_pub": "RESERVEPUB", \
                 "amount": "%s:10"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("withdraw-headless", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("headless_wallet", "headless_password"),
        )
        self.assertEqual(200, response.status_code)

        # Try withdrawing more than owning.
        data = (
            '{"auth": {"type": "basic"}, \
                 "reserve_pub": "RESERVEPUB", \
                 "amount": "%s:100"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("withdraw-headless", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("headless_wallet", "headless_password"),
        )
        self.assertEqual(406, response.status_code)

        # Try withdrawing giving exchange field.
        data = (
            '{"auth": {"type": "basic"}, \
                 "exchange_wire_details": "payto://x-taler-bank/bank.example.com/2", \
                 "reserve_pub": "RESERVEPUB", \
                 "amount": "%s:0.4"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("withdraw-headless", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("headless_wallet", "headless_password"),
        )
        self.assertEqual(200, response.status_code)

        # Try withdrawing giving non-existent recipient.
        data = (
            '{"auth": {"type": "basic"}, \
                 "exchange_wire_details": "payto://x-taler-bank/bank.example.com/2222", \
                 "reserve_pub": "RESERVEPUB", \
                 "amount": "%s:0.4"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("withdraw-headless", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("headless_wallet", "headless_password"),
        )
        self.assertEqual(404, response.status_code)

        # Try withdrawing giving invalid JSON.
        data = (
            '{"auth": {"type": "basic"}, \
                 "XXX": "YYY", \
                 "amount": "%s:0.4"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("withdraw-headless", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("headless_wallet", "headless_password"),
        )
        self.assertEqual(400, response.status_code)

    def tearDown(self):
        clear_db()


class AddIncomingTestCase(TestCase):
    """Test money transfer's API"""

    def setUp(self):
        BankAccount(
            user=User.objects.create_user(
                username="bank_user", password="bank_password"
            )
        ).save()
        BankAccount(
            user=User.objects.create_user(
                username="user_user", password="user_password"
            )
        ).save()

    def tearDown(self):
        clear_db()

    def test_add_incoming(self):
        client = Client()
        data = (
            '{"auth": {"type": "basic"}, \
                 "credit_account": 1, \
                 "subject": "TESTWTID", \
                 "exchange_url": "https://exchange.test", \
                 "amount": "%s:1.0"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("add-incoming", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("user_user", "user_password"),
        )
        self.assertEqual(200, response.status_code)

        # Trying the same request, but compressed.
        zdata = zlib.compress(bytes(data, "utf-8"))
        response = client.post(
            reverse("add-incoming", urlconf=urls),
            data=zdata,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("user_user", "user_password"),
            HTTP_CONTENT_ENCODING="deflate",
        )
        self.assertEqual(200, response.status_code)

        data = '{"auth": {"type": "basic"}, \
                 "credit_account": 1, \
                 "subject": "TESTWTID", \
                 "exchange_url": "https://exchange.test", \
                 "amount": "WRONGCURRENCY:1.0"}'

        response = client.post(
            reverse("add-incoming", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("user_user", "user_password"),
        )
        # note: a bad currency request gets 400.
        self.assertEqual(400, response.status_code)
        LOGGER.info(response.content.decode("utf-8"))

        # Try to go debit
        data = (
            '{"auth": {"type": "basic"}, \
                 "credit_account": 1, \
                 "subject": "TESTWTID", \
                 "exchange_url": "https://exchange.test", \
                 "amount": "%s:50.1"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("add-incoming", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("user_user", "user_password"),
        )
        self.assertEqual(406, response.status_code)
        # Try use a non-existent recipient.
        data = (
            '{"auth": {"type": "basic"}, \
                 "credit_account": 1987, \
                 "subject": "TESTWTID", \
                 "exchange_url": "https://exchange.test", \
                 "amount": "%s:1"}'
            % settings.TALER_CURRENCY
        )
        response = client.post(
            reverse("add-incoming", urlconf=urls),
            data=data,
            content_type="application/json",
            follow=True,
            HTTP_AUTHORIZATION=make_auth_line("user_user", "user_password"),
        )
        self.assertEqual(404, response.status_code)


class CustomDoesNotExistTestCase(TestCase):
    def test_bankaccount_doesnotexist(self):
        with self.assertRaises(BankAccount.DoesNotExist):
            BankAccount.objects.get(account_no=1000)
        with self.assertRaises(BankTransaction.DoesNotExist):
            BankTransaction.objects.get(subject="1000")


class HistoryTestCase(TestCase):
    def setUp(self):
        clear_db()
        debit_account = BankAccount(
            user=User.objects.create_user(username="User", password="Password"),
            balance=SignedAmount(True, Amount(settings.TALER_CURRENCY, 100, 0)),
        )
        debit_account.save()
        credit_account = BankAccount(
            user=User.objects.create_user(username="User0", password="Password0")
        )
        credit_account.save()
        for subject in ("a", "b", "c", "d", "e", "f", "g", "h", "i"):
            wire_transfer(
                Amount(settings.TALER_CURRENCY, 1, 0),
                debit_account,
                credit_account,
                subject,
            )

    def tearDown(self):
        clear_db()

    def test_history(self):

        def histquery(**urlargs):
            response = self.client.get(
                reverse("history", urlconf=urls),
                urlargs,
                HTTP_AUTHORIZATION=make_auth_line("User", "Password"),
            )
            return response

        # test query #1
        r = histquery(delta="-4", direction="both")
        rd = json.loads(r.content)
        self.assertEqual(r.status_code, 200)

        # test query #2
        r = histquery(delta="+1", start="5", direction="both")
        self.assertEqual(r.status_code, 200)
        rd = json.loads(r.content)
        self.assertEqual(r.status_code, 200)
        self.assertEqual(rd["data"][0]["row_id"], 6)

        # test query #3
        r = histquery(delta="+1", start="2", direction="both")
        self.assertEqual(r.status_code, 200)
        rd = json.loads(r.content)
        self.assertEqual(rd["data"][0]["wt_subject"], "c")

        # test query #4
        r = histquery(delta="-1", start="2", direction="both")
        self.assertEqual(r.status_code, 200)
        rd = json.loads(r.content)
        self.assertEqual(rd["data"][0]["wt_subject"], "a")

        # test query #5
        r = histquery(delta="1", start="11", direction="both")
        self.assertEqual(r.status_code, 200)
        rd = json.loads(r.content)
        self.assertEqual(len(rd["data"]), 0)


class DBCustomColumnTestCase(TestCase):
    def setUp(self):
        BankAccount(user=User.objects.create_user(username="U")).save()

    def tearDown(self):
        clear_db()

    def test_exists(self):
        user_bankaccount = BankAccount.objects.get(user=User.objects.get(username="U"))
        self.assertTrue(isinstance(user_bankaccount.balance, SignedAmount))


## This tests whether a bank account goes debit and then goes >=0
## again
class DebitTestCase(TestCase):
    def setUp(self):
        BankAccount(user=User.objects.create_user(username="U")).save()
        BankAccount(user=User.objects.create_user(username="U0")).save()

    def tearDown(self):
        clear_db()

    def test_green(self):
        user_bankaccount = BankAccount.objects.get(user=User.objects.get(username="U"))
        self.assertTrue(user_bankaccount.balance.is_zero())

    def test_red(self):
        user_bankaccount = BankAccount.objects.get(user=User.objects.get(username="U"))
        user_bankaccount0 = BankAccount.objects.get(
            user=User.objects.get(username="U0")
        )

        wire_transfer(
            Amount(settings.TALER_CURRENCY, 10, 0),
            user_bankaccount0,
            user_bankaccount,
            "Go green",
        )

        wire_transfer(
            Amount(settings.TALER_CURRENCY, 11, 0),
            user_bankaccount,
            user_bankaccount0,
            "Go red",
        )

        amt_one = SignedAmount.parse(f"{settings.TALER_CURRENCY}:1")

        self.assertEqual(user_bankaccount.balance, -amt_one)
        self.assertEqual(user_bankaccount0.balance, amt_one)


class MeasureHistory(TestCase):
    def setUp(self):
        self.user_bankaccount0 = BankAccount(
            user=User.objects.create_user(username="U0"),
            balance=SignedAmount(True, Amount(settings.TALER_CURRENCY, 3000, 0)),
        )
        self.user_bankaccount0.save()

        user_bankaccount = BankAccount(user=User.objects.create_user(username="U"))
        user_bankaccount.save()

        self.ntransfers = 1000

        # Make sure logging level is WARNING, otherwise the loop
        # will overwhelm the console.
        for i in range(self.ntransfers):
            del i  # to pacify PEP checkers
            wire_transfer(
                Amount(settings.TALER_CURRENCY, 1, 0),
                self.user_bankaccount0,
                user_bankaccount,
                "bulk",
            )

    def tearDown(self):
        clear_db()

    def test_extract_history(self):

        # Measure the time extract_history() needs to retrieve
        # ~ntransfers records.
        timer = timeit.Timer(
            stmt="extract_history(self.user_bankaccount0, False)",
            setup="from talerbank.app.views import extract_history",
            globals=locals(),
        )
        total_time = timer.timeit(number=1)
        allowed_time_per_record = 0.003
        self.assertLess(total_time, self.ntransfers * allowed_time_per_record)


class BalanceTestCase(TestCase):
    def setUp(self):
        self.the_bank = BankAccount(
            user=User.objects.create_user(username="U0", password="U0PASS"),
            balance=SignedAmount(True, Amount(settings.TALER_CURRENCY, 3, 0)),
        )
        self.the_bank.save()

        user = BankAccount(
            user=User.objects.create_user(username="U"),
            balance=SignedAmount(True, Amount(settings.TALER_CURRENCY, 10, 0)),
        )
        user.save()

        # bank: 3, user: 10 (START).

        # bank: 2, user: 11
        wire_transfer(
            Amount(settings.TALER_CURRENCY, 1, 0), self.the_bank, user, "mock"
        )

        # bank: 4, user: 9
        wire_transfer(
            Amount(settings.TALER_CURRENCY, 2, 0), user, self.the_bank, "mock"
        )

        # bank: -1, user: 14
        wire_transfer(
            Amount(settings.TALER_CURRENCY, 5, 0), self.the_bank, user, "mock"
        )

        # bank: 7, user: 6 (END)
        wire_transfer(
            Amount(settings.TALER_CURRENCY, 8, 0), user, self.the_bank, "mock"
        )

        # bank: -3, user: 16 (END)
        wire_transfer(
            Amount(settings.TALER_CURRENCY, 10, 0), self.the_bank, user, "mock"
        )

        self.client = Client()

    def tearDown(self):
        clear_db()

    def test_balance(self):
        self.client.login(username="U0", password="U0PASS")
        response = self.client.get(
            reverse("history", urlconf=urls),
            {"delta": -30, "direction": "both", "account_number": 55},
            HTTP_AUTHORIZATION=make_auth_line("U0", "U0PASS"),
        )
        data = response.content.decode("utf-8")
        self.assertEqual(response.status_code, 200)
        entries = json.loads(data)

        acc_bal = SignedAmount(True, Amount(settings.TALER_CURRENCY, 10, 0))
        print("acc_bal start", acc_bal)

        for entry in reversed(entries["data"]):
            print("entry", entry)
            if entry["sign"] == "-":
                acc_bal += SignedAmount.parse(entry["amount"])
            if entry["sign"] == "+":
                acc_bal -= SignedAmount.parse(entry["amount"])
            print("acc_bal after", acc_bal)

        expected_amount = SignedAmount.parse(f"{settings.TALER_CURRENCY}:16.0")
        self.assertEqual(acc_bal, expected_amount)