import os, sys from twisted.trial import unittest from twisted.python import runtime from twisted.internet import defer from allmydata.interfaces import IDirectoryNode, NoSuchChildError from allmydata.util import fake_inotify, fileutil from allmydata.util.encodingutil import get_filesystem_encoding, to_filepath from allmydata.util.consumer import download_to_data from allmydata.test.no_network import GridTestMixin from allmydata.test.common_util import ReallyEqualMixin, NonASCIIPathMixin from allmydata.test.common import ShouldFailMixin from allmydata.frontends.drop_upload import DropUploader from allmydata.util.fileutil import abspath_expanduser_unicode class DropUploadTestMixin(GridTestMixin, ShouldFailMixin, ReallyEqualMixin, NonASCIIPathMixin): """ These tests will be run both with a mock notifier, and (on platforms that support it) with the real INotify. """ def setUp(self): GridTestMixin.setUp(self) temp = self.mktemp() self.basedir = abspath_expanduser_unicode(temp.decode(get_filesystem_encoding())) def _get_count(self, name): return self.stats_provider.get_stats()["counters"].get(name, 0) def _test(self): self.uploader = None self.set_up_grid() self.local_dir = os.path.join(self.basedir, self.unicode_or_fallback(u"loc\u0101l_dir", u"local_dir")) self.mkdir_nonascii(self.local_dir) self.client = self.g.clients[0] self.stats_provider = self.client.stats_provider d = self.client.create_dirnode() def _made_upload_dir(n): self.failUnless(IDirectoryNode.providedBy(n)) self.upload_dirnode = n self.upload_dircap = n.get_uri() self.uploader = DropUploader(self.client, self.upload_dircap, self.local_dir.encode('utf-8'), inotify=self.inotify) return self.uploader.startService() d.addCallback(_made_upload_dir) # Write something short enough for a LIT file. d.addCallback(lambda ign: self._test_file(u"short", "test")) # Write to the same file again with different data. d.addCallback(lambda ign: self._test_file(u"short", "different")) # Test that temporary files are not uploaded. d.addCallback(lambda ign: self._test_file(u"tempfile", "test", temporary=True)) # Test that we tolerate creation of a subdirectory. d.addCallback(lambda ign: os.mkdir(os.path.join(self.local_dir, u"directory"))) # Write something longer, and also try to test a Unicode name if the fs can represent it. name_u = self.unicode_or_fallback(u"l\u00F8ng", u"long") d.addCallback(lambda ign: self._test_file(name_u, "test"*100)) # TODO: test that causes an upload failure. d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_failed'), 0)) # Prevent unclean reactor errors. def _cleanup(res): d = defer.succeed(None) if self.uploader is not None: d.addCallback(lambda ign: self.uploader.finish(for_tests=True)) d.addCallback(lambda ign: res) return d d.addBoth(_cleanup) return d def _test_file(self, name_u, data, temporary=False): previously_uploaded = self._get_count('drop_upload.files_uploaded') previously_disappeared = self._get_count('drop_upload.files_disappeared') d = defer.Deferred() # Note: this relies on the fact that we only get one IN_CLOSE_WRITE notification per file # (otherwise we would get a defer.AlreadyCalledError). Should we be relying on that? self.uploader.set_uploaded_callback(d.callback) path_u = abspath_expanduser_unicode(name_u, base=self.local_dir) path = to_filepath(path_u) # We don't use FilePath.setContent() here because it creates a temporary file that # is renamed into place, which causes events that the test is not expecting. f = open(path_u, "wb") try: if temporary and sys.platform != "win32": os.unlink(path_u) f.write(data) finally: f.close() if temporary and sys.platform == "win32": os.unlink(path_u) fileutil.flush_volume(path_u) self.notify_close_write(path) if temporary: d.addCallback(lambda ign: self.shouldFail(NoSuchChildError, 'temp file not uploaded', None, self.upload_dirnode.get, name_u)) d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_disappeared'), previously_disappeared + 1)) else: d.addCallback(lambda ign: self.upload_dirnode.get(name_u)) d.addCallback(download_to_data) d.addCallback(lambda actual_data: self.failUnlessReallyEqual(actual_data, data)) d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_uploaded'), previously_uploaded + 1)) d.addCallback(lambda ign: self.failUnlessReallyEqual(self._get_count('drop_upload.files_queued'), 0)) return d class MockTest(DropUploadTestMixin, unittest.TestCase): """This can run on any platform, and even if twisted.internet.inotify can't be imported.""" def test_errors(self): self.set_up_grid() errors_dir = abspath_expanduser_unicode(u"errors_dir", base=self.basedir) os.mkdir(errors_dir) not_a_dir = abspath_expanduser_unicode(u"NOT_A_DIR", base=self.basedir) fileutil.write(not_a_dir, "") magicfolderdb = abspath_expanduser_unicode(u"magicfolderdb", base=self.basedir) doesnotexist = abspath_expanduser_unicode(u"doesnotexist", base=self.basedir) client = self.g.clients[0] d = client.create_dirnode() def _made_upload_dir(n): self.failUnless(IDirectoryNode.providedBy(n)) upload_dircap = n.get_uri() readonly_dircap = n.get_readonly_uri() self.shouldFail(AssertionError, 'nonexistent local.directory', 'there is no directory', DropUploader, client, upload_dircap, doesnotexist, inotify=fake_inotify) self.shouldFail(AssertionError, 'non-directory local.directory', 'is not a directory', DropUploader, client, upload_dircap, not_a_dir, inotify=fake_inotify) self.shouldFail(AssertionError, 'bad upload.dircap', 'does not refer to a directory', DropUploader, client, 'bad', errors_dir, inotify=fake_inotify) self.shouldFail(AssertionError, 'non-directory upload.dircap', 'does not refer to a directory', DropUploader, client, 'URI:LIT:foo', errors_dir, inotify=fake_inotify) self.shouldFail(AssertionError, 'readonly upload.dircap', 'is not a writecap to a directory', DropUploader, client, readonly_dircap, errors_dir, inotify=fake_inotify) d.addCallback(_made_upload_dir) return d def test_drop_upload(self): self.inotify = fake_inotify self.basedir = "drop_upload.MockTest.test_drop_upload" return self._test() def notify_close_write(self, path): self.uploader._notifier.event(path, self.inotify.IN_CLOSE_WRITE) class RealTest(DropUploadTestMixin, unittest.TestCase): """This is skipped unless both Twisted and the platform support inotify.""" def test_drop_upload(self): self.inotify = None # use the appropriate inotify for the platform self.basedir = "drop_upload.RealTest.test_drop_upload" return self._test() def notify_close_write(self, path): # Writing to the file causes the notification. pass if sys.platform != "win32" and not runtime.platform.supportsINotify(): RealTest.skip = "Drop-upload support can only be tested for-real on an OS that supports inotify or equivalent."