1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
|
import os
import secrets
from dataclasses import dataclass
from subprocess import run
import psutil
import requests
from .taler_service import TalerService
@dataclass
class BankUser:
username: str
password: str
@dataclass
class WithdrawUriResponse:
taler_withdraw_uri: str
withdrawal_id: str
class Bank(TalerService):
def __init__(self, config, watcher_getter, request):
super().__init__(config, watcher_getter, request)
# get localhost port and store bank URL
r = run(["taler-config", "-c", config.conf, "-s", "BANK", "-o", "HTTP_PORT"],
check=True, text=True, capture_output=True)
self.url = "http://localhost:%s/" % r.stdout.rstrip()
def start(self):
db = "postgres:///%s" % self.config.db
log_path = os.path.join(self.config.tmpdir, "bank.log")
log_file = open(log_path, 'w')
self.watcher_getter(
name='taler-bank-manage-testing',
arguments=[self.config.conf, db, 'serve-http'],
checker=self.test_process,
kwargs=dict(stderr=log_file, stdout=log_file),
request=self.request, # Needed for the correct execution order of finalizers
)
def close_log():
log_file.close()
self.request.addfinalizer(close_log)
def register_random_user(self):
username = f"testuser-{secrets.token_hex(10)}"
password = f"testpw-{secrets.token_hex(10)}"
requests.post(
f"{self.url}testing/register",
json=dict(username=username, password=password)
)
return BankUser(username, password)
def generate_withdraw_uri(self, bankuser, amount):
auth = (bankuser.username, bankuser.password)
resp = requests.post(
f"{self.url}accounts/{bankuser.username}/withdrawals",
json=dict(amount=amount),
auth=auth
)
rj = resp.json()
return WithdrawUriResponse(
taler_withdraw_uri=rj["taler_withdraw_uri"],
withdrawal_id=rj["withdrawal_id"],
)
def confirm_withdrawal(self, bank_user, withdrawal_id):
auth = (bank_user.username, bank_user.password)
requests.post(
f"{self.url}accounts/{bank_user.username}/withdrawals/{withdrawal_id}/confirm",
auth=auth
)
# Alternative way to check if the bank came up.
# Testing the URL has the issue that on the CI, django keeps closing the connection.
@staticmethod
def test_process():
for p in psutil.process_iter(['name', 'cmdline']):
if p.info["name"] == "uwsgi" and p.info["cmdline"][-1] == "talerbank.wsgi":
return True
return False
|
