Implemented live-edge walker

Tribler/Test/Community/Multichain/test_multichain_community.py

@@ -1,9 +1,12 @@
 """
 This file contains the tests for the community.py for MultiChain community.
 """
+import time
+
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
 from twisted.internet.task import deferLater
+from twisted.internet.threads import blockingCallFromThread
 
 from Tribler.Core.Session import Session
 
@@ -403,6 +406,32 @@ def test_crawl_block_specified_sequence_number(self):
         self.assertBlocksInDatabase(crawler, 2)
         self.assertBlocksAreEqual(node, crawler)
 
+    def test_crawl_blocks_negative_sequence_number(self):
+        """
+        Test the crawler to fetch blocks starting from a negative sequence number.
+        """
+        # Arrange
+        node, other, crawler = self.create_nodes(3)
+
+        # Act
+        TestMultiChainCommunity.create_block(node, other, self._create_target(node, other), 10, 5) # sq 1
+        TestMultiChainCommunity.create_block(node, other, self._create_target(node, other), 20, 30) # sq 2
+        TestMultiChainCommunity.create_block(node, other, self._create_target(node, other), 40, 50) # sq 3
+
+        self.clean_database(crawler)
+        self.assertBlocksInDatabase(crawler, 0)
+        TestMultiChainCommunity.crawl_node(crawler, node, self._create_target(crawler, node), -2)
+
+        # Assert
+        self.assertBlocksInDatabase(node, 6)
+        self.assertBlocksInDatabase(crawler, 4)
+        self.assertIsNone(self.get_node_sq_from_db(crawler, node, 1))
+        self.assertIsNone(self.get_node_sq_from_db(crawler, other, 1))
+        self.assertIsNotNone(self.get_node_sq_from_db(crawler, node, 2))
+        self.assertIsNotNone(self.get_node_sq_from_db(crawler, other, 2))
+        self.assertIsNotNone(self.get_node_sq_from_db(crawler, node, 3))
+        self.assertIsNotNone(self.get_node_sq_from_db(crawler, other, 3))
+
     def test_crawl_no_block(self):
         """
         Test crawl without a block.
@@ -487,9 +516,10 @@ def test_crawler_on_introduction_received(self):
         # and we don't actually want to send the crawl request since the counter party is fake, just count if it is run
         counter = [0]
 
-        def replacement(cand, pk):
-            counter[0] += 1
-        crawler._community.send_crawl_request = replacement
+        def cb_replacement(cand, pk, sq=None):
+            if sq != -1: # Ignore live edge last-block request
+                counter[0] += 1
+        crawler.community.send_crawl_request = cb_replacement
 
         # Act
         crawler.call(crawler.community.on_introduction_response, [intro_response])
@@ -532,6 +562,155 @@ def test_get_statistics_for_not_self(self):
         assert isinstance(statistics, dict), type(statistics)
         assert len(statistics) > 0
 
+    def test_get_trust(self):
+        """
+        Test that the trust nodes have for each other is the upload + the download total of all blocks.
+        """
+        # Arrange
+        node, other = self.create_nodes(2)
+        TestMultiChainCommunity.create_block(node, other, self._create_target(node, other), 10, 5)
+
+        # Get statistics
+        node_trust = blockingCallFromThread(reactor, node.community.get_trust, other.community.my_member)
+        other_trust = blockingCallFromThread(reactor, other.community.get_trust, node.community.my_member)
+        self.assertEqual(node_trust, 15)
+        self.assertEqual(other_trust, 15)
+
+    def test_get_default_trust(self):
+        """
+        Test that the trust between nodes without blocks is 1.
+        """
+        # Arrange
+        node, other = self.create_nodes(2)
+
+        # Get statistics
+        node_trust = blockingCallFromThread(reactor, node.community.get_trust, other.community.my_member)
+        other_trust = blockingCallFromThread(reactor, other.community.get_trust, node.community.my_member)
+        self.assertEqual(node_trust, 1)
+        self.assertEqual(other_trust, 1)
+
+    def test_live_edge_bootstrapping(self):
+        """
+        A node without trust for anyone should still find a candidate.
+        """
+        # Arrange
+        node, other = self.create_nodes(2)
+        candidate = node.community.create_or_update_walkcandidate(other.my_candidate.sock_addr,
+                                                                  other.my_candidate.sock_addr,
+                                                                  ('0.0.0.0', 0),
+                                                                  other.my_candidate.tunnel,
+                                                                  u"unknown")
+        candidate.associate(other.community.my_member)
+        candidate.walk_response(time.time())
+
+        # Assert
+        intro = blockingCallFromThread(reactor, node.community.dispersy_get_introduce_candidate,
+                                       node.my_candidate)
+        self.assertIsNotNone(intro)
+        self.assertIsInstance(intro, Candidate)
+        self.assertEqual(intro, candidate)
+
+    def test_live_edge_recommend_valid(self):
+        """
+        Live edges should never include invalid/old candidates.
+        """
+        # Arrange
+        node, other, another = self.create_nodes(3)
+
+        # Stop the community from walking/crawling once it gets reactor control
+        node.community.cancel_all_pending_tasks()
+        node.community.reset_live_edges()
+        node.community.candidates.clear()
+
+        candidate = node.community.create_or_update_walkcandidate(other.my_candidate.sock_addr,
+                                                                  other.my_candidate.sock_addr,
+                                                                  ('0.0.0.0', 0),
+                                                                  other.my_candidate.tunnel,
+                                                                  u"unknown")
+        candidate.associate(other.community.my_member)
+        candidate.walk_response(time.time())
+
+        node.community.create_or_update_walkcandidate(another.my_candidate.sock_addr,
+                                                      another.my_candidate.sock_addr,
+                                                      ('0.0.0.0', 0),
+                                                      another.my_candidate.tunnel,
+                                                      u"unknown")
+
+        # Assert
+        intro = blockingCallFromThread(reactor, node.community.dispersy_get_introduce_candidate,
+                                       node.my_candidate)
+        self.assertIsNotNone(intro)
+        self.assertIsInstance(intro, Candidate)
+        self.assertEqual(intro, candidate)
+
+    def test_live_edge_callback_no_candidates(self):
+        """
+        Test live edges start with my member.
+        """
+        # Arrange
+        node, = self.create_nodes(1)
+
+        called = [False]
+
+        def check_live_edge(edge_id, candidates):
+            self.assertEqual(1, edge_id)
+            self.assertEqual(node.my_member.mid, candidates[0].get_member().mid)
+            called[0] = True
+
+        node.community.set_live_edge_callback(check_live_edge)
+
+        # Stop the community from walking/crawling once it gets reactor control
+        node.community.cancel_all_pending_tasks()
+        node.community.reset_live_edges()
+
+        # Act
+        node.community.take_step()
+
+        # Assert
+        self.assertTrue(called)
+
+    def test_live_edge_callback(self):
+        """
+        Test creation and handling of a new live edge.
+        """
+        # Arrange
+        node, other = self.create_nodes(2)
+
+        # Create a cache, so our introduction response is expected by the node
+        cache = object.__new__(IntroductionRequestCache)
+        blockingCallFromThread(reactor, IntroductionRequestCache.__init__, cache,
+                               node.community, other.my_candidate.sock_addr)
+        cache = blockingCallFromThread(reactor, node.community.request_cache.add, cache)
+
+        # Create the actual response message
+        response = other.create_introduction_response(node.my_candidate,
+                                                      other.my_candidate.sock_addr,
+                                                      other.my_candidate.sock_addr,
+                                                      ("0.0.0.0", 0),
+                                                      ("0.0.0.0", 0),
+                                                      u"unknown",
+                                                      False,
+                                                      cache.number)
+        response._candidate = other.my_candidate # Fake its arrival from other
+
+        called = [False]
+
+        def check_live_edge(edge_id, candidates):
+            # We have no more valid candidates, increment id
+            self.assertEqual(1, edge_id)
+            # Start with our member
+            self.assertEqual(node.my_member.mid, candidates[0].get_member().mid)
+            # End with new member
+            self.assertEqual(other.my_member.mid, candidates[1].get_member().mid)
+            called[0] = True
+        node.community.set_live_edge_callback(check_live_edge)
+
+        # Act
+        blockingCallFromThread(reactor, node.community.on_introduction_response, [response])
+
+        # Assert
+        self.assertTrue(called[0])
+
     @blocking_call_on_reactor_thread
     def assertBlocksInDatabase(self, node, amount):
         count = node.community.persistence.execute(u"SELECT COUNT(*) FROM multi_chain").fetchone()[0]
@@ -546,6 +725,10 @@ def assertBlocksAreEqual(self, node, other):
             node.community.persistence.crawl(other.community.my_member.public_key, 0),
             other.community.persistence.crawl(other.community.my_member.public_key, 0))
 
+    @blocking_call_on_reactor_thread
+    def get_node_sq_from_db(self, node, sq_owner_node, sequence_number):
+        return node.community.persistence.get(sq_owner_node.community.my_member.public_key, sequence_number)
+
     @staticmethod
     def set_expectation(node, req, up, down):
         if node.community.pending_bytes.get(req.community.my_member.public_key):