p2pool/test/test_p2p.py

   1 import random
   2
   3 from twisted.internet import defer
   4 from twisted.trial import unittest
   5
   6 from p2pool import networks, p2p
   7
   8 class MyNode(p2p.Node):
   9     def __init__(self, df):
  10         p2p.Node.__init__(self, lambda: None, 29333, networks.nets['bitcoin'], {}, set([('127.0.0.1', 9333)]), 0, 0, 0, 0)
  11
  12         self.id_to_use = random.randrange(2**256)
  13         self.df = df
  14
  15     def handle_share_hashes(self, hashes, peer):
  16         peer.send_sharereq(
  17             id=self.id_to_use,
  18             hashes=[hashes[0]],
  19             parents=5,
  20             stops=[],
  21         )
  22
  23     def handle_share_reply(self, id, result, shares, peer):
  24         if id == self.id_to_use:
  25             self.df.callback(None)
  26
  27 class Test(unittest.TestCase):
  28     @defer.inlineCallbacks
  29     def test_sharereq(self):
  30         df = defer.Deferred()
  31         n = MyNode(df)
  32         n.start()
  33         try:
  34             yield df
  35         finally:
  36             yield n.stop()