First cloud test

src/beacon.rs

@@ -12,11 +12,12 @@ use std::fs::{self, Permissions, File};
 use std::os::unix::fs::PermissionsExt;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
+use std::marker::PhantomData;
 use std::mem;
 use std::thread;
 use std::process::{Command, Stdio};
 
-use super::util::{now, Encoder};
+use super::util::{Encoder, TimeSource};
 use std::net::{SocketAddr, SocketAddrV4, Ipv4Addr, SocketAddrV6, Ipv6Addr};
 
 
@@ -33,34 +34,36 @@ fn sha512(data: &[u8]) -> Vec<u8> {
     digest::digest(&digest::SHA512, data).as_ref().iter().map(|b| *b).collect()
 }
 
-fn now_hour_16() -> u16 {
-    ((now() / 3600) & 0xffff) as u16
-}
-
 struct FutureResult<T> {
     has_result: AtomicBool,
     result: Mutex<T>
 }
 
 #[derive(Clone)]
-pub struct BeaconSerializer {
+pub struct BeaconSerializer<TS> {
     magic: Vec<u8>,
     shared_key: Vec<u8>,
     future_peers: Arc<FutureResult<Vec<SocketAddr>>>,
+    _dummy_ts: PhantomData<TS>
 }
 
-impl BeaconSerializer {
+impl<TS: TimeSource> BeaconSerializer<TS> {
     pub fn new(magic: &[u8], shared_key: &[u8]) -> Self {
-        BeaconSerializer {
+        Self {
             magic: magic.to_owned(),
             shared_key: shared_key.to_owned(),
             future_peers: Arc::new(FutureResult {
                 has_result: AtomicBool::new(false),
                 result: Mutex::new(Vec::new())
-            })
+            }),
+            _dummy_ts: PhantomData
         }
     }
 
+    fn now_hour_16() -> u16 {
+        ((TS::now() / 3600) & 0xffff) as u16
+    }
+
     fn get_keystream(&self, type_: u8, seed: u8, iter: u8) -> Vec<u8> {
         let mut data = Vec::new();
         data.extend_from_slice(&[type_, seed, iter]);
@@ -110,10 +113,10 @@ impl BeaconSerializer {
         seed == sha512(data as &[u8])[0]
     }
 
-    fn peerlist_encode(&self, peers: &[SocketAddr], now_hour: u16) -> String {
+    fn peerlist_encode(&self, peers: &[SocketAddr]) -> String {
         let mut data = Vec::new();
         // Add timestamp
-        data.extend_from_slice(&now_hour.to_be_bytes());
+        data.extend_from_slice(&Self::now_hour_16().to_be_bytes());
         // Split addresses into v4 and v6
         let mut v4addrs = Vec::new();
         let mut v6addrs = Vec::new();
@@ -151,7 +154,7 @@ impl BeaconSerializer {
         base_62::encode(&data)
     }
 
-    fn peerlist_decode(&self, data: &str, ttl_hours: Option<u16>, now_hour: u16) -> Vec<SocketAddr> {
+    fn peerlist_decode(&self, data: &str, ttl_hours: Option<u16>) -> Vec<SocketAddr> {
         let mut data = base_62::decode(data).expect("Invalid input");
         let mut peers = Vec::new();
         let mut pos = 0;
@@ -163,7 +166,7 @@ impl BeaconSerializer {
         }
         let then = Wrapping(Encoder::read_u16(&data[pos..=pos+1]));
         if let Some(ttl) = ttl_hours {
-            let now = Wrapping(now_hour);
+            let now = Wrapping(Self::now_hour_16());
             if now - then > Wrapping(ttl) && then - now > Wrapping(ttl) {
                 return peers
             }
@@ -199,12 +202,8 @@ impl BeaconSerializer {
         peers
     }
 
-    fn encode_internal(&self, peers: &[SocketAddr], now_hour: u16) -> String {
-        format!("{}{}{}", self.begin(), self.peerlist_encode(peers, now_hour), self.end())
-    }
-
     pub fn encode(&self, peers: &[SocketAddr]) -> String {
-        self.encode_internal(peers, now_hour_16())
+        format!("{}{}{}", self.begin(), self.peerlist_encode(peers), self.end())
     }
 
     pub fn write_to_file<P: AsRef<Path>>(&self, peers: &[SocketAddr], path: P) -> Result<(), io::Error> {
@@ -218,7 +217,7 @@ impl BeaconSerializer {
 
     pub fn write_to_cmd(&self, peers: &[SocketAddr], cmd: &str) -> Result<(), io::Error> {
         let begin = self.begin();
-        let data = self.peerlist_encode(peers, now_hour_16());
+        let data = self.peerlist_encode(peers);
         let end = self.end();
         let beacon = format!("{}{}{}", begin, data, end);
         debug!("Calling beacon command: {}", cmd);
@@ -236,7 +235,7 @@ impl BeaconSerializer {
         Ok(())
     }
 
-    fn decode_internal(&self, data: &str, ttl_hours: Option<u16>, now_hour: u16) -> Vec<SocketAddr> {
+    pub fn decode(&self, data: &str, ttl_hours: Option<u16>) -> Vec<SocketAddr> {
         let data = base_62_sanitize(data);
         let mut peers = Vec::new();
         let begin = self.begin();
@@ -247,7 +246,7 @@ impl BeaconSerializer {
             let start_pos = pos + begin.len();
             if let Some(found) = data[pos..].find(&end) {
                 let end_pos = pos + found;
-                peers.append(&mut self.peerlist_decode(&data[start_pos..end_pos], ttl_hours, now_hour));
+                peers.append(&mut self.peerlist_decode(&data[start_pos..end_pos], ttl_hours));
                 pos = start_pos
             } else {
                 break
@@ -256,10 +255,6 @@ impl BeaconSerializer {
         peers
     }
 
-    pub fn decode(&self, data: &str, ttl_hours: Option<u16>) -> Vec<SocketAddr> {
-        self.decode_internal(data, ttl_hours, now_hour_16())
-    }
-
     pub fn read_from_file<P: AsRef<Path>>(&self, path: P, ttl_hours: Option<u16>) -> Result<Vec<SocketAddr>, io::Error> {
         let mut f = try!(File::open(&path));
         let mut contents = String::new();
@@ -306,94 +301,112 @@ impl BeaconSerializer {
 #[cfg(test)] use std::str::FromStr;
 #[cfg(test)] use std::time::Duration;
 #[cfg(test)] use tempfile;
+#[cfg(test)] use ::util::MockTimeSource;
 
 #[test]
 fn encode() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", ser.encode_internal(&peers, 2000));
+    assert_eq!("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", ser.encode(&peers));
     peers.push(SocketAddr::from_str("[::1]:5678").unwrap());
-    assert_eq!("juWwKjF5qZG7PE5imnpi5XARaXnP3UsMsGBLxM4FNFDzvjlKt1SO55LN", ser.encode_internal(&peers, 2000));
+    assert_eq!("juWwKjF5qZG7PE5imnpi5XARaXnP3UsMsGBLxM4FNFDzvjlKt1SO55LN", ser.encode(&peers));
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:54").unwrap());
-    assert_eq!("juWwKIgSqTammVFRNoIVzLPO0BEO55LN", ser.encode_internal(&peers, 2000));
+    assert_eq!("juWwKIgSqTammVFRNoIVzLPO0BEO55LN", ser.encode(&peers));
 }
 
 #[test]
 fn decode() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None)));
     peers.push(SocketAddr::from_str("[::1]:5678").unwrap());
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("juWwKjF5qZG7PE5imnpi5XARaXnP3UsMsGBLxM4FNFDzvjlKt1SO55LN", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("juWwKjF5qZG7PE5imnpi5XARaXnP3UsMsGBLxM4FNFDzvjlKt1SO55LN", None)));
 }
 
 #[test]
 fn decode_split() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("juWwK-hj.VT:Yj bw\tJj\ntY(AZ)lM[fE]j7üIDäO55LN", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("juWwK-hj.VT:Yj bw\tJj\ntY(AZ)lM[fE]j7üIDäO55LN", None)));
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("j -, \nuW--wKhjVTYjbwJjtYAZlMfEj7IDO(5}5ÖÄÜ\nLN", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("j -, \nuW--wKhjVTYjbwJjtYAZlMfEj7IDO(5}5ÖÄÜ\nLN", None)));
 }
 
 #[test]
 fn decode_offset() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("Hello World: juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN! End of the World", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("Hello World: juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN! End of the World", None)));
 }
 
 #[test]
 fn decode_multiple() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode_internal("juWwKkBEVBp9SsDiN3BO55LN juWwKtGGPQz1gXIBd68O55LN", None, 2000)));
+    assert_eq!(format!("{:?}", peers), format!("{:?}", ser.decode("juWwKkBEVBp9SsDiN3BO55LN juWwKtGGPQz1gXIBd68O55LN", None)));
 }
 
 #[test]
 fn decode_ttl() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 2000).len());
+    MockTimeSource::set_time(2000*3600);
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 2100).len());
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None).len());
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 2005).len());
+    MockTimeSource::set_time(2100*3600);
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 1995).len());
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None).len());
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24), 2000).len());
+    MockTimeSource::set_time(2005*3600);
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24), 1995).len());
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None).len());
-    assert_eq!(2, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24), 2005).len());
+    MockTimeSource::set_time(1995*3600);
-    assert_eq!(0, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24), 2100).len());
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None).len());
-    assert_eq!(0, ser.decode_internal("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24), 1900).len());
+    MockTimeSource::set_time(2000*3600);
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24)).len());
+    MockTimeSource::set_time(1995*3600);
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24)).len());
+    MockTimeSource::set_time(2005*3600);
+    assert_eq!(2, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24)).len());
+    MockTimeSource::set_time(2100*3600);
+    assert_eq!(0, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24)).len());
+    MockTimeSource::set_time(1900*3600);
+    assert_eq!(0, ser.decode("juWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", Some(24)).len());
 }
 
 #[test]
 fn decode_invalid() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
-    assert_eq!(0, ser.decode_internal("", None, 2000).len());
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
-    assert_eq!(0, ser.decode_internal("juWwKO55LN", None, 2000).len());
+    assert_eq!(0, ser.decode("", None).len());
-    assert_eq!(0, ser.decode_internal("juWwK--", None, 2000).len());
+    assert_eq!(0, ser.decode("juWwKO55LN", None).len());
-    assert_eq!(0, ser.decode_internal("--O55LN", None, 2000).len());
+    assert_eq!(0, ser.decode("juWwK--", None).len());
-    assert_eq!(0, ser.decode_internal("juWwKhjVTYjbwJjtYAZXMfEj7IDO55LN", None, 2000).len());
+    assert_eq!(0, ser.decode("--O55LN", None).len());
-    assert_eq!(2, ser.decode_internal("SGrivjuWwKhjVTYjbwJjtYAZlMfEj7IDO55LNjuWwK", None, 2000).len());
+    assert_eq!(0, ser.decode("juWwKhjVTYjbwJjtYAZXMfEj7IDO55LN", None).len());
-    assert_eq!(2, ser.decode_internal("juWwKjuWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None, 2000).len());
+    assert_eq!(2, ser.decode("SGrivjuWwKhjVTYjbwJjtYAZlMfEj7IDO55LNjuWwK", None).len());
+    assert_eq!(2, ser.decode("juWwKjuWwKhjVTYjbwJjtYAZlMfEj7IDO55LN", None).len());
 }
 
 
 #[test]
 fn encode_decode() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
@@ -404,7 +417,8 @@ fn encode_decode() {
 
 #[test]
 fn encode_decode_file() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());
@@ -417,7 +431,8 @@ fn encode_decode_file() {
 
 #[test]
 fn encode_decode_cmd() {
-    let ser = BeaconSerializer::new(b"vpnc", b"mysecretkey");
+    MockTimeSource::set_time(2000*3600);
+    let ser = BeaconSerializer::<MockTimeSource>::new(b"vpnc", b"mysecretkey");
     let mut peers = Vec::new();
     peers.push(SocketAddr::from_str("1.2.3.4:5678").unwrap());
     peers.push(SocketAddr::from_str("6.6.6.6:53").unwrap());

src/cloud.rs

@@ -21,7 +21,7 @@ use super::device::Device;
 use super::udpmessage::{encode, decode, Message};
 use super::crypto::Crypto;
 use super::port_forwarding::PortForwarding;
-use super::util::{now, Time, Duration, resolve, CtrlC};
+use super::util::{TimeSource, Time, Duration, resolve, CtrlC};
 use super::poll::{WaitImpl, WaitResult};
 use super::traffic::TrafficStats;
 use super::beacon::BeaconSerializer;
@@ -40,25 +40,27 @@ struct PeerData {
     alt_addrs: Vec<SocketAddr>,
 }
 
-struct PeerList {
+struct PeerList<TS: TimeSource> {
     timeout: Duration,
     peers: HashMap<SocketAddr, PeerData, Hash>,
     nodes: HashMap<NodeId, SocketAddr, Hash>,
-    addresses: HashMap<SocketAddr, NodeId, Hash>
+    addresses: HashMap<SocketAddr, NodeId, Hash>,
+    _dummy_ts: PhantomData<TS>
 }
 
-impl PeerList {
+impl<TS: TimeSource> PeerList<TS> {
-    fn new(timeout: Duration) -> PeerList {
+    fn new(timeout: Duration) -> PeerList<TS> {
         PeerList{
             peers: HashMap::default(),
             timeout,
             nodes: HashMap::default(),
-            addresses: HashMap::default()
+            addresses: HashMap::default(),
+            _dummy_ts: PhantomData
         }
     }
 
     fn timeout(&mut self) -> Vec<SocketAddr> {
-        let now = now();
+        let now = TS::now();
         let mut del: Vec<SocketAddr> = Vec::new();
         for (&addr, ref data) in &self.peers {
             if data.timeout < now {
@@ -104,7 +106,7 @@ impl PeerList {
         if self.nodes.insert(node_id, addr).is_none() {
             info!("New peer: {}", addr);
             self.peers.insert(addr, PeerData {
-                timeout: now() + Time::from(self.timeout),
+                timeout: TS::now() + Time::from(self.timeout),
                 node_id,
                 alt_addrs: vec![]
             });
@@ -115,7 +117,7 @@ impl PeerList {
     #[inline]
     fn refresh(&mut self, addr: &SocketAddr) {
         if let Some(ref mut data) = self.peers.get_mut(addr) {
-            data.timeout = now()+Time::from(self.timeout);
+            data.timeout = TS::now()+Time::from(self.timeout);
         }
     }
 
@@ -180,8 +182,9 @@ impl PeerList {
     #[inline]
     fn write_out<W: Write>(&self, out: &mut W) -> Result<(), io::Error> {
         try!(writeln!(out, "Peers:"));
+        let now = TS::now();
         for (addr, data) in &self.peers {
-            try!(writeln!(out, " - {} (ttl: {} s)", addr, data.timeout-now()));
+            try!(writeln!(out, " - {} (ttl: {} s)", addr, data.timeout-now));
         }
         Ok(())
     }
@@ -198,11 +201,11 @@ pub struct ReconnectEntry {
 }
 
 
-pub struct GenericCloud<P: Protocol, T: Table, S: Socket> {
+pub struct GenericCloud<D: Device, P: Protocol, T: Table, S: Socket, TS: TimeSource> {
     config: Config,
     magic: HeaderMagic,
     node_id: NodeId,
-    peers: PeerList,
+    peers: PeerList<TS>,
     addresses: Vec<Range>,
     learning: bool,
     broadcast: bool,
@@ -211,7 +214,7 @@ pub struct GenericCloud<P: Protocol, T: Table, S: Socket> {
     table: T,
     socket4: S,
     socket6: S,
-    device: Device,
+    device: D,
     crypto: Crypto,
     next_peerlist: Time,
     update_freq: Duration,
@@ -221,12 +224,13 @@ pub struct GenericCloud<P: Protocol, T: Table, S: Socket> {
     next_beacon: Time,
     port_forwarding: Option<PortForwarding>,
     traffic: TrafficStats,
-    beacon_serializer: BeaconSerializer,
+    beacon_serializer: BeaconSerializer<TS>,
     _dummy_p: PhantomData<P>,
+    _dummy_ts: PhantomData<TS>
 }
 
-impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
+impl<D: Device, P: Protocol, T: Table, S: Socket, TS: TimeSource> GenericCloud<D, P, T, S, TS> {
-    pub fn new(config: &Config, device: Device, table: T,
+    pub fn new(config: &Config, device: D, table: T,
         learning: bool, broadcast: bool, addresses: Vec<Range>,
         crypto: Crypto, port_forwarding: Option<PortForwarding>
     ) -> Self {
@@ -238,6 +242,7 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
             Ok(socket) => socket,
             Err(err) => fail!("Failed to open ipv6 address ::{}: {}", config.port, err)
         };
+        let now = TS::now();
         let mut res = GenericCloud{
             magic: config.get_magic(),
             node_id: random(),
@@ -251,18 +256,19 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
             socket4,
             socket6,
             device,
-            next_peerlist: now(),
+            next_peerlist: now,
             update_freq: config.get_keepalive(),
             buffer_out: [0; 64*1024],
-            next_housekeep: now(),
+            next_housekeep: now,
-            next_stats_out: now() + STATS_INTERVAL,
+            next_stats_out: now + STATS_INTERVAL,
-            next_beacon: now(),
+            next_beacon: now,
             port_forwarding,
             traffic: TrafficStats::default(),
             beacon_serializer: BeaconSerializer::new(&config.get_magic(), crypto.get_key()),
             crypto,
             config: config.clone(),
             _dummy_p: PhantomData,
+            _dummy_ts: PhantomData
         };
         res.initialize();
         return res
@@ -344,13 +350,14 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
     /// This method adds a peer to the list of nodes to reconnect to. A periodic task will try to
     /// connect to the peer if it is not already connected.
     pub fn add_reconnect_peer(&mut self, add: String) {
+        let now = TS::now();
         self.reconnect_peers.push(ReconnectEntry {
             address: add,
             tries: 0,
             timeout: 1,
             resolved: vec![],
-            next_resolve: now(),
+            next_resolve: now,
-            next: now()
+            next: now
         })
     }
 
@@ -431,7 +438,7 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
             pfw.check_extend();
         }
         // Periodically send peer list to peers
-        let now = now();
+        let now = TS::now();
         if self.next_peerlist <= now {
             debug!("Send peer list to all peers");
             let mut peer_num = self.peers.len();
@@ -705,6 +712,10 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
         }
     }
 
+    fn decode_message<'a>(&self, msg: &'a mut [u8]) -> Result<Message<'a>, Error> {
+        decode(msg, self.magic, &self.crypto)
+    }
+
     fn handle_socket_data(&mut self, src: SocketAddr, data: &mut [u8]) {
         let size = data.len();
         if let Err(e) = decode(data, self.magic, &mut self.crypto).and_then(|msg| {
@@ -760,7 +771,7 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
                 WaitResult::SocketV6 => self.handle_socket_v6_event(&mut buffer),
                 WaitResult::Device => self.handle_device_event(&mut buffer)
             }
-            if self.next_housekeep < now() {
+            if self.next_housekeep < TS::now() {
                 poll_error = false;
                 if ctrlc.was_pressed() {
                     break
@@ -768,10 +779,51 @@ impl<P: Protocol, T: Table, S: Socket> GenericCloud<P, T, S> {
                 if let Err(e) = self.housekeep() {
                     error!("Error: {}", e)
                 }
-                self.next_housekeep = now() + 1
+                self.next_housekeep = TS::now() + 1
             }
         }
         info!("Shutting down...");
         self.broadcast_msg(&mut Message::Close).ok();
     }
 }
+
+
+
+#[cfg(test)] use super::ethernet::{self, SwitchTable};
+#[cfg(test)] use super::util::MockTimeSource;
+#[cfg(test)] use super::net::MockSocket;
+#[cfg(test)] use super::device::MockDevice;
+
+#[cfg(test)]
+impl<P: Protocol, T: Table, TS: TimeSource> GenericCloud<MockDevice, P, T, MockSocket, TS> {
+    fn is_empty(&self) -> bool {
+        self.device.is_empty() && self.socket4.is_empty() && self.socket6.is_empty()
+    }
+}
+
+#[cfg(test)]
+type TestNode = GenericCloud<MockDevice, ethernet::Frame, SwitchTable<MockTimeSource>, MockSocket, MockTimeSource>;
+
+#[cfg(test)]
+fn create_node() -> TestNode {
+    TestNode::new(
+        &Config::default(),
+        MockDevice::new(),
+        SwitchTable::new(1800, 10),
+        true, true, vec![], Crypto::None, None
+    )
+}
+
+#[test]
+fn connect() {
+    let mut node = create_node();
+    assert!(node.is_empty());
+    node.connect("1.2.3.4:5678").unwrap();
+    assert!(node.device.is_empty());
+    assert!(node.socket6.is_empty());
+    let (addr, mut message) = node.socket4.pop_outbound().unwrap();
+    assert_eq!("1.2.3.4:5678".to_socket_addrs().unwrap().next().unwrap(), addr);
+    let message = node.decode_message(&mut message).unwrap();
+    assert_eq!(Message::Init(0, node.node_id, vec![]), message);
+
+}

src/device.rs

@@ -3,9 +3,10 @@
 // This software is licensed under GPL-3 or newer (see LICENSE.md)
 
 use std::os::unix::io::{AsRawFd, RawFd};
-use std::io::{self, Error as IoError, Read, Write};
+use std::io::{self, Error as IoError, ErrorKind, Read, Write};
 use std::fs;
 use std::fmt;
+use std::collections::VecDeque;
 
 use super::types::Error;
 
@@ -40,15 +41,48 @@ impl fmt::Display for Type {
 }
 
 
+pub trait Device: AsRawFd {
+    /// Returns the type of this device
+    fn get_type(&self) -> Type;
+
+    /// Returns the interface name of this device.
+    fn ifname(&self) -> &str;
+
+    /// Reads a packet/frame from the device
+    ///
+    /// This method reads one packet or frame (depending on the device type) into the `buffer`.
+    /// The `buffer` must be large enough to hold a packet/frame of maximum size, otherwise the
+    /// packet/frame will be split.
+    /// The method will block until a packet/frame is ready to be read.
+    /// On success, the method will return the starting position and the amount of bytes read into
+    /// the buffer.
+    ///
+    /// # Errors
+    /// This method will return an error if the underlying read call fails.
+    fn read(&mut self, buffer: &mut [u8]) -> Result<(usize, usize), Error>;
+
+    /// Writes a packet/frame to the device
+    ///
+    /// This method writes one packet or frame (depending on the device type) from `data` to the
+    /// device. The data starts at the position `start` in the buffer. The buffer should have at
+    /// least 4 bytes of space before the start of the packet.
+    /// The method will block until the packet/frame has been written.
+    ///
+    /// # Errors
+    /// This method will return an error if the underlying read call fails.
+    fn write(&mut self, data: &mut [u8], start: usize) -> Result<(), Error>;
+}
+
+
 /// Represents a tun/tap device
-pub struct Device {
+pub struct TunTapDevice {
     fd: fs::File,
     ifname: String,
     type_: Type,
 }
 
 
-impl Device {
+impl TunTapDevice {
     /// Creates a new tun/tap device
     ///
     /// This method creates a new device of the `type_` kind with the name `ifname`.
@@ -91,7 +125,7 @@ impl Device {
                 while ifname_c.last() == Some(&0) {
                     ifname_c.pop();
                 }
-                Ok(Device{fd, ifname: String::from_utf8(ifname_c).unwrap(), type_})
+                Ok(Self{fd, ifname: String::from_utf8(ifname_c).unwrap(), type_})
             },
             _ => Err(IoError::last_os_error())
         }
@@ -106,19 +140,6 @@ impl Device {
         }
     }
 
-    /// Returns the interface name of this device.
-    #[inline]
-    pub fn ifname(&self) -> &str {
-        &self.ifname
-    }
-
-    /// Returns the type of this device
-    #[allow(dead_code)]
-    #[inline]
-    pub fn get_type(&self) -> Type {
-        self.type_
-    }
-
     /// Creates a dummy device based on an existing file
     ///
     /// This method opens a regular or special file and reads from it to receive packets and
@@ -134,31 +155,13 @@ impl Device {
     /// This method will return an error if the file can not be opened for reading and writing.
     #[allow(dead_code)]
     pub fn dummy(ifname: &str, path: &str, type_: Type) -> io::Result<Self> {
-        Ok(Device{
+        Ok(TunTapDevice{
             fd: try!(fs::OpenOptions::new().create(true).read(true).write(true).open(path)),
             ifname: ifname.to_string(),
             type_
         })
     }
 
-    /// Reads a packet/frame from the device
-    ///
-    /// This method reads one packet or frame (depending on the device type) into the `buffer`.
-    /// The `buffer` must be large enough to hold a packet/frame of maximum size, otherwise the
-    /// packet/frame will be split.
-    /// The method will block until a packet/frame is ready to be read.
-    /// On success, the method will return the starting position and the amount of bytes read into
-    /// the buffer.
-    ///
-    /// # Errors
-    /// This method will return an error if the underlying read call fails.
-    #[inline]
-    pub fn read(&mut self, mut buffer: &mut [u8]) -> Result<(usize, usize), Error> {
-        let read = try!(self.fd.read(&mut buffer).map_err(|e| Error::TunTapDev("Read error", e)));
-        let (start, read) = self.correct_data_after_read(&mut buffer, 0, read);
-        Ok((start, read))
-    }
-
     #[cfg(any(target_os = "linux", target_os = "android"))]
     #[inline]
     fn correct_data_after_read(&mut self, _buffer: &mut [u8], start: usize, read: usize) -> (usize, usize) {
@@ -179,24 +182,6 @@ impl Device {
         }
     }
 
-    /// Writes a packet/frame to the device
-    ///
-    /// This method writes one packet or frame (depending on the device type) from `data` to the
-    /// device. The data starts at the position `start` in the buffer. The buffer should have at
-    /// least 4 bytes of space before the start of the packet.
-    /// The method will block until the packet/frame has been written.
-    ///
-    /// # Errors
-    /// This method will return an error if the underlying read call fails.
-    #[inline]
-    pub fn write(&mut self, mut data: &mut [u8], start: usize) -> Result<(), Error> {
-        let start = self.correct_data_before_write(&mut data, start);
-        match self.fd.write_all(&data[start..]) {
-            Ok(_) => self.fd.flush().map_err(|e| Error::TunTapDev("Flush error", e)),
-            Err(e) => Err(Error::TunTapDev("Write error", e))
-        }
-    }
-
     #[cfg(any(target_os = "linux", target_os = "android"))]
     #[inline]
     fn correct_data_before_write(&mut self, _buffer: &mut [u8], start: usize) -> usize {
@@ -223,9 +208,89 @@ impl Device {
     }
 }
 
-impl AsRawFd for Device {
+impl Device for TunTapDevice {
+    fn get_type(&self) -> Type {
+        self.type_
+    }
+
+    fn ifname(&self) -> &str {
+        &self.ifname
+    }
+
+    fn read(&mut self, mut buffer: &mut [u8]) -> Result<(usize, usize), Error> {
+        let read = try!(self.fd.read(&mut buffer).map_err(|e| Error::TunTapDev("Read error", e)));
+        let (start, read) = self.correct_data_after_read(&mut buffer, 0, read);
+        Ok((start, read))
+    }
+
+    fn write(&mut self, mut data: &mut [u8], start: usize) -> Result<(), Error> {
+        let start = self.correct_data_before_write(&mut data, start);
+        match self.fd.write_all(&data[start..]) {
+            Ok(_) => self.fd.flush().map_err(|e| Error::TunTapDev("Flush error", e)),
+            Err(e) => Err(Error::TunTapDev("Write error", e))
+        }
+    }
+
+}
+
+impl AsRawFd for TunTapDevice {
     #[inline]
     fn as_raw_fd(&self) -> RawFd {
         self.fd.as_raw_fd()
     }
 }
+
+
+pub struct MockDevice {
+    inbound: VecDeque<Vec<u8>>,
+    outbound: VecDeque<Vec<u8>>
+}
+
+impl MockDevice {
+    pub fn new() -> Self {
+        Self { outbound: VecDeque::new(), inbound: VecDeque::new() }
+    }
+
+    pub fn put_inbound(&mut self, data: Vec<u8>) {
+        self.inbound.push_back(data)
+    }
+
+    pub fn pop_outbound(&mut self) -> Option<Vec<u8>> {
+        self.outbound.pop_front()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.inbound.is_empty() && self.outbound.is_empty()
+    }
+}
+
+impl Device for MockDevice {
+    fn get_type(&self) -> Type {
+        Type::Dummy
+    }
+
+    fn ifname(&self) -> &str {
+        unimplemented!()
+    }
+
+    fn read(&mut self, buffer: &mut [u8]) -> Result<(usize, usize), Error> {
+        if let Some(data) = self.inbound.pop_front() {
+            buffer[0..data.len()].copy_from_slice(&data);
+            Ok((0, data.len()))
+        } else {
+            Err(Error::TunTapDev("empty", io::Error::from(ErrorKind::UnexpectedEof)))
+        }
+    }
+
+    fn write(&mut self, data: &mut [u8], start: usize) -> Result<(), Error> {
+        self.outbound.push_back(data[start..].to_owned());
+        Ok(())
+    }
+}
+
+impl AsRawFd for MockDevice {
+    #[inline]
+    fn as_raw_fd(&self) -> RawFd {
+        unimplemented!()
+    }
+}

src/ethernet.rs

@@ -7,11 +7,12 @@ use std::collections::HashMap;
 use std::collections::hash_map::Entry;
 use std::hash::BuildHasherDefault;
 use std::io::{self, Write};
+use std::marker::PhantomData;
 
 use fnv::FnvHasher;
 
 use super::types::{Error, Table, Protocol, Address};
-use super::util::{now, Time, Duration};
+use super::util::{TimeSource, Time, Duration, MockTimeSource};
 
 /// An ethernet frame dissector
 ///
@@ -68,26 +69,27 @@ type Hash = BuildHasherDefault<FnvHasher>;
 ///
 /// This table is a simple hash map between an address and the destination peer. It learns
 /// addresses as they are seen and forgets them after some time.
-pub struct SwitchTable {
+pub struct SwitchTable<TS> {
     /// The table storing the actual mapping
     table: HashMap<Address, SwitchTableValue, Hash>,
     /// Timeout period for forgetting learnt addresses
     timeout: Duration,
     // Timeout period for not overwriting learnt addresses
     protection_period: Duration,
+    _dummy_ts: PhantomData<TS>
 }
 
-impl SwitchTable {
+impl<TS: TimeSource> SwitchTable<TS> {
     /// Creates a new switch table
     pub fn new(timeout: Duration, protection_period: Duration) -> Self {
-        SwitchTable{table: HashMap::default(), timeout, protection_period}
+        Self{table: HashMap::default(), timeout, protection_period, _dummy_ts: PhantomData}
     }
 }
 
-impl Table for SwitchTable {
+impl<TS: TimeSource> Table for SwitchTable<TS> {
     /// Forget addresses that have not been seen for the configured timeout
     fn housekeep(&mut self) {
-        let now = now();
+        let now = TS::now();
         let mut del: Vec<Address> = Vec::new();
         for (key, val) in &self.table {
             if val.timeout < now {
@@ -102,7 +104,7 @@ impl Table for SwitchTable {
 
     /// Write out the table
     fn write_out<W: Write>(&self, out: &mut W) -> Result<(), io::Error> {
-        let now = now();
+        let now = TS::now();
         try!(writeln!(out, "Switch table:"));
         for (addr, val) in &self.table {
             try!(writeln!(out, " - {} => {} (ttl: {} s)", addr, val.address, val.timeout - now));
@@ -113,7 +115,7 @@ impl Table for SwitchTable {
     /// Learns the given address, inserting it in the hash map
     #[inline]
     fn learn(&mut self, key: Address, _prefix_len: Option<u8>, addr: SocketAddr) {
-        let deadline = now() + Time::from(self.timeout);
+        let deadline = TS::now() + Time::from(self.timeout);
         match self.table.entry(key) {
             Entry::Vacant(entry) => {
                 entry.insert(SwitchTableValue{address: addr, timeout: deadline});
@@ -163,7 +165,6 @@ impl Table for SwitchTable {
 
 #[cfg(test)] use std::str::FromStr;
 #[cfg(test)] use std::net::ToSocketAddrs;
-#[cfg(test)] use std::thread;
 
 #[test]
 fn decode_frame_without_vlan() {
@@ -192,17 +193,19 @@ fn decode_invalid_frame() {
 
 #[test]
 fn switch() {
-    let mut table = SwitchTable::new(10, 1);
+    MockTimeSource::set_time(1000);
+    let mut table = SwitchTable::<MockTimeSource>::new(10, 1);
     let addr = Address::from_str("12:34:56:78:90:ab").unwrap();
     let peer = "1.2.3.4:5678".to_socket_addrs().unwrap().next().unwrap();
     let peer2 = "1.2.3.5:7890".to_socket_addrs().unwrap().next().unwrap();
     assert!(table.lookup(&addr).is_none());
+    MockTimeSource::set_time(1000);
     table.learn(addr.clone(), None, peer.clone());
     assert_eq!(table.lookup(&addr), Some(peer));
-    // Do not override within 1 seconds
+    MockTimeSource::set_time(1000);
     table.learn(addr.clone(), None, peer2.clone());
     assert_eq!(table.lookup(&addr), Some(peer));
-    thread::sleep(std::time::Duration::from_secs(1));
+    MockTimeSource::set_time(1010);
     table.learn(addr.clone(), None, peer2.clone());
     assert_eq!(table.lookup(&addr), Some(peer2));
 }

src/main.rs

@@ -48,17 +48,17 @@ use std::process::Command;
 use std::fs::File;
 use std::path::Path;
 use std::io::{self, Write};
+use std::net::UdpSocket;
 
-use device::{Device, Type};
+use device::{TunTapDevice, Device, Type};
 use ethernet::SwitchTable;
 use ip::RoutingTable;
 use types::{Mode, Range, Protocol, HeaderMagic, Error};
 use cloud::GenericCloud;
 use crypto::{Crypto, CryptoMethod};
 use port_forwarding::PortForwarding;
-use util::Duration;
+use util::{Duration, SystemTimeSource};
 use config::Config;
-use std::net::UdpSocket;
 
 
 const VERSION: u8 = 1;
@@ -156,25 +156,25 @@ fn run_script(script: &str, ifname: &str) {
 }
 
 enum AnyTable {
-    Switch(SwitchTable),
+    Switch(SwitchTable<SystemTimeSource>),
     Routing(RoutingTable)
 }
 
 enum AnyCloud<P: Protocol> {
-    Switch(GenericCloud<P, SwitchTable, UdpSocket>),
+    Switch(GenericCloud<TunTapDevice, P, SwitchTable<SystemTimeSource>, UdpSocket, SystemTimeSource>),
-    Routing(GenericCloud<P, RoutingTable, UdpSocket>)
+    Routing(GenericCloud<TunTapDevice, P, RoutingTable, UdpSocket, SystemTimeSource>)
 }
 
 impl<P: Protocol> AnyCloud<P> {
-    #[allow(unknown_lints,clippy::too_many_arguments)]
+    #[allow(unknown_lints, clippy::too_many_arguments)]
-    fn new(config: &Config, device: Device, table: AnyTable,
+    fn new(config: &Config, device: TunTapDevice, table: AnyTable,
             learning: bool, broadcast: bool, addresses: Vec<Range>,
             crypto: Crypto, port_forwarding: Option<PortForwarding>) -> Self {
         match table {
-            AnyTable::Switch(t) => AnyCloud::Switch(GenericCloud::<P, SwitchTable, UdpSocket>::new(
+            AnyTable::Switch(t) => AnyCloud::Switch(GenericCloud::<TunTapDevice, P, SwitchTable<SystemTimeSource>, UdpSocket, SystemTimeSource>::new(
                 config, device,t, learning, broadcast, addresses, crypto, port_forwarding
             )),
-            AnyTable::Routing(t) => AnyCloud::Routing(GenericCloud::<P, RoutingTable, UdpSocket>::new(
+            AnyTable::Routing(t) => AnyCloud::Routing(GenericCloud::<TunTapDevice, P, RoutingTable, UdpSocket, SystemTimeSource>::new(
                 config, device,t, learning, broadcast, addresses, crypto, port_forwarding
             ))
         }
@@ -211,7 +211,7 @@ impl<P: Protocol> AnyCloud<P> {
 
 
 fn run<P: Protocol> (config: Config) {
-    let device = try_fail!(Device::new(&config.device_name, config.device_type, config.device_path.as_ref().map(|s| s as &str)),
+    let device = try_fail!(TunTapDevice::new(&config.device_name, config.device_type, config.device_path.as_ref().map(|s| s as &str)),
         "Failed to open virtual {} interface {}: {}", config.device_type, config.device_name);
     info!("Opened device {}", device.ifname());
     let mut ranges = Vec::with_capacity(config.subnets.len());

src/net.rs

@@ -54,6 +54,10 @@ impl MockSocket {
     pub fn pop_outbound(&mut self) -> Option<(SocketAddr, Vec<u8>)> {
         self.outbound.pop_front()
     }
+
+    pub fn is_empty(&self) -> bool {
+        self.inbound.is_empty() && self.outbound.is_empty()
+    }
 }
 
 impl AsRawFd for MockSocket {

src/poll/epoll.rs

@@ -7,7 +7,6 @@ use libc;
 use std::os::unix::io::RawFd;
 use std::io;
 use device::Device;
-use std::os::unix::io::AsRawFd;
 
 use super::WaitResult;
 use ::device::Type;

src/port_forwarding.rs

@@ -7,7 +7,7 @@ use std::io;
 
 use igd::*;
 
-use super::util::{Time, now};
+use super::util::{SystemTimeSource, Time, TimeSource};
 
 const LEASE_TIME: u32 = 300;
 const DESCRIPTION: &str = "VpnCloud";
@@ -90,7 +90,7 @@ impl PortForwarding {
         };
         info!("Port-forwarding: sucessfully activated port forward on {}, timeout: {}", external_addr, timeout);
         let next_extension = if timeout > 0 {
-            Some(now() + Time::from(timeout) - 60)
+            Some(SystemTimeSource::now() + Time::from(timeout) - 60)
         } else {
             None
         };
@@ -104,7 +104,7 @@ impl PortForwarding {
 
     pub fn check_extend(&mut self) {
         if let Some(deadline) = self.next_extension {
-            if deadline > now() {
+            if deadline > SystemTimeSource::now() {
                 return
             }
         } else {
@@ -114,7 +114,7 @@ impl PortForwarding {
             Ok(()) => debug!("Port-forwarding: extended port forwarding"),
             Err(err) => error!("Port-forwarding: failed to extend port forwarding: {}", err)
         };
-        self.next_extension = Some(now() + Time::from(LEASE_TIME) - 60);
+        self.next_extension = Some(SystemTimeSource::now() + Time::from(LEASE_TIME) - 60);
     }
 
     fn deactivate(&self) {

src/udpmessage.rs

@@ -78,7 +78,7 @@ impl<'a> fmt::Debug for Message<'a> {
 }
 
 #[allow(unknown_lints,clippy::needless_range_loop)]
-pub fn decode<'a>(data: &'a mut [u8], magic: HeaderMagic, crypto: &mut Crypto) -> Result<Message<'a>, Error> {
+pub fn decode<'a>(data: &'a mut [u8], magic: HeaderMagic, crypto: &Crypto) -> Result<Message<'a>, Error> {
     let mut end = data.len();
     let (header, mut pos) = try!(TopHeader::read_from(&data[..end]));
     if header.magic != magic {

src/util.rs

@@ -4,6 +4,7 @@
 
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::fmt;
+use std::sync::atomic::{AtomicIsize, Ordering};
 
 use super::types::Error;
 
@@ -20,19 +21,6 @@ use std::time::Instant;
 pub type Duration = u32;
 pub type Time = i64;
 
-#[inline]
-#[cfg(target_os = "linux")]
-pub fn now() -> Time {
-    let mut tv = libc::timespec { tv_sec: 0, tv_nsec: 0 };
-    unsafe { libc::clock_gettime(6, &mut tv); }
-    tv.tv_sec as Time
-}
-
-#[inline]
-#[cfg(not(target_os = "linux"))]
-pub fn now() -> Time {
-    time::get_time().sec
-}
 
 const HEX_CHARS: &[u8] = b"0123456789abcdef";
 
@@ -189,3 +177,44 @@ impl CtrlC {
         self.trap.wait(self.dummy_time).is_some()
     }
 }
+
+
+pub trait TimeSource: Sync + Copy + Send + 'static {
+    fn now() -> Time;
+}
+
+#[derive(Clone, Copy)]
+pub struct SystemTimeSource;
+
+impl TimeSource for SystemTimeSource {
+    #[cfg(target_os = "linux")]
+    fn now() -> Time {
+        let mut tv = libc::timespec { tv_sec: 0, tv_nsec: 0 };
+        unsafe { libc::clock_gettime(6, &mut tv); }
+        tv.tv_sec as Time
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    fn now() -> Time {
+        time::get_time().sec
+    }
+}
+
+thread_local! {
+    static MOCK_TIME: AtomicIsize = AtomicIsize::new(0);
+}
+
+#[derive(Clone, Copy)]
+pub struct MockTimeSource;
+
+impl MockTimeSource {
+    pub fn set_time(time: Time) {
+        MOCK_TIME.with(|t| t.store(time as isize, Ordering::SeqCst))
+    }
+}
+
+impl TimeSource for MockTimeSource {
+    fn now() -> Time {
+        MOCK_TIME.with(|t| t.load(Ordering::SeqCst) as Time)
+    }
+}