p2pool/test/test_p2p.py

   1 import random
   2
   3 from twisted.internet import defer, reactor
   4 from twisted.trial import unittest
   5
   6 from p2pool import data, networks, p2p
   7 from p2pool.util import deferral
   8
   9 class MyNode(p2p.Node):
  10     def handle_share_hashes(self, hashes, peer):
  11         peer.send_sharereq(id=random.randrange(2**256), hashes=[hashes[0]], parents=5, stops=[])
  12         print 'handle_share_hashes', (hashes, peer)
  13
  14     def handle_share_reply(self, id, result, shares, peer):
  15         print (id, result, shares)
  16
  17 class Test(unittest.TestCase):
  18     @defer.inlineCallbacks
  19     def test_sharereq(self):
  20         n = MyNode(lambda: None, 29333, networks.nets['bitcoin'], {}, set([('127.0.0.1', 9333)]), 0, 0, 0, 0)
  21         n.start()
  22         try:
  23             yield deferral.sleep(10)
  24         finally:
  25             n.stop()