src/allmydata/test/test_upload.py

   1
   2 from twisted.trial import unittest
   3 from twisted.python.failure import Failure
   4 from cStringIO import StringIO
   5
   6 from allmydata import upload
   7 from allmydata.uri import unpack_uri
   8
   9 from test_encode import FakePeer
  10
  11 class FakeClient:
  12     def __init__(self, mode="good"):
  13         self.mode = mode
  14     def get_permuted_peers(self, verifierid):
  15         return [ ("%20d"%fakeid, "%20d"%fakeid, FakePeer(self.mode),) for fakeid in range(50) ]
  16
  17 class GoodServer(unittest.TestCase):
  18     def setUp(self):
  19         self.node = FakeClient(mode="good")
  20         self.u = upload.Uploader()
  21         self.u.running = True
  22         self.u.parent = self.node
  23
  24     def _check(self, uri):
  25         self.failUnless(isinstance(uri, str))
  26         self.failUnless(uri.startswith("URI:"))
  27         d = unpack_uri(uri)
  28         self.failUnless(isinstance(d['verifierid'], str))
  29         self.failUnlessEqual(len(d['verifierid']), 20)
  30         self.failUnless(isinstance(d['fileid'], str))
  31         self.failUnlessEqual(len(d['fileid']), 20)
  32         self.failUnless(isinstance(d['key'], str))
  33         self.failUnlessEqual(len(d['key']), 16)
  34         self.failUnless(isinstance(d['codec_params'], str))
  35
  36     def testData(self):
  37         data = "This is some data to upload"
  38         d = self.u.upload_data(data)
  39         d.addCallback(self._check)
  40         return d
  41     testData.timeout = 300
  42
  43     def testFileHandle(self):
  44         data = "This is some data to upload"
  45         d = self.u.upload_filehandle(StringIO(data))
  46         d.addCallback(self._check)
  47         return d
  48     testFileHandle.timeout = 300
  49
  50     def testFilename(self):
  51         fn = "Uploader-testFilename.data"
  52         f = open(fn, "wb")
  53         data = "This is some data to upload"
  54         f.write(data)
  55         f.close()
  56         d = self.u.upload_filename(fn)
  57         d.addCallback(self._check)
  58         return d
  59     testFilename.test = 300
  60
  61 class FullServer(unittest.TestCase):
  62     def setUp(self):
  63         self.node = FakeClient(mode="full")
  64         self.u = upload.Uploader()
  65         self.u.running = True
  66         self.u.parent = self.node
  67
  68     def _should_fail(self, f):
  69         self.failUnless(isinstance(f, Failure) and f.check(upload.NotEnoughPeersError))
  70
  71     def testData(self):
  72         data = "This is some data to upload"
  73         d = self.u.upload_data(data)
  74         d.addBoth(self._should_fail)
  75         return d
  76
  77
  78 # TODO:
  79 #  upload with exactly 75 peers (shares_of_happiness)
  80 #  have a download fail
  81 #  cancel a download (need to implement more cancel stuff)