src/allmydata/test/test_encode.py

   1 #! /usr/bin/env python
   2
   3 from twisted.trial import unittest
   4 from twisted.internet import defer
   5 from allmydata import encode_new
   6 from cStringIO import StringIO
   7
   8 class MyEncoder(encode_new.Encoder):
   9     def send(self, share_num, methname, *args, **kwargs):
  10         if False and share_num < 10:
  11             print "send[%d].%s()" % (share_num, methname)
  12             if methname == "put_share_hashes":
  13                 print " ", [i for i,h in args[0]]
  14         return defer.succeed(None)
  15
  16 class Encode(unittest.TestCase):
  17     def test_1(self):
  18         e = MyEncoder()
  19         data = StringIO("some data to encode\n")
  20         e.setup(data)
  21         d = e.start()
  22         return d
  23