commit c465c00f6b06715d05adb1dc91e2f3fcc2f51302
parent c9991b9bf82586699abf35c0b6b2cc25c9eb5bf6
Author: Florian Dold <florian@dold.me>
Date: Tue, 3 Nov 2026 00:37:45 +0100
backenddb: remove brittle one-off test
Test should likely not have been commited in the first place.
Diffstat:
2 files changed, 0 insertions(+), 614 deletions(-)
diff --git a/src/backend/meson.build b/src/backend/meson.build
@@ -435,17 +435,3 @@ test_merchant_kyccheck = executable(
)
test('test_merchant_kyccheck', test_merchant_kyccheck, suite: ['backend'])
-
-# Uses a disposable PostgreSQL cluster and real HTTP long polls.
-test(
- 'kyc-refresh',
- find_program('python3'),
- args: [
- files('../backenddb/test_kyc_refresh.py'),
- meson.project_source_root(),
- meson.project_build_root(),
- ],
- depends: [merchant_httpd, test_merchant_kyccheck, gprocedures_sql, iprocedures_sql] + migration_sql,
- suite: ['backenddb', 'integrationtests'],
- timeout: 180,
-)
diff --git a/src/backenddb/test_kyc_refresh.py b/src/backenddb/test_kyc_refresh.py
@@ -1,600 +0,0 @@
-#!/usr/bin/env python3
-
-# This file is part of TALER
-# Copyright (C) 2026 Taler Systems SA
-#
-# TALER is free software; you can redistribute it and/or modify it under the
-# terms of the GNU 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/>
-
-"""Exercise KYC refreshes with real PostgreSQL notifications and merchant HTTPD.
-
-Requires PostgreSQL server tools and Python psycopg2. All database state and
-HTTP listeners belong to the fixture. No installed database is used.
-"""
-
-import base64
-from concurrent.futures import ThreadPoolExecutor
-import hashlib
-import json
-from pathlib import Path
-import select
-import shutil
-import socket
-import struct
-import subprocess
-import sys
-import tempfile
-import time
-import unittest
-from urllib.parse import urlencode
-from urllib.request import Request, urlopen
-
-from test_order_sequence_migrations import Database, postgres_cluster, run
-
-
-ALPHABET = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ234567",
- "0123456789ABCDEFGHJKMNPQRSTVWXYZ")
-EXCHANGE = "http://localhost:1/"
-OTHER_EXCHANGE = "http://localhost:2/"
-# Synthetic exchange master key (private seed is 31 zero bytes followed by 1)
-# and its signature over the account below with no restrictions or gateways.
-MASTER_PUB = bytes.fromhex(
- "4cb5abf6ad79fbf5abbccafcc269d85cd2651ed4b885b5869f241aedf0a5ba29")
-ACCOUNT_SIG = bytes.fromhex(
- "42063a0cc9152e9896bb23b0446e14d9bf6e072fbbdcbf3097ac640eba3744ee"
- "817368fe93ae567c98ad8f5c27ab5022c82bf73cd024a52baa4e2920bca01806")
-FOREVER = 9223372036854775807
-WIRE = bytes([42]) * 64
-TOKEN = bytes([43]) * 32
-
-
-def encode(data):
- return base64.b32encode(data).decode().rstrip("=").translate(ALPHABET)
-
-
-def channel(event_type, body=b""):
- header = struct.pack("!HH", 4 + len(body), event_type)
- return "x" + encode(hashlib.sha512(header + body).digest()[:32]).lower()
-
-
-def public_key(instance):
- return instance.to_bytes(32, "big")
-
-
-# Use the existing protocol event's documented channel, independent of its enum.
-REFRESH = "xj40p0cfmz0dt6sfz70vrq19kg1hp6aj1q6vcdzn4n2fgpsag4kdg"
-
-
-class KycRefresh(unittest.TestCase):
- def sql(self, statement, params=()):
- with self.db.cursor() as cur:
- cur.execute(statement, params or None)
- return cur.fetchall() if cur.description else []
-
- def events(self, wait=0):
- deadline = time.monotonic() + wait
- events = []
- while True:
- self.listener.poll()
- events.extend(self.listener.notifies)
- self.listener.notifies.clear()
- remaining = deadline - time.monotonic()
- if remaining <= 0:
- return events
- select.select([self.listener], [], [], remaining)
-
- def setUp(self):
- self.db = self.connect()
- self.listener = self.connect()
- self.listener.autocommit = True
- with self.listener.cursor() as cur:
- for event in [REFRESH] + [channel(t, public_key(i) + suffix)
- for i in (1, 2)
- for t, suffix in ((1115, b""), (1113, WIRE))]:
- cur.execute('LISTEN "' + event + '"')
- self.sql("SET search_path TO merchant_instance_1")
- # Direct reset avoids generating status events before each scenario.
- for i in (1, 2):
- self.sql(f'SET search_path TO merchant_instance_{i}')
- self.sql("DELETE FROM merchant_tos_accepted")
- self.sql(f"UPDATE merchant_instance_{i}.merchant_kyc SET "
- "kyc_ok=true, exchange_http_status=200, exchange_ec_code=0, "
- "aml_review=false, jaccount_limits='[]', last_rule_gen=1, "
- "access_token=%s, next_kyc_poll=%s, kyc_backoff=0",
- (TOKEN, FOREVER))
- self.sql('SET search_path TO merchant_instance_1')
- self.events()
-
- def tearDown(self):
- self.listener.close()
- self.db.close()
-
- def read(self, refresh, instance=1, exchange=EXCHANGE, wire=WIRE):
- self.sql(f"SET search_path TO merchant_instance_{instance}")
- return self.sql("SELECT * FROM merchant_do_account_kyc_get_status"
- "(%s,%s,%s,%s,%s)",
- (time.time_ns() // 1000, exchange, wire, refresh, instance))
-
- def store(self, instance=1, exchange=EXCHANGE, **changes):
- values = dict(http=200, ec=0, token=TOKEN, limits="[]", aml=False,
- ok=True, generation=1, due=FOREVER, backoff=0)
- values.update(changes)
- self.sql(f"SET search_path TO merchant_instance_{instance}")
- return self.sql("SELECT * FROM merchant_do_account_kyc_set_status"
- "(%s,%s,%s,%s,%s,%s,%s::jsonb,%s,%s,%s,%s,%s,%s,%s)",
- (WIRE, exchange, int(time.time()) * 1000000,
- values['http'], values['ec'], values['token'],
- values['limits'], values['aml'], values['ok'],
- channel(1113, public_key(instance) + WIRE),
- channel(1115, public_key(instance)), values['generation'],
- values['due'], values['backoff']))
-
- def test_read_only_and_transactional_refresh(self):
- self.assertEqual(1, len(self.read(False)))
- self.assertEqual([], self.events())
- self.assertEqual([(FOREVER,)], self.sql("SELECT next_kyc_poll FROM merchant_kyc"))
- self.assertIsNone(self.read(True, exchange="http://unmatched.invalid/")[0][2])
- self.assertEqual([], self.events())
- self.db.autocommit = False
- self.read(True)
- self.assertEqual([], self.events())
- self.db.rollback()
- self.assertEqual([], self.events())
- self.db.autocommit = True
- self.read(True)
- events = self.events(0.1)
- self.assertEqual([(REFRESH, encode(struct.pack('!Q', 1)))],
- [(e.channel, e.payload) for e in events])
- due = self.sql("SELECT next_kyc_poll FROM merchant_kyc")
- self.read(False)
- self.assertEqual(due, self.sql("SELECT next_kyc_poll FROM merchant_kyc"))
- self.assertEqual([], self.events())
-
- def test_status_change_notifications(self):
- self.store(due=17, backoff=123)
- self.assertEqual([], self.events())
- changes = dict(http=202, ec=123, token=None, limits=None, aml=True,
- ok=False, generation=2)
- current = {}
- for field, value in changes.items():
- with self.subTest(field=field):
- current[field] = value
- self.store(**current)
- self.assertEqual({channel(1113, public_key(1) + WIRE),
- channel(1115, public_key(1))},
- {e.channel for e in self.events(0.1)})
- self.store(**current)
- self.assertEqual([], self.events())
- self.store()
- self.assertEqual(2, len(self.events(0.1))) # Includes NULL -> value.
- self.sql("DELETE FROM merchant_kyc")
- self.store()
- self.assertEqual(2, len(self.events(0.1)))
- args = (WIRE, EXCHANGE, int(time.time()) * 1000000, 502, False,
- channel(1113, public_key(1) + WIRE), channel(1115, public_key(1)))
- for expected in (2, 0):
- self.sql("SELECT * FROM merchant_do_account_kyc_set_failed"
- "(%s,%s,%s,%s,%s,%s,%s)", args)
- self.assertEqual(expected, len(self.events(0.1)))
-
- def test_refresh_only_accesses_target_schema(self):
- self.read(True, instance=1)
- self.read(True, instance=2)
- self.db.autocommit = False
- rows = self.sql("SELECT * FROM merchant.account_kyc_get_outdated(%s,1)",
- (time.time_ns() // 1000,))
- self.assertEqual([('test-1', WIRE, EXCHANGE)],
- [(i, bytes(w), e) for i, w, e in rows])
- touched = self.sql("SELECT DISTINCT n.nspname FROM pg_locks l "
- "JOIN pg_class c ON c.oid=l.relation "
- "JOIN pg_namespace n ON n.oid=c.relnamespace "
- "WHERE l.pid=pg_backend_pid() "
- "AND n.nspname LIKE 'merchant_instance_%'")
- self.assertEqual([('merchant_instance_1',)], touched)
- self.db.rollback()
- self.db.autocommit = True
- self.assertEqual([], self.sql(
- "SELECT * FROM merchant.account_kyc_get_outdated(%s,999999)", (FOREVER,)))
- self.sql("SET search_path TO merchant_instance_1")
- self.sql("UPDATE merchant_accounts SET active=false")
- try:
- self.assertEqual([], self.sql(
- "SELECT * FROM merchant.account_kyc_get_outdated(%s,1)", (FOREVER,)))
- finally:
- self.sql("UPDATE merchant_accounts SET active=true")
-
- def get(self, instance=1, query=""):
- with urlopen(self.url + f"instances/test-{instance}/private/kyc" + query,
- timeout=8) as response:
- return json.load(response), response.headers['ETag']
-
- def wait_for_refresh(self):
- deadline = time.monotonic() + 5
- while time.monotonic() < deadline:
- if any(e.channel == REFRESH for e in self.events(0.05)):
- return
- self.fail("HTTP request did not prompt its initial refresh")
-
- def test_long_poll_rereads_do_not_refresh(self):
- self.check_long_poll()
-
- def test_account_long_poll_rereads_do_not_refresh(self):
- self.check_long_poll("&h_wire=" + encode(WIRE))
-
- def check_long_poll(self, account_filter=""):
- body, etag = self.get()
- self.assertEqual('ready', body['kyc_data'][0]['status'])
- self.wait_for_refresh()
- with ThreadPoolExecutor() as pool:
- pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag="
- + etag.strip('"') + account_filter)
- self.wait_for_refresh()
- # A real status change that does not affect the response ETag forces
- # an internal reread, which must not request another exchange check.
- self.store(generation=2)
- self.assertFalse(any(e.channel == REFRESH for e in self.events(0.25)))
- self.assertFalse(pending.done())
- # Another instance's change must not wake this long poll either.
- self.store(instance=2, http=202, ok=False)
- self.assertFalse(any(e.channel == REFRESH for e in self.events(0.25)))
- self.assertFalse(pending.done())
- self.store(http=202, ok=False, generation=2)
- changed, _ = pending.result(timeout=3)
- self.assertEqual('kyc-required', changed['kyc_data'][0]['status'])
- self.assertFalse(any(e.channel == REFRESH for e in self.events(0.1)))
- self.get()
- self.wait_for_refresh() # Separate HTTP requests still refresh.
-
- def install_keys(self, exchange=EXCHANGE, **changes):
- keys = dict(
- version='33:0:0', currency='EUR', asset_type='fiat',
- master_public_key=encode(MASTER_PUB),
- reserve_closing_delay={'d_us': 3600000000},
- list_issue_date={'t_s': int(time.time())},
- global_fees=[], signkeys=[], denominations=[], auditors=[],
- wire_fees={}, wads=[], kyc_enabled=True, kyc_swap_tos_acceptance=False,
- hard_limits=[], zero_limits=[], stefan_abs='EUR:0', stefan_log='EUR:0',
- stefan_lin=0.0, currency_specification=dict(
- name='Euro', num_fractional_input_digits=2,
- num_fractional_normal_digits=2, num_fractional_trailing_zero_digits=2,
- alt_unit_names={'0': 'EUR'}),
- accounts=[dict(
- payto_uri='payto://x-taler-bank/localhost/exchange?receiver-name=Exchange',
- credit_restrictions=[], debit_restrictions=[], master_sig=encode(ACCOUNT_SIG))])
- keys.update(changes)
- self.sql("INSERT INTO merchant.merchant_exchange_keys "
- "(exchange_url,keys_json,first_retry,expiration_time,"
- "exchange_http_status,exchange_ec_code) VALUES (%s,%s,0,0,200,0) "
- "ON CONFLICT (exchange_url) DO UPDATE SET keys_json=EXCLUDED.keys_json",
- (exchange, json.dumps(dict(version=0, exchange_url=exchange,
- expire={'t_s': int(time.time()) + 3600}, keys=keys))))
- self.sql("SELECT pg_notify(%s,%s)",
- (channel(1110), encode((exchange + '\0').encode())))
-
- def test_initial_keys_wake_long_poll(self):
- query = '?' + urlencode({'exchange_url': OTHER_EXCHANGE})
- self.store(instance=2, exchange=OTHER_EXCHANGE,
- http=404, token=None, limits=None, ok=False, generation=0)
- try:
- body, etag = self.get(instance=2, query=query)
- self.assertTrue(body['kyc_data'][0]['no_keys'])
- self.wait_for_refresh()
- with ThreadPoolExecutor() as pool:
- pending = pool.submit(self.get, 2, query + "&timeout_ms=5000&lp_not_etag="
- + etag.strip('"'))
- self.wait_for_refresh()
- self.install_keys(exchange=OTHER_EXCHANGE)
- body, new_etag = pending.result(timeout=2)
- self.assertNotEqual(etag, new_etag)
- self.assertFalse(body['kyc_data'][0]['no_keys'])
- self.assertEqual('kyc-wire-required', body['kyc_data'][0]['status'])
- self.assertEqual([], self.events(0.1))
- finally:
- self.sql("SET search_path TO merchant_instance_2")
- self.sql("DELETE FROM merchant_kyc WHERE exchange_url=%s", (OTHER_EXCHANGE,))
-
- def test_keys_long_poll(self):
- self.check_keys_long_poll()
-
- def test_keys_exchange_long_poll(self):
- self.check_keys_long_poll("&" + urlencode({'exchange_url': EXCHANGE}))
-
- def test_keys_account_long_poll(self):
- self.check_keys_long_poll("&h_wire=" + encode(WIRE))
-
- def check_keys_long_poll(self, account_filter=""):
- status = dict(http=404, token=None, limits=None, ok=False, generation=0)
- self.store(**status)
- self.install_keys()
- # Wait until HTTPD has consumed the key notification. The keys go
- # through the production deserializer, including signature validation.
- deadline = time.monotonic() + 5
- while True:
- body, etag = self.get()
- self.wait_for_refresh()
- data = body['kyc_data'][0]
- if not data['no_keys'] and not data['kyc_swap_tos_acceptance'] and data['limits'] == []:
- break
- self.assertLess(time.monotonic(), deadline)
- time.sleep(0.02)
- self.assertEqual('kyc-wire-required', data['status'])
- self.assertTrue(data['payto_kycauths'])
- cases = [
- ('kyc_swap_tos_acceptance', True),
- ('hard_limits', [dict(operation_type='DEPOSIT', threshold='EUR:10',
- timeframe={'d_us': 60000000}, soft_limit=False)]),
- ('zero_limits', [dict(operation_type='WITHDRAW')]),
- ('accounts', []),
- ]
- changes = {}
- with ThreadPoolExecutor() as pool:
- for field, value in cases:
- with self.subTest(field=field):
- pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag="
- + etag.strip('"') + account_filter)
- self.wait_for_refresh()
- # Repeated or unrelated keys do not finish this poll or
- # request another exchange KYC check.
- self.install_keys(**changes)
- self.install_keys(exchange=OTHER_EXCHANGE, kyc_swap_tos_acceptance=True)
- self.assertFalse(any(e.channel == REFRESH for e in self.events(0.15)))
- self.assertFalse(pending.done())
- changes[field] = value
- self.install_keys(**changes)
- self.store(**status) # The same cached /kyc-check result.
- body, new_etag = pending.result(timeout=2)
- self.assertNotEqual(etag, new_etag)
- etag = new_etag
- data = body['kyc_data'][0]
- if field == 'kyc_swap_tos_acceptance':
- self.assertTrue(data[field])
- elif field == 'hard_limits':
- self.assertEqual('EUR:10', data['limits'][0]['threshold'])
- elif field == 'zero_limits':
- self.assertTrue(data['limits'][1]['disallowed'])
- else:
- self.assertEqual('kyc-wire-impossible', data['status'])
- self.assertFalse(data.get('payto_kycauths'))
- self.assertEqual([], self.events(0.1))
-
- def accept_tos(self, version, instance=1):
- request = Request(self.url + f"instances/test-{instance}/private/accept-tos-early",
- data=json.dumps(dict(exchange_url=EXCHANGE,
- tos_version=version)).encode(),
- headers={'Content-Type': 'application/json'})
- with urlopen(request, timeout=3) as response:
- self.assertEqual(204, response.status)
-
- def set_tos(self, version):
- return self.sql("SELECT merchant_do_set_tos_accepted_early(%s,%s,%s,%s)",
- (EXCHANGE, version, struct.pack('!HH', 100, 1113) + public_key(1),
- channel(1115, public_key(1))))
-
- def test_tos_notifications_are_transactional_and_scoped(self):
- # Even accounts without a cached KYC row can report early acceptance.
- extra_wire = bytes([44]) * 64
- extra_channel = channel(1113, public_key(1) + extra_wire)
- self.sql("INSERT INTO merchant_accounts(h_wire,salt,payto_uri,active) "
- "VALUES(%s,%s,'payto://x-taler-bank/localhost/extra?receiver-name=Test',true)",
- (extra_wire, bytes(16)))
- with self.listener.cursor() as cur:
- cur.execute('LISTEN "' + extra_channel + '"')
- self.events()
- try:
- self.db.autocommit = False
- self.assertEqual([(True,)], self.set_tos('v1'))
- self.assertEqual([], self.events(0.1))
- self.db.rollback()
- self.db.autocommit = True
- self.assertEqual([], self.events())
- self.assertEqual([], self.sql("SELECT * FROM merchant_tos_accepted"))
- for version in ('v1', 'v2', None):
- self.assertEqual([(True,)], self.set_tos(version))
- self.assertEqual({channel(1115, public_key(1)),
- channel(1113, public_key(1) + WIRE), extra_channel},
- {e.channel for e in self.events(0.1)})
- self.assertEqual([(False,)], self.set_tos(version))
- self.assertEqual([], self.events(0.1))
- finally:
- self.sql("DELETE FROM merchant_accounts WHERE h_wire=%s", (extra_wire,))
-
- def test_tos_long_poll(self):
- self.check_tos_long_poll()
-
- def test_tos_account_long_poll(self):
- self.check_tos_long_poll("&h_wire=" + encode(WIRE))
-
- def check_tos_long_poll(self, account_filter=""):
- self.store(http=202, ok=False)
- self.events(0.1)
- body, etag = self.get()
- self.assertNotIn('tos_accepted_early', body['kyc_data'][0])
- self.wait_for_refresh()
- with ThreadPoolExecutor() as pool:
- for version in ('v1', 'v2', None):
- with self.subTest(version=version):
- pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag="
- + etag.strip('"') + account_filter)
- self.wait_for_refresh()
- if version == 'v2':
- self.accept_tos('v1') # An identical acceptance is silent.
- self.assertEqual([], self.events(0.1))
- self.accept_tos(version or 'v3', instance=2)
- self.assertFalse(any(e.channel == REFRESH for e in self.events(0.1)))
- self.assertFalse(pending.done())
- if version is None:
- # The actual 409 callback clears acceptance; the following
- # unchanged KYC write must not be needed to wake the poll.
- self.transition('tos-conflict')
- self.store(http=202, ok=False)
- else:
- self.accept_tos(version)
- body, new_etag = pending.result(timeout=2)
- self.assertNotEqual(etag, new_etag)
- etag = new_etag
- self.assertEqual(version, body['kyc_data'][0].get('tos_accepted_early'))
- self.assertEqual('kyc-required', body['kyc_data'][0]['status'])
- self.assertEqual({channel(1115, public_key(1)),
- channel(1113, public_key(1) + WIRE)},
- {e.channel for e in self.events(0.1)})
-
- def test_eligibility_loss_wakes_long_poll(self):
- body, etag = self.get()
- self.assertEqual('ready', body['kyc_data'][0]['status'])
- self.wait_for_refresh()
- with ThreadPoolExecutor() as pool:
- pending = pool.submit(self.get, 1, "?timeout_ms=5000&lp_not_etag="
- + etag.strip('"'))
- self.wait_for_refresh()
- self.transition('eligibility')
- body, new_etag = pending.result(timeout=2)
- self.assertNotEqual(etag, new_etag)
- # /keys snapshots are injected into kyccheck, while HTTPD has no
- # keys in this fixture. Verify that it returns the persisted
- # ineligibility error and drops the stale ready status/token.
- self.assertEqual('no-exchange-keys', body['kyc_data'][0]['status'])
- self.assertEqual(2628, body['kyc_data'][0]['exchange_code'])
- self.assertNotIn('access_token', body['kyc_data'][0])
- self.assertEqual({channel(1115, public_key(1)),
- channel(1113, public_key(1) + WIRE)},
- {e.channel for e in self.events(0.1)})
- body, _ = self.get(instance=2)
- self.assertEqual('ready', body['kyc_data'][0]['status'])
-
- def test_upgrade_existing_procedure_signature(self):
- # Procedure synchronization must replace the old three-argument API.
- self.sql("CREATE FUNCTION merchant_do_account_kyc_get_status(bigint,text,bytea) "
- "RETURNS integer LANGUAGE sql AS 'SELECT 0'")
- self.sql('CALL merchant.sync_instance_procedures(1)')
- self.assertEqual([(None,)], self.sql(
- "SELECT to_regprocedure('merchant_do_account_kyc_get_status(bigint,text,bytea)')"))
- self.assertEqual(1, len(self.read(False)))
-
-
-def main():
- try:
- import psycopg2
- except ImportError:
- print("KYC refresh integration tests require Python psycopg2")
- return 77
- source, build = (Path(arg).resolve() for arg in sys.argv[1:])
- if not shutil.which('pg_config'):
- return 77
- bindir = Path(run(['pg_config', '--bindir']))
- with postgres_cluster(bindir, max_locks=1024) as env, tempfile.TemporaryDirectory(
- prefix='merchant-kyc-') as tmp:
- sql_dir = build / 'src/backenddb/sql-schema'
- db = Database('talercheck', bindir, env, sql_dir)
- run(['createdb', 'talercheck'], env=env)
- db.apply_file(source / 'src/backenddb/sql-schema/versioning.sql')
- for migration in sorted(sql_dir.glob('merchant-????.sql')):
- db.apply_file(migration)
- db.apply_file(sql_dir / 'global_procedures.sql')
- db.apply_file(sql_dir / 'instance_procedures.sql')
- for i in range(1, 201):
- db.sql(f"""
- INSERT INTO merchant.merchant_instances
- (merchant_serial, merchant_id, merchant_name, merchant_pub,
- merchant_priv, address, jurisdiction, default_wire_transfer_delay,
- default_pay_delay, use_stefan)
- VALUES ({i}, 'test-{i}', 'Test', decode('{public_key(i).hex()}','hex'),
- decode('{public_key(i).hex()}','hex'), '{{}}', '{{}}', 1, 1, false);
- SET search_path TO merchant_instance_{i};
- INSERT INTO merchant_instance_{i}.merchant_accounts
- (h_wire, salt, payto_uri, active)
- VALUES (decode('{WIRE.hex()}','hex'), decode(repeat('00',16),'hex'),
- 'payto://x-taler-bank/localhost/account-{i}?receiver-name=Test', true);
- INSERT INTO merchant_instance_{i}.merchant_kyc
- (account_serial,exchange_url,kyc_timestamp,kyc_ok,access_token,
- exchange_http_status,jaccount_limits,last_rule_gen,next_kyc_poll)
- VALUES (1,'{EXCHANGE}',0,true,decode('{TOKEN.hex()}','hex'),200,'[]',1,{FOREVER});
- """)
- def connect():
- conn = psycopg2.connect(dbname='talercheck', user='postgres',
- host=env['PGHOST'], port=env['PGPORT'])
- conn.autocommit = True
- return conn
- KycRefresh.connect = staticmethod(connect)
- with socket.socket() as sock:
- sock.bind(('127.0.0.1', 0))
- port = sock.getsockname()[1]
- KycRefresh.url = f'http://127.0.0.1:{port}/'
- config = Path(tmp, 'merchant.conf')
- config.write_text(f"""
-@INLINE@ {source / "src/backend/merchant.conf"}
-@INLINE@ {source / "src/util/currencies.conf"}
-[merchant]
-CURRENCY=EUR
-SERVE=tcp
-PORT={port}
-BIND_TO=127.0.0.1
-BASE_URL={KycRefresh.url}
-[merchantdb-postgres]
-CONFIG=postgres:///talercheck
-SQL_DIR={sql_dir}/
-[taler]
-CURRENCY=EUR
-[merchant-exchange-kyc-test]
-EXCHANGE_BASE_URL={EXCHANGE}
-CURRENCY=EUR
-MASTER_KEY={encode(MASTER_PUB)}
-[merchant-exchange-other-test]
-EXCHANGE_BASE_URL={OTHER_EXCHANGE}
-CURRENCY=EUR
-MASTER_KEY={encode(MASTER_PUB)}
-[merchant-exchange-kudos]
-DISABLED=YES
-[merchant-exchange-chf]
-DISABLED=YES
-""")
- env['LD_LIBRARY_PATH'] = ':'.join(str(build / 'src' / d)
- for d in ('backenddb', 'util', 'bank', 'lib')) \
- + ':' + env.get('LD_LIBRARY_PATH', '')
- prefix = Path(tmp, 'prefix')
- resources = prefix / 'share/taler-merchant'
- resources.mkdir(parents=True)
- (resources / 'templates').symlink_to(source / 'src/frontend')
- (resources / 'spa').symlink_to(source / 'contrib/spa')
- env['TALER_MERCHANT_PREFIX'] = str(prefix)
- KycRefresh.transition = staticmethod(lambda mode: run(
- [str(build / 'src/backend/test_merchant_kyccheck'), str(config), mode], env=env))
- with open(Path(tmp, 'httpd.log'), 'w+') as log:
- process = subprocess.Popen([str(build / 'src/backend/taler-merchant-httpd'),
- '-c', str(config), '-L', 'INFO'],
- env=env, stdout=log, stderr=log)
- try:
- deadline = time.monotonic() + 10
- while time.monotonic() < deadline:
- if process.poll() is not None:
- raise RuntimeError('merchant HTTPD exited at startup')
- try:
- with urlopen(KycRefresh.url + 'config', timeout=0.2):
- break
- except OSError:
- time.sleep(0.05)
- else:
- raise RuntimeError('merchant HTTPD did not start')
- suite = unittest.defaultTestLoader.loadTestsFromTestCase(KycRefresh)
- result = unittest.TextTestRunner(verbosity=2).run(suite)
- return 0 if result.wasSuccessful() else 1
- finally:
- process.terminate()
- try:
- process.wait(timeout=10)
- except subprocess.TimeoutExpired:
- process.kill()
- process.wait()
- log.seek(0)
- Path(build / 'kyc-refresh-httpd.log').write_text(log.read())
-
-
-if __name__ == '__main__':
- sys.exit(main())
