2 from twisted.trial import unittest
3 from twisted.application import service
5 from foolscap import Tub, eventual
6 from foolscap.logging import log
8 from allmydata import upload, offloaded
9 from allmydata.util import hashutil
11 class CHKUploadHelper_fake(offloaded.CHKUploadHelper):
12 def start_encrypted(self, eu):
13 needed_shares, happy, total_shares = self._encoding_parameters
16 return (hashutil.uri_extension_hash(""),
17 needed_shares, total_shares, size)
18 d.addCallback(_got_size)
21 class CHKUploadHelper_already_uploaded(offloaded.CHKUploadHelper):
23 res = {'uri_extension_hash': hashutil.uri_extension_hash("")}
26 class FakeClient(service.MultiService):
27 def log(self, *args, **kwargs):
28 return log.msg(*args, **kwargs)
29 def get_push_to_ourselves(self):
31 def get_encoding_parameters(self):
34 def flush_but_dont_ignore(res):
35 d = eventual.flushEventualQueue()
41 class AssistedUpload(unittest.TestCase):
47 t.setServiceParent(self.s)
49 # we never actually use this for network traffic, so it can use a
51 t.setLocation("bogus:1234")
53 self.helper = h = offloaded.Helper(".")
54 h.chk_upload_helper_class = CHKUploadHelper_fake
55 h.setServiceParent(self.s)
56 self.helper_furl = t.registerReference(h)
59 d = self.s.stopService()
60 d.addCallback(eventual.fireEventually)
61 d.addBoth(flush_but_dont_ignore)
66 u = upload.Uploader(self.helper_furl)
67 u.setServiceParent(self.s)
70 d = eventual.fireEventually()
71 d.addCallback(eventual.fireEventually)
72 d.addCallback(eventual.fireEventually)
77 DATA = "I need help\n" * 1000
78 return u.upload_data(DATA)
82 d.addCallback(_uploaded)
87 def test_already_uploaded(self):
88 self.helper.chk_upload_helper_class = CHKUploadHelper_already_uploaded
89 u = upload.Uploader(self.helper_furl)
90 u.setServiceParent(self.s)
93 d = eventual.fireEventually()
94 d.addCallback(eventual.fireEventually)
95 d.addCallback(eventual.fireEventually)
100 DATA = "I need help\n" * 1000
101 return u.upload_data(DATA)
102 d.addCallback(_ready)
105 d.addCallback(_uploaded)