mirror of
https://github.com/ElementsProject/lightning.git
synced 2024-11-19 18:11:28 +01:00
90527498bc
CI always runs with TEST_DEBUG=1 which prints logs anyway, and testing locally should also be done this way, combined with pytest which captures the logs. No need to duplicate the functionality of pytest. Signed-off-by: Christian Decker <decker.christian@gmail.com>
382 lines
13 KiB
Python
382 lines
13 KiB
Python
from bitcoin.rpc import RawProxy as BitcoinProxy
|
|
from lightning import LightningRpc
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
|
|
BITCOIND_CONFIG = {
|
|
"rpcuser": "rpcuser",
|
|
"rpcpassword": "rpcpass",
|
|
"rpcport": 18332,
|
|
}
|
|
|
|
|
|
LIGHTNINGD_CONFIG = {
|
|
"bitcoind-poll": "1s",
|
|
"log-level": "debug",
|
|
"cltv-delta": 6,
|
|
"cltv-final": 5,
|
|
"locktime-blocks": 5,
|
|
}
|
|
|
|
DEVELOPER = os.getenv("DEVELOPER", "0") == "1"
|
|
|
|
def write_config(filename, opts):
|
|
with open(filename, 'w') as f:
|
|
for k, v in opts.items():
|
|
f.write("{}={}\n".format(k, v))
|
|
|
|
|
|
class TailableProc(object):
|
|
"""A monitorable process that we can start, stop and tail.
|
|
|
|
This is the base class for the daemons. It allows us to directly
|
|
tail the processes and react to their output.
|
|
"""
|
|
|
|
def __init__(self, outputDir=None):
|
|
self.logs = []
|
|
self.logs_cond = threading.Condition(threading.RLock())
|
|
self.cmd_line = None
|
|
self.env = os.environ
|
|
self.running = False
|
|
self.proc = None
|
|
self.outputDir = outputDir
|
|
self.logsearch_start = 0
|
|
|
|
def start(self):
|
|
"""Start the underlying process and start monitoring it.
|
|
"""
|
|
logging.debug("Starting '%s'", " ".join(self.cmd_line))
|
|
self.proc = subprocess.Popen(self.cmd_line, stdout=subprocess.PIPE, env=self.env)
|
|
self.thread = threading.Thread(target=self.tail)
|
|
self.thread.daemon = True
|
|
self.thread.start()
|
|
self.running = True
|
|
|
|
def save_log(self):
|
|
if self.outputDir:
|
|
logpath = os.path.join(self.outputDir, 'log')
|
|
with open(logpath, 'w') as f:
|
|
for l in self.logs:
|
|
f.write(l + '\n')
|
|
|
|
def stop(self, timeout=10):
|
|
self.save_log()
|
|
self.proc.terminate()
|
|
|
|
# Now give it some time to react to the signal
|
|
rc = self.proc.wait(timeout)
|
|
|
|
if rc is None:
|
|
self.proc.kill()
|
|
|
|
self.proc.wait()
|
|
self.thread.join()
|
|
|
|
if self.proc.returncode:
|
|
raise ValueError("Process '{}' did not cleanly shutdown: return code {}".format(self.proc.pid, rc))
|
|
|
|
return self.proc.returncode
|
|
|
|
def tail(self):
|
|
"""Tail the stdout of the process and remember it.
|
|
|
|
Stores the lines of output produced by the process in
|
|
self.logs and signals that a new line was read so that it can
|
|
be picked up by consumers.
|
|
"""
|
|
for line in iter(self.proc.stdout.readline, ''):
|
|
if len(line) == 0:
|
|
break
|
|
with self.logs_cond:
|
|
self.logs.append(str(line.rstrip()))
|
|
logging.debug("%s: %s", self.prefix, line.decode().rstrip())
|
|
self.logs_cond.notifyAll()
|
|
self.running = False
|
|
self.proc.stdout.close()
|
|
|
|
def is_in_log(self, regex):
|
|
"""Look for `regex` in the logs."""
|
|
|
|
ex = re.compile(regex)
|
|
for l in self.logs:
|
|
if ex.search(l):
|
|
logging.debug("Found '%s' in logs", regex)
|
|
return True
|
|
|
|
logging.debug("Did not find '%s' in logs", regex)
|
|
return False
|
|
|
|
def wait_for_logs(self, regexs, timeout=60):
|
|
"""Look for `regexs` in the logs.
|
|
|
|
We tail the stdout of the process and look for each regex in `regexs`,
|
|
starting from last of the previous waited-for log entries (if any). We
|
|
fail if the timeout is exceeded or if the underlying process
|
|
exits before all the `regexs` were found.
|
|
"""
|
|
logging.debug("Waiting for {} in the logs".format(regexs))
|
|
exs = [re.compile(r) for r in regexs]
|
|
start_time = time.time()
|
|
pos = self.logsearch_start
|
|
initial_pos = len(self.logs)
|
|
while True:
|
|
if time.time() > start_time + timeout:
|
|
print("Can't find {} in logs".format(exs))
|
|
for r in exs:
|
|
if self.is_in_log(r):
|
|
print("({} was previously in logs!)".format(r))
|
|
raise TimeoutError('Unable to find "{}" in logs.'.format(exs))
|
|
elif not self.running:
|
|
raise ValueError('Process died while waiting for logs')
|
|
|
|
with self.logs_cond:
|
|
if pos >= len(self.logs):
|
|
self.logs_cond.wait(1)
|
|
continue
|
|
|
|
for r in exs.copy():
|
|
self.logsearch_start = pos+1
|
|
if r.search(self.logs[pos]):
|
|
logging.debug("Found '%s' in logs", r)
|
|
exs.remove(r)
|
|
break
|
|
if len(exs) == 0:
|
|
return self.logs[pos]
|
|
pos += 1
|
|
|
|
def wait_for_log(self, regex, timeout=60):
|
|
"""Look for `regex` in the logs.
|
|
|
|
Convenience wrapper for the common case of only seeking a single entry.
|
|
"""
|
|
return self.wait_for_logs([regex], timeout)
|
|
|
|
|
|
class SimpleBitcoinProxy:
|
|
"""Wrapper for BitcoinProxy to reconnect.
|
|
|
|
Long wait times between calls to the Bitcoin RPC could result in
|
|
`bitcoind` closing the connection, so here we just create
|
|
throwaway connections. This is easier than to reach into the RPC
|
|
library to close, reopen and reauth upon failure.
|
|
"""
|
|
def __init__(self, btc_conf_file, *args, **kwargs):
|
|
self.__btc_conf_file__= btc_conf_file
|
|
|
|
def __getattr__(self, name):
|
|
if name.startswith('__') and name.endswith('__'):
|
|
# Python internal stuff
|
|
raise AttributeError
|
|
|
|
# Create a callable to do the actual call
|
|
proxy = BitcoinProxy(btc_conf_file=self.__btc_conf_file__)
|
|
|
|
f = lambda *args: proxy._call(name, *args)
|
|
|
|
# Make debuggers show <function bitcoin.rpc.name> rather than <function
|
|
# bitcoin.rpc.<lambda>>
|
|
f.__name__ = name
|
|
return f
|
|
|
|
|
|
class BitcoinD(TailableProc):
|
|
|
|
def __init__(self, bitcoin_dir="/tmp/bitcoind-test", rpcport=18332):
|
|
TailableProc.__init__(self, bitcoin_dir)
|
|
|
|
self.bitcoin_dir = bitcoin_dir
|
|
self.rpcport = rpcport
|
|
self.prefix = 'bitcoind'
|
|
|
|
regtestdir = os.path.join(bitcoin_dir, 'regtest')
|
|
if not os.path.exists(regtestdir):
|
|
os.makedirs(regtestdir)
|
|
|
|
self.cmd_line = [
|
|
'bitcoind',
|
|
'-datadir={}'.format(bitcoin_dir),
|
|
'-printtoconsole',
|
|
'-server',
|
|
'-regtest',
|
|
'-logtimestamps',
|
|
'-nolisten',
|
|
]
|
|
BITCOIND_CONFIG['rpcport'] = rpcport
|
|
btc_conf_file = os.path.join(regtestdir, 'bitcoin.conf')
|
|
write_config(os.path.join(bitcoin_dir, 'bitcoin.conf'), BITCOIND_CONFIG)
|
|
write_config(btc_conf_file, BITCOIND_CONFIG)
|
|
self.rpc = SimpleBitcoinProxy(btc_conf_file=btc_conf_file)
|
|
|
|
def start(self):
|
|
TailableProc.start(self)
|
|
self.wait_for_log("Done loading", timeout=10)
|
|
|
|
logging.info("BitcoinD started")
|
|
|
|
def generate_block(self, numblocks=1):
|
|
# As of 0.16, generate() is removed; use generatetoaddress.
|
|
self.rpc.generatetoaddress(numblocks, self.rpc.getnewaddress())
|
|
|
|
# lightning-1 => 0266e4598d1d3c415f572a8488830b60f7e744ed9235eb0b1ba93283b315c03518 aka JUNIORBEAM #0266e4
|
|
# lightning-2 => 022d223620a359a47ff7f7ac447c85c46c923da53389221a0054c11c1e3ca31d59 aka SILENTARTIST #022d22
|
|
# lightning-3 => 035d2b1192dfba134e10e540875d366ebc8bc353d5aa766b80c090b39c3a5d885d aka HOPPINGFIRE #035d2b
|
|
# lightning-4 => 0382ce59ebf18be7d84677c2e35f23294b9992ceca95491fcf8a56c6cb2d9de199 aka JUNIORFELONY #0382ce
|
|
# lightning-5 => 032cf15d1ad9c4a08d26eab1918f732d8ef8fdc6abb9640bf3db174372c491304e aka SOMBERFIRE #032cf1
|
|
|
|
class LightningD(TailableProc):
|
|
def __init__(self, lightning_dir, bitcoin_dir, port=9735, random_hsm=False):
|
|
TailableProc.__init__(self, lightning_dir)
|
|
self.lightning_dir = lightning_dir
|
|
self.port = port
|
|
# Last 32-bytes of final part of dir -> seed.
|
|
seed = (bytes(re.search('([^/]+)/*$', lightning_dir).group(1), encoding='utf-8') + bytes(32))[:32]
|
|
self.cmd_line = [
|
|
'lightningd/lightningd',
|
|
'--bitcoin-datadir={}'.format(bitcoin_dir),
|
|
'--lightning-dir={}'.format(lightning_dir),
|
|
'--port={}'.format(port),
|
|
'--override-fee-rates=15000/7500/1000',
|
|
'--network=regtest'
|
|
]
|
|
if DEVELOPER:
|
|
self.cmd_line += ['--dev-broadcast-interval=1000']
|
|
if not random_hsm:
|
|
self.cmd_line += ['--dev-hsm-seed={}'.format(seed.hex())]
|
|
self.cmd_line += ["--{}={}".format(k, v) for k, v in sorted(LIGHTNINGD_CONFIG.items())]
|
|
self.prefix = 'lightningd(%d)' % (port)
|
|
|
|
if not os.path.exists(lightning_dir):
|
|
os.makedirs(lightning_dir)
|
|
|
|
def start(self):
|
|
TailableProc.start(self)
|
|
self.wait_for_log("Hello world from")
|
|
logging.info("LightningD started")
|
|
|
|
def wait(self, timeout=10):
|
|
"""Wait for the daemon to stop for up to timeout seconds
|
|
|
|
Returns the returncode of the process, None if the process did
|
|
not return before the timeout triggers.
|
|
"""
|
|
self.proc.wait(timeout)
|
|
return self.proc.returncode
|
|
|
|
class LightningNode(object):
|
|
def __init__(self, daemon, rpc, btc, executor, may_fail=False):
|
|
self.rpc = rpc
|
|
self.daemon = daemon
|
|
self.bitcoin = btc
|
|
self.executor = executor
|
|
self.may_fail = may_fail
|
|
|
|
# Use batch if you're doing more than one async.
|
|
def connect(self, remote_node, capacity, async=False):
|
|
# Collect necessary information
|
|
addr = self.rpc.newaddr()['address']
|
|
txid = self.bitcoin.rpc.sendtoaddress(addr, capacity)
|
|
tx = self.bitcoin.rpc.gettransaction(txid)
|
|
start_size = self.bitcoin.rpc.getmempoolinfo()['size']
|
|
|
|
def call_connect():
|
|
try:
|
|
self.rpc.connect('127.0.0.1', remote_node.daemon.port, tx['hex'], async=False)
|
|
except:
|
|
pass
|
|
t = threading.Thread(target=call_connect)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
def wait_connected():
|
|
# Up to 10 seconds to get tx into mempool.
|
|
start_time = time.time()
|
|
while self.bitcoin.rpc.getmempoolinfo()['size'] == start_size:
|
|
if time.time() > start_time + 10:
|
|
raise TimeoutError('No new transactions in mempool')
|
|
time.sleep(0.1)
|
|
|
|
self.bitcoin.generate_block(1)
|
|
|
|
#fut.result(timeout=5)
|
|
|
|
# Now wait for confirmation
|
|
self.daemon.wait_for_log("-> CHANNELD_NORMAL|STATE_NORMAL")
|
|
remote_node.daemon.wait_for_log("-> CHANNELD_NORMAL|STATE_NORMAL")
|
|
|
|
if async:
|
|
return self.executor.submit(wait_connected)
|
|
else:
|
|
return wait_connected()
|
|
|
|
def openchannel(self, remote_node, capacity):
|
|
addr = self.rpc.newaddr()['address']
|
|
txid = self.bitcoin.rpc.sendtoaddress(addr, capacity / 10**6)
|
|
tx = self.bitcoin.rpc.getrawtransaction(txid)
|
|
self.rpc.addfunds(tx)
|
|
self.rpc.fundchannel(remote_node.info['id'], capacity)
|
|
self.daemon.wait_for_log('sendrawtx exit 0, gave')
|
|
time.sleep(1)
|
|
self.bitcoin.generate_block(6)
|
|
self.daemon.wait_for_log('-> CHANNELD_NORMAL|STATE_NORMAL')
|
|
|
|
def db_query(self, query):
|
|
db = sqlite3.connect(os.path.join(self.daemon.lightning_dir, "lightningd.sqlite3"))
|
|
db.row_factory = sqlite3.Row
|
|
c = db.cursor()
|
|
c.execute(query)
|
|
rows = c.fetchall()
|
|
|
|
result = []
|
|
for row in rows:
|
|
result.append(dict(zip(row.keys(), row)))
|
|
|
|
c.close()
|
|
db.close()
|
|
return result
|
|
|
|
def stop(self, timeout=10):
|
|
""" Attempt to do a clean shutdown, but kill if it hangs
|
|
"""
|
|
|
|
# Tell the daemon to stop
|
|
try:
|
|
# May fail if the process already died
|
|
self.rpc.stop()
|
|
except:
|
|
pass
|
|
|
|
rc = self.daemon.wait(timeout)
|
|
|
|
# If it did not stop be more insistent
|
|
if rc is None:
|
|
rc = self.daemon.stop()
|
|
|
|
self.daemon.save_log()
|
|
|
|
if rc != 0 and not self.may_fail:
|
|
raise ValueError("Node did not exit cleanly, rc={}".format(rc))
|
|
else:
|
|
return rc
|
|
|
|
def restart(self, timeout=10, clean=True):
|
|
"""Stop and restart the lightning node.
|
|
|
|
Keyword arguments:
|
|
timeout: number of seconds to wait for a shutdown
|
|
clean: whether to issue a `stop` RPC command before killing
|
|
"""
|
|
if clean:
|
|
self.stop(timeout)
|
|
else:
|
|
self.daemon.stop()
|
|
|
|
self.daemon.start()
|