mirror of
https://github.com/lnbits/lnbits-legend.git
synced 2025-03-10 09:19:42 +01:00
62435b3723
* [FIX] workflows testing why workflwos unstable instable * debug run_command * add timeout to process run * log lncli output * try catch json exception in clnrest * settle does not return a json * add 10 minutes timeout to regtests if they ever get stuck * better subprocess handling * fix race condition * test 3 * remove * run nr 4 * increase waits for race conditions * revert clnrest * better fiat amount test
133 lines
3.4 KiB
Python
133 lines
3.4 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import random
|
|
import string
|
|
import time
|
|
from subprocess import PIPE, Popen, TimeoutExpired
|
|
from typing import Tuple
|
|
|
|
from loguru import logger
|
|
|
|
from lnbits.wallets import get_wallet_class, set_wallet_class
|
|
|
|
|
|
def get_random_string(N: int = 10):
|
|
return "".join(
|
|
random.SystemRandom().choice(string.ascii_uppercase + string.digits)
|
|
for _ in range(N)
|
|
)
|
|
|
|
|
|
async def get_random_invoice_data():
|
|
return {"out": False, "amount": 10, "memo": f"test_memo_{get_random_string(10)}"}
|
|
|
|
|
|
set_wallet_class()
|
|
WALLET = get_wallet_class()
|
|
is_fake: bool = WALLET.__class__.__name__ == "FakeWallet"
|
|
is_regtest: bool = not is_fake
|
|
|
|
|
|
docker_lightning_cli = [
|
|
"docker",
|
|
"exec",
|
|
"lnbits-legend-lnd-1-1",
|
|
"lncli",
|
|
"--network",
|
|
"regtest",
|
|
"--rpcserver=lnd-1",
|
|
]
|
|
|
|
docker_bitcoin_cli = [
|
|
"docker",
|
|
"exec",
|
|
"lnbits-legend-bitcoind-1-1" "bitcoin-cli",
|
|
"-rpcuser=lnbits",
|
|
"-rpcpassword=lnbits",
|
|
"-regtest",
|
|
]
|
|
|
|
|
|
def run_cmd(cmd: list) -> str:
|
|
timeout = 20
|
|
process = Popen(cmd, stdout=PIPE, stderr=PIPE)
|
|
|
|
def process_communication(comm):
|
|
stdout, stderr = comm
|
|
output = stdout.decode("utf-8").strip()
|
|
error = stderr.decode("utf-8").strip()
|
|
return output, error
|
|
|
|
try:
|
|
now = time.time()
|
|
output, error = process_communication(process.communicate(timeout=timeout))
|
|
took = time.time() - now
|
|
logger.debug(f"ran command output: {output}, error: {error}, took: {took}s")
|
|
return output
|
|
except TimeoutExpired:
|
|
process.kill()
|
|
output, error = process_communication(process.communicate())
|
|
logger.error(f"timeout command: {cmd}, output: {output}, error: {error}")
|
|
raise
|
|
|
|
|
|
def run_cmd_json(cmd: list) -> dict:
|
|
output = run_cmd(cmd)
|
|
try:
|
|
return json.loads(output) if output else {}
|
|
except json.decoder.JSONDecodeError:
|
|
logger.error(f"failed to decode json from cmd `{cmd}`: {output}")
|
|
raise
|
|
|
|
|
|
def get_hold_invoice(sats: int) -> Tuple[str, dict]:
|
|
preimage = os.urandom(32)
|
|
preimage_hash = hashlib.sha256(preimage).hexdigest()
|
|
cmd = docker_lightning_cli.copy()
|
|
cmd.extend(["addholdinvoice", preimage_hash, str(sats)])
|
|
json = run_cmd_json(cmd)
|
|
return preimage.hex(), json
|
|
|
|
|
|
def settle_invoice(preimage: str) -> str:
|
|
cmd = docker_lightning_cli.copy()
|
|
cmd.extend(["settleinvoice", preimage])
|
|
return run_cmd(cmd)
|
|
|
|
|
|
def cancel_invoice(preimage_hash: str) -> str:
|
|
cmd = docker_lightning_cli.copy()
|
|
cmd.extend(["cancelinvoice", preimage_hash])
|
|
return run_cmd(cmd)
|
|
|
|
|
|
def get_real_invoice(sats: int) -> dict:
|
|
cmd = docker_lightning_cli.copy()
|
|
cmd.extend(["addinvoice", str(sats)])
|
|
return run_cmd_json(cmd)
|
|
|
|
|
|
def pay_real_invoice(invoice: str) -> str:
|
|
cmd = docker_lightning_cli.copy()
|
|
cmd.extend(["payinvoice", "--force", invoice])
|
|
return run_cmd(cmd)
|
|
|
|
|
|
def mine_blocks(blocks: int = 1) -> str:
|
|
cmd = docker_bitcoin_cli.copy()
|
|
cmd.extend(["-generate", str(blocks)])
|
|
return run_cmd(cmd)
|
|
|
|
|
|
def create_onchain_address(address_type: str = "bech32") -> str:
|
|
cmd = docker_bitcoin_cli.copy()
|
|
cmd.extend(["getnewaddress", address_type])
|
|
return run_cmd(cmd)
|
|
|
|
|
|
def pay_onchain(address: str, sats: int) -> str:
|
|
btc = sats * 0.00000001
|
|
cmd = docker_bitcoin_cli.copy()
|
|
cmd.extend(["sendtoaddress", address, str(btc)])
|
|
return run_cmd(cmd)
|