Integrate Persist into ChainMonitor.
[rust-lightning] / lightning / src / util / test_utils.rs
index 6b3649bcf0ecc34075dffaa86b00f4982387a5dd..8643ef0f205d9ad7b44d2d9ef86ec3c41ad65824 100644 (file)
@@ -1,28 +1,44 @@
+// This file is Copyright its original authors, visible in version control
+// history.
+//
+// This file is licensed under the Apache License, Version 2.0 <LICENSE-APACHE
+// or http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
+// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your option.
+// You may not use this file except in accordance with one or both of these
+// licenses.
+
+use chain;
 use chain::chaininterface;
 use chain::chaininterface::ConfirmationTarget;
+use chain::chainmonitor;
+use chain::channelmonitor;
+use chain::channelmonitor::MonitorEvent;
 use chain::transaction::OutPoint;
 use chain::keysinterface;
-use ln::channelmonitor;
-use ln::features::InitFeatures;
+use ln::features::{ChannelFeatures, InitFeatures};
 use ln::msgs;
-use ln::msgs::LightningError;
-use ln::channelmonitor::HTLCUpdate;
+use ln::msgs::OptionalField;
 use util::enforcing_trait_impls::EnforcingChannelKeys;
 use util::events;
 use util::logger::{Logger, Level, Record};
-use util::ser::{ReadableArgs, Writer};
+use util::ser::{Readable, Writer, Writeable};
 
-use bitcoin::blockdata::transaction::Transaction;
-use bitcoin::blockdata::script::Script;
-use bitcoin_hashes::sha256d::Hash as Sha256dHash;
+use bitcoin::blockdata::constants::genesis_block;
+use bitcoin::blockdata::transaction::{Transaction, TxOut};
+use bitcoin::blockdata::script::{Builder, Script};
+use bitcoin::blockdata::opcodes;
 use bitcoin::network::constants::Network;
+use bitcoin::hash_types::{BlockHash, Txid};
+
+use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1, Signature};
 
-use secp256k1::{SecretKey, PublicKey};
+use regex;
 
-use std::time::{SystemTime, UNIX_EPOCH};
-use std::sync::{Arc,Mutex};
-use std::{mem};
-use std::collections::HashMap;
+use std::time::Duration;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::{cmp, mem};
+use std::collections::{HashMap, HashSet};
 
 pub struct TestVecWriter(pub Vec<u8>);
 impl Writer for TestVecWriter {
@@ -36,45 +52,97 @@ impl Writer for TestVecWriter {
 }
 
 pub struct TestFeeEstimator {
-       pub sat_per_kw: u64,
+       pub sat_per_kw: u32,
 }
 impl chaininterface::FeeEstimator for TestFeeEstimator {
-       fn get_est_sat_per_1000_weight(&self, _confirmation_target: ConfirmationTarget) -> u64 {
+       fn get_est_sat_per_1000_weight(&self, _confirmation_target: ConfirmationTarget) -> u32 {
                self.sat_per_kw
        }
 }
 
-pub struct TestChannelMonitor {
-       pub added_monitors: Mutex<Vec<(OutPoint, channelmonitor::ChannelMonitor)>>,
-       pub simple_monitor: Arc<channelmonitor::SimpleManyChannelMonitor<OutPoint>>,
+pub struct TestChainMonitor<'a> {
+       pub added_monitors: Mutex<Vec<(OutPoint, channelmonitor::ChannelMonitor<EnforcingChannelKeys>)>>,
+       pub latest_monitor_update_id: Mutex<HashMap<[u8; 32], (OutPoint, u64)>>,
+       pub chain_monitor: chainmonitor::ChainMonitor<EnforcingChannelKeys, &'a TestChainSource, &'a chaininterface::BroadcasterInterface, &'a TestFeeEstimator, &'a TestLogger, &'a channelmonitor::Persist<EnforcingChannelKeys>>,
        pub update_ret: Mutex<Result<(), channelmonitor::ChannelMonitorUpdateErr>>,
+       // If this is set to Some(), after the next return, we'll always return this until update_ret
+       // is changed:
+       pub next_update_ret: Mutex<Option<Result<(), channelmonitor::ChannelMonitorUpdateErr>>>,
 }
-impl TestChannelMonitor {
-       pub fn new(chain_monitor: Arc<chaininterface::ChainWatchInterface>, broadcaster: Arc<chaininterface::BroadcasterInterface>, logger: Arc<Logger>, fee_estimator: Arc<chaininterface::FeeEstimator>) -> Self {
+impl<'a> TestChainMonitor<'a> {
+       pub fn new(chain_source: Option<&'a TestChainSource>, broadcaster: &'a chaininterface::BroadcasterInterface, logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator, persister: &'a channelmonitor::Persist<EnforcingChannelKeys>) -> Self {
                Self {
                        added_monitors: Mutex::new(Vec::new()),
-                       simple_monitor: channelmonitor::SimpleManyChannelMonitor::new(chain_monitor, broadcaster, logger, fee_estimator),
+                       latest_monitor_update_id: Mutex::new(HashMap::new()),
+                       chain_monitor: chainmonitor::ChainMonitor::new(chain_source, broadcaster, logger, fee_estimator, persister),
                        update_ret: Mutex::new(Ok(())),
+                       next_update_ret: Mutex::new(None),
                }
        }
 }
-impl channelmonitor::ManyChannelMonitor for TestChannelMonitor {
-       fn add_update_monitor(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+impl<'a> chain::Watch for TestChainMonitor<'a> {
+       type Keys = EnforcingChannelKeys;
+
+       fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<EnforcingChannelKeys>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
                // At every point where we get a monitor update, we should be able to send a useful monitor
                // to a watchtower and disk...
                let mut w = TestVecWriter(Vec::new());
                monitor.write_for_disk(&mut w).unwrap();
-               assert!(<(Sha256dHash, channelmonitor::ChannelMonitor)>::read(
-                               &mut ::std::io::Cursor::new(&w.0), Arc::new(TestLogger::new())).unwrap().1 == monitor);
+               let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingChannelKeys>)>::read(
+                       &mut ::std::io::Cursor::new(&w.0)).unwrap().1;
+               assert!(new_monitor == monitor);
+               self.latest_monitor_update_id.lock().unwrap().insert(funding_txo.to_channel_id(), (funding_txo, monitor.get_latest_update_id()));
+               self.added_monitors.lock().unwrap().push((funding_txo, monitor));
+               assert!(self.chain_monitor.watch_channel(funding_txo, new_monitor).is_ok());
+
+               let ret = self.update_ret.lock().unwrap().clone();
+               if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
+                       *self.update_ret.lock().unwrap() = next_ret;
+               }
+               ret
+       }
+
+       fn update_channel(&self, funding_txo: OutPoint, update: channelmonitor::ChannelMonitorUpdate) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+               // Every monitor update should survive roundtrip
+               let mut w = TestVecWriter(Vec::new());
+               update.write(&mut w).unwrap();
+               assert!(channelmonitor::ChannelMonitorUpdate::read(
+                               &mut ::std::io::Cursor::new(&w.0)).unwrap() == update);
+
+               self.latest_monitor_update_id.lock().unwrap().insert(funding_txo.to_channel_id(), (funding_txo, update.update_id));
+               assert!(self.chain_monitor.update_channel(funding_txo, update).is_ok());
+               // At every point where we get a monitor update, we should be able to send a useful monitor
+               // to a watchtower and disk...
+               let monitors = self.chain_monitor.monitors.lock().unwrap();
+               let monitor = monitors.get(&funding_txo).unwrap();
                w.0.clear();
-               monitor.write_for_watchtower(&mut w).unwrap(); // This at least shouldn't crash...
-               self.added_monitors.lock().unwrap().push((funding_txo, monitor.clone()));
-               assert!(self.simple_monitor.add_update_monitor(funding_txo, monitor).is_ok());
-               self.update_ret.lock().unwrap().clone()
+               monitor.write_for_disk(&mut w).unwrap();
+               let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingChannelKeys>)>::read(
+                               &mut ::std::io::Cursor::new(&w.0)).unwrap().1;
+               assert!(new_monitor == *monitor);
+               self.added_monitors.lock().unwrap().push((funding_txo, new_monitor));
+
+               let ret = self.update_ret.lock().unwrap().clone();
+               if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
+                       *self.update_ret.lock().unwrap() = next_ret;
+               }
+               ret
        }
 
-       fn fetch_pending_htlc_updated(&self) -> Vec<HTLCUpdate> {
-               return self.simple_monitor.fetch_pending_htlc_updated();
+       fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
+               return self.chain_monitor.release_pending_monitor_events();
+       }
+}
+
+pub struct TestPersister {}
+
+impl channelmonitor::Persist<EnforcingChannelKeys> for TestPersister {
+       fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<EnforcingChannelKeys>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+               Ok(())
+       }
+
+       fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<EnforcingChannelKeys>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+               Ok(())
        }
 }
 
@@ -130,30 +198,107 @@ impl events::MessageSendEventsProvider for TestChannelMessageHandler {
        }
 }
 
-pub struct TestRoutingMessageHandler {}
+fn get_dummy_channel_announcement(short_chan_id: u64) -> msgs::ChannelAnnouncement {
+       use bitcoin::secp256k1::ffi::Signature as FFISignature;
+       let secp_ctx = Secp256k1::new();
+       let network = Network::Testnet;
+       let node_1_privkey = SecretKey::from_slice(&[42; 32]).unwrap();
+       let node_2_privkey = SecretKey::from_slice(&[41; 32]).unwrap();
+       let node_1_btckey = SecretKey::from_slice(&[40; 32]).unwrap();
+       let node_2_btckey = SecretKey::from_slice(&[39; 32]).unwrap();
+       let unsigned_ann = msgs::UnsignedChannelAnnouncement {
+               features: ChannelFeatures::known(),
+               chain_hash: genesis_block(network).header.block_hash(),
+               short_channel_id: short_chan_id,
+               node_id_1: PublicKey::from_secret_key(&secp_ctx, &node_1_privkey),
+               node_id_2: PublicKey::from_secret_key(&secp_ctx, &node_2_privkey),
+               bitcoin_key_1: PublicKey::from_secret_key(&secp_ctx, &node_1_btckey),
+               bitcoin_key_2: PublicKey::from_secret_key(&secp_ctx, &node_2_btckey),
+               excess_data: Vec::new(),
+       };
+
+       msgs::ChannelAnnouncement {
+               node_signature_1: Signature::from(FFISignature::new()),
+               node_signature_2: Signature::from(FFISignature::new()),
+               bitcoin_signature_1: Signature::from(FFISignature::new()),
+               bitcoin_signature_2: Signature::from(FFISignature::new()),
+               contents: unsigned_ann,
+       }
+}
+
+fn get_dummy_channel_update(short_chan_id: u64) -> msgs::ChannelUpdate {
+       use bitcoin::secp256k1::ffi::Signature as FFISignature;
+       let network = Network::Testnet;
+       msgs::ChannelUpdate {
+               signature: Signature::from(FFISignature::new()),
+               contents: msgs::UnsignedChannelUpdate {
+                       chain_hash: genesis_block(network).header.block_hash(),
+                       short_channel_id: short_chan_id,
+                       timestamp: 0,
+                       flags: 0,
+                       cltv_expiry_delta: 0,
+                       htlc_minimum_msat: 0,
+                       htlc_maximum_msat: OptionalField::Absent,
+                       fee_base_msat: 0,
+                       fee_proportional_millionths: 0,
+                       excess_data: vec![],
+               }
+       }
+}
+
+pub struct TestRoutingMessageHandler {
+       pub chan_upds_recvd: AtomicUsize,
+       pub chan_anns_recvd: AtomicUsize,
+       pub chan_anns_sent: AtomicUsize,
+       pub request_full_sync: AtomicBool,
+}
 
 impl TestRoutingMessageHandler {
        pub fn new() -> Self {
-               TestRoutingMessageHandler {}
+               TestRoutingMessageHandler {
+                       chan_upds_recvd: AtomicUsize::new(0),
+                       chan_anns_recvd: AtomicUsize::new(0),
+                       chan_anns_sent: AtomicUsize::new(0),
+                       request_full_sync: AtomicBool::new(false),
+               }
        }
 }
 impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
-       fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<bool, LightningError> {
-               Err(LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+       fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<bool, msgs::LightningError> {
+               Err(msgs::LightningError { err: "".to_owned(), action: msgs::ErrorAction::IgnoreError })
        }
-       fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, LightningError> {
-               Err(LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+       fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, msgs::LightningError> {
+               self.chan_anns_recvd.fetch_add(1, Ordering::AcqRel);
+               Err(msgs::LightningError { err: "".to_owned(), action: msgs::ErrorAction::IgnoreError })
        }
-       fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<bool, LightningError> {
-               Err(LightningError { err: "", action: msgs::ErrorAction::IgnoreError })
+       fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<bool, msgs::LightningError> {
+               self.chan_upds_recvd.fetch_add(1, Ordering::AcqRel);
+               Err(msgs::LightningError { err: "".to_owned(), action: msgs::ErrorAction::IgnoreError })
        }
        fn handle_htlc_fail_channel_update(&self, _update: &msgs::HTLCFailChannelUpdate) {}
-       fn get_next_channel_announcements(&self, _starting_point: u64, _batch_amount: u8) -> Vec<(msgs::ChannelAnnouncement, msgs::ChannelUpdate,msgs::ChannelUpdate)> {
-               Vec::new()
+       fn get_next_channel_announcements(&self, starting_point: u64, batch_amount: u8) -> Vec<(msgs::ChannelAnnouncement, Option<msgs::ChannelUpdate>, Option<msgs::ChannelUpdate>)> {
+               let mut chan_anns = Vec::new();
+               const TOTAL_UPDS: u64 = 100;
+               let end: u64 = cmp::min(starting_point + batch_amount as u64, TOTAL_UPDS - self.chan_anns_sent.load(Ordering::Acquire) as u64);
+               for i in starting_point..end {
+                       let chan_upd_1 = get_dummy_channel_update(i);
+                       let chan_upd_2 = get_dummy_channel_update(i);
+                       let chan_ann = get_dummy_channel_announcement(i);
+
+                       chan_anns.push((chan_ann, Some(chan_upd_1), Some(chan_upd_2)));
+               }
+
+               self.chan_anns_sent.fetch_add(chan_anns.len(), Ordering::AcqRel);
+               chan_anns
        }
+
        fn get_next_node_announcements(&self, _starting_point: Option<&PublicKey>, _batch_amount: u8) -> Vec<msgs::NodeAnnouncement> {
                Vec::new()
        }
+
+       fn should_request_full_sync(&self, _node_id: &PublicKey) -> bool {
+               self.request_full_sync.load(Ordering::Acquire)
+       }
 }
 
 pub struct TestLogger {
@@ -180,6 +325,30 @@ impl TestLogger {
                let log_entries = self.lines.lock().unwrap();
                assert_eq!(log_entries.get(&(module, line)), Some(&count));
        }
+
+       /// Search for the number of occurrence of the logged lines which
+       /// 1. belongs to the specified module and
+       /// 2. contains `line` in it.
+       /// And asserts if the number of occurrences is the same with the given `count`
+       pub fn assert_log_contains(&self, module: String, line: String, count: usize) {
+               let log_entries = self.lines.lock().unwrap();
+               let l: usize = log_entries.iter().filter(|&(&(ref m, ref l), _c)| {
+                       m == &module && l.contains(line.as_str())
+               }).map(|(_, c) | { c }).sum();
+               assert_eq!(l, count)
+       }
+
+    /// Search for the number of occurrences of logged lines which
+    /// 1. belong to the specified module and
+    /// 2. match the given regex pattern.
+    /// Assert that the number of occurrences equals the given `count`
+       pub fn assert_log_regex(&self, module: String, pattern: regex::Regex, count: usize) {
+               let log_entries = self.lines.lock().unwrap();
+               let l: usize = log_entries.iter().filter(|&(&(ref m, ref l), _c)| {
+                       m == &module && pattern.is_match(&l)
+               }).map(|(_, c) | { c }).sum();
+               assert_eq!(l, count)
+       }
 }
 
 impl Logger for TestLogger {
@@ -193,7 +362,7 @@ impl Logger for TestLogger {
 
 pub struct TestKeysInterface {
        backing: keysinterface::KeysManager,
-       pub override_session_priv: Mutex<Option<SecretKey>>,
+       pub override_session_priv: Mutex<Option<[u8; 32]>>,
        pub override_channel_id_priv: Mutex<Option<[u8; 32]>>,
 }
 
@@ -203,30 +372,75 @@ impl keysinterface::KeysInterface for TestKeysInterface {
        fn get_node_secret(&self) -> SecretKey { self.backing.get_node_secret() }
        fn get_destination_script(&self) -> Script { self.backing.get_destination_script() }
        fn get_shutdown_pubkey(&self) -> PublicKey { self.backing.get_shutdown_pubkey() }
-       fn get_channel_keys(&self, inbound: bool) -> EnforcingChannelKeys { EnforcingChannelKeys::new(self.backing.get_channel_keys(inbound)) }
-
-       fn get_onion_rand(&self) -> (SecretKey, [u8; 32]) {
-               match *self.override_session_priv.lock().unwrap() {
-                       Some(key) => (key.clone(), [0; 32]),
-                       None => self.backing.get_onion_rand()
-               }
+       fn get_channel_keys(&self, inbound: bool, channel_value_satoshis: u64) -> EnforcingChannelKeys {
+               EnforcingChannelKeys::new(self.backing.get_channel_keys(inbound, channel_value_satoshis))
        }
 
-       fn get_channel_id(&self) -> [u8; 32] {
-               match *self.override_channel_id_priv.lock().unwrap() {
-                       Some(key) => key.clone(),
-                       None => self.backing.get_channel_id()
+       fn get_secure_random_bytes(&self) -> [u8; 32] {
+               let override_channel_id = self.override_channel_id_priv.lock().unwrap();
+               let override_session_key = self.override_session_priv.lock().unwrap();
+               if override_channel_id.is_some() && override_session_key.is_some() {
+                       panic!("We don't know which override key to use!");
+               }
+               if let Some(key) = &*override_channel_id {
+                       return *key;
                }
+               if let Some(key) = &*override_session_key {
+                       return *key;
+               }
+               self.backing.get_secure_random_bytes()
        }
 }
 
 impl TestKeysInterface {
-       pub fn new(seed: &[u8; 32], network: Network, logger: Arc<Logger>) -> Self {
-               let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("Time went backwards");
+       pub fn new(seed: &[u8; 32], network: Network) -> Self {
+               let now = Duration::from_secs(genesis_block(network).header.time as u64);
                Self {
-                       backing: keysinterface::KeysManager::new(seed, network, logger, now.as_secs(), now.subsec_nanos()),
+                       backing: keysinterface::KeysManager::new(seed, network, now.as_secs(), now.subsec_nanos()),
                        override_session_priv: Mutex::new(None),
                        override_channel_id_priv: Mutex::new(None),
                }
        }
+       pub fn derive_channel_keys(&self, channel_value_satoshis: u64, user_id_1: u64, user_id_2: u64) -> EnforcingChannelKeys {
+               EnforcingChannelKeys::new(self.backing.derive_channel_keys(channel_value_satoshis, user_id_1, user_id_2))
+       }
+}
+
+pub struct TestChainSource {
+       pub genesis_hash: BlockHash,
+       pub utxo_ret: Mutex<Result<TxOut, chain::AccessError>>,
+       pub watched_txn: Mutex<HashSet<(Txid, Script)>>,
+       pub watched_outputs: Mutex<HashSet<(OutPoint, Script)>>,
+}
+
+impl TestChainSource {
+       pub fn new(network: Network) -> Self {
+               let script_pubkey = Builder::new().push_opcode(opcodes::OP_TRUE).into_script();
+               Self {
+                       genesis_hash: genesis_block(network).block_hash(),
+                       utxo_ret: Mutex::new(Ok(TxOut { value: u64::max_value(), script_pubkey })),
+                       watched_txn: Mutex::new(HashSet::new()),
+                       watched_outputs: Mutex::new(HashSet::new()),
+               }
+       }
+}
+
+impl chain::Access for TestChainSource {
+       fn get_utxo(&self, genesis_hash: &BlockHash, _short_channel_id: u64) -> Result<TxOut, chain::AccessError> {
+               if self.genesis_hash != *genesis_hash {
+                       return Err(chain::AccessError::UnknownChain);
+               }
+
+               self.utxo_ret.lock().unwrap().clone()
+       }
+}
+
+impl chain::Filter for TestChainSource {
+       fn register_tx(&self, txid: &Txid, script_pubkey: &Script) {
+               self.watched_txn.lock().unwrap().insert((*txid, script_pubkey.clone()));
+       }
+
+       fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script) {
+               self.watched_outputs.lock().unwrap().insert((*outpoint, script_pubkey.clone()));
+       }
 }