4 from twisted.trial import unittest
5 from twisted.python import runtime
6 from twisted.internet import defer
8 from allmydata.interfaces import IDirectoryNode, NoSuchChildError
10 from allmydata.util import fake_inotify, fileutil
11 from allmydata.util.encodingutil import get_filesystem_encoding, to_filepath
12 from allmydata.util.consumer import download_to_data
13 from allmydata.test.no_network import GridTestMixin
14 from allmydata.test.common_util import ReallyEqualMixin, NonASCIIPathMixin
15 from allmydata.test.common import ShouldFailMixin
17 from allmydata.frontends.magic_folder import MagicFolder
18 from allmydata.util.fileutil import abspath_expanduser_unicode
21 class MagicFolderTestMixin(GridTestMixin, ShouldFailMixin, ReallyEqualMixin, NonASCIIPathMixin):
23 These tests will be run both with a mock notifier, and (on platforms that support it)
24 with the real INotify.
28 GridTestMixin.setUp(self)
30 self.basedir = abspath_expanduser_unicode(temp.decode(get_filesystem_encoding()))
31 def _get_count(self, name):
32 return self.stats_provider.get_stats()["counters"].get(name, 0)
37 self.local_dir = os.path.join(self.basedir, self.unicode_or_fallback(u"loc\u0101l_dir", u"local_dir"))
38 self.mkdir_nonascii(self.local_dir)
40 self.client = self.g.clients[0]
41 self.stats_provider = self.client.stats_provider
43 d = self.client.create_dirnode()
44 def _made_upload_dir(n):
45 self.failUnless(IDirectoryNode.providedBy(n))
46 self.upload_dirnode = n
47 self.upload_dircap = n.get_uri()
48 self.uploader = DropUploader(self.client, self.upload_dircap, self.local_dir.encode('utf-8'),
50 return self.uploader.startService()
51 d.addCallback(_made_upload_dir)
53 # Write something short enough for a LIT file.
54 d.addCallback(lambda ign: self._test_file(u"short", "test"))
56 # Write to the same file again with different data.
57 d.addCallback(lambda ign: self._test_file(u"short", "different"))
59 # Test that temporary files are not uploaded.
60 d.addCallback(lambda ign: self._test_file(u"tempfile", "test", temporary=True))
62 # Test that we tolerate creation of a subdirectory.
63 d.addCallback(lambda ign: os.mkdir(os.path.join(self.local_dir, u"directory")))
65 # Write something longer, and also try to test a Unicode name if the fs can represent it.
66 name_u = self.unicode_or_fallback(u"l\u00F8ng", u"long")
67 d.addCallback(lambda ign: self._test_file(name_u, "test"*100))
69 # TODO: test that causes an upload failure.
70 d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_failed'), 0))
72 # Prevent unclean reactor errors.
74 d = defer.succeed(None)
75 if self.uploader is not None:
76 d.addCallback(lambda ign: self.uploader.finish(for_tests=True))
77 d.addCallback(lambda ign: res)
82 def _test_file(self, name_u, data, temporary=False):
83 previously_uploaded = self._get_count('drop_upload.files_uploaded')
84 previously_disappeared = self._get_count('drop_upload.files_disappeared')
88 # Note: this relies on the fact that we only get one IN_CLOSE_WRITE notification per file
89 # (otherwise we would get a defer.AlreadyCalledError). Should we be relying on that?
90 self.uploader.set_uploaded_callback(d.callback)
92 path_u = abspath_expanduser_unicode(name_u, base=self.local_dir)
93 path = to_filepath(path_u)
95 # We don't use FilePath.setContent() here because it creates a temporary file that
96 # is renamed into place, which causes events that the test is not expecting.
97 f = open(path_u, "wb")
99 if temporary and sys.platform != "win32":
104 if temporary and sys.platform == "win32":
106 fileutil.flush_volume(path_u)
107 self.notify_close_write(path)
110 d.addCallback(lambda ign: self.shouldFail(NoSuchChildError, 'temp file not uploaded', None,
111 self.upload_dirnode.get, name_u))
112 d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_disappeared'),
113 previously_disappeared + 1))
115 d.addCallback(lambda ign: self.upload_dirnode.get(name_u))
116 d.addCallback(download_to_data)
117 d.addCallback(lambda actual_data: self.failUnlessReallyEqual(actual_data, data))
118 d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_uploaded'),
119 previously_uploaded + 1))
121 d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_queued'), 0))
125 class MockTest(MagicFolderTestMixin, unittest.TestCase):
126 """This can run on any platform, and even if twisted.internet.inotify can't be imported."""
128 def test_errors(self):
131 errors_dir = abspath_expanduser_unicode(u"errors_dir", base=self.basedir)
133 not_a_dir = abspath_expanduser_unicode(u"NOT_A_DIR", base=self.basedir)
134 fileutil.write(not_a_dir, "")
135 magicfolderdb = abspath_expanduser_unicode(u"magicfolderdb", base=self.basedir)
136 doesnotexist = abspath_expanduser_unicode(u"doesnotexist", base=self.basedir)
138 client = self.g.clients[0]
139 d = client.create_dirnode()
140 def _made_upload_dir(n):
141 self.failUnless(IDirectoryNode.providedBy(n))
142 upload_dircap = n.get_uri()
143 readonly_dircap = n.get_readonly_uri()
145 self.shouldFail(AssertionError, 'nonexistent local.directory', 'there is no directory',
146 MagicFolder, client, upload_dircap, doesnotexist, inotify=fake_inotify)
147 self.shouldFail(AssertionError, 'non-directory local.directory', 'is not a directory',
148 MagicFolder, client, upload_dircap, not_a_dir, inotify=fake_inotify)
149 self.shouldFail(AssertionError, 'bad upload.dircap', 'does not refer to a directory',
150 MagicFolder, client, 'bad', errors_dir, inotify=fake_inotify)
151 self.shouldFail(AssertionError, 'non-directory upload.dircap', 'does not refer to a directory',
152 MagicFolder, client, 'URI:LIT:foo', errors_dir, inotify=fake_inotify)
153 self.shouldFail(AssertionError, 'readonly upload.dircap', 'is not a writecap to a directory',
154 MagicFolder, client, readonly_dircap, errors_dir, inotify=fake_inotify)
155 d.addCallback(_made_upload_dir)
158 def test_drop_upload(self):
159 self.inotify = fake_inotify
160 self.basedir = "drop_upload.MockTest.test_drop_upload"
163 def notify_close_write(self, path):
164 self.uploader._notifier.event(path, self.inotify.IN_CLOSE_WRITE)
167 class RealTest(MagicFolderTestMixin, unittest.TestCase):
168 """This is skipped unless both Twisted and the platform support inotify."""
170 def test_drop_upload(self):
171 self.inotify = None # use the appropriate inotify for the platform
172 self.basedir = "drop_upload.RealTest.test_drop_upload"
175 def notify_close_write(self, path):
176 # Writing to the file causes the notification.
179 if sys.platform != "win32" and not runtime.platform.supportsINotify():
180 RealTest.skip = "Magic Folder support can only be tested for-real on an OS that supports inotify or equivalent."