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
   9 class Test(unittest.TestCase):
  10     @defer.inlineCallbacks
  11     def test_sharereq(self):
  12         class MyNode(p2p.Node):
  13             def __init__(self, df):
  14                 p2p.Node.__init__(self, lambda: None, 29333, networks.nets['bitcoin'], {}, set([('127.0.0.1', 9333)]), 0, 0, 0, 0)
  15
  16                 self.df = df
  17
  18             def handle_share_hashes(self, hashes, peer):
  19                 peer.get_shares(
  20                     hashes=[hashes[0]],
  21                     parents=5,
  22                     stops=[],
  23                 ).chainDeferred(self.df)
  24
  25         df = defer.Deferred()
  26         n = MyNode(df)
  27         n.start()
  28         try:
  29             yield df
  30         finally:
  31             yield n.stop()