from twisted.internet import reactor
from twisted.python import usage
+from allmydata.util.assertutil import precondition
from allmydata.util import fileutil
-from allmydata.util.fileutil import precondition_abspath
from allmydata.scripts.common import get_aliases
from allmydata.test.no_network import GridTestMixin
from .test_cli import CLITestMixin
def do_invite(self, client_num, nickname):
nickname_arg = unicode_to_argv(nickname)
d = self.do_cli("magic-folder", "invite", "magic:", nickname_arg, client_num=client_num)
- def _done((rc,stdout,stderr)):
- self.failUnless(rc == 0)
- return (rc,stdout,stderr)
+ def _done((rc, stdout, stderr)):
+ self.failUnlessEqual(rc, 0)
+ return (rc, stdout, stderr)
d.addCallback(_done)
return d
def do_join(self, client_num, local_dir, invite_code):
- magic_readonly_cap, dmd_write_cap = invite_code.split(magic_folder_cli.INVITE_SEPARATOR)
- d = self.do_cli("magic-folder", "join", invite_code, local_dir, client_num=client_num)
- def _done((rc,stdout,stderr)):
- self.failUnless(rc == 0)
- return (rc,stdout,stderr)
+ precondition(isinstance(local_dir, unicode), local_dir=local_dir)
+ precondition(isinstance(invite_code, str), invite_code=invite_code)
+
+ local_dir_arg = unicode_to_argv(local_dir)
+ d = self.do_cli("magic-folder", "join", invite_code, local_dir_arg, client_num=client_num)
+ def _done((rc, stdout, stderr)):
+ self.failUnlessEqual(rc, 0)
+ return (rc, stdout, stderr)
d.addCallback(_done)
return d
"""Tests that our collective directory has the readonly cap of
our upload directory.
"""
- collective_readonly_cap = fileutil.read(os.path.join(self.get_clientdir(i=client_num), "private/collective_dircap"))
+ collective_readonly_cap = fileutil.read(os.path.join(self.get_clientdir(i=client_num),
+ u"private", u"collective_dircap"))
d = self.do_cli("ls", "--json", collective_readonly_cap, client_num=client_num)
- def _done((rc,stdout,stderr)):
- self.failUnless(rc == 0)
- return (rc,stdout,stderr)
+ def _done((rc, stdout, stderr)):
+ self.failUnlessEqual(rc, 0)
+ return (rc, stdout, stderr)
d.addCallback(_done)
def test_joined_magic_folder((rc,stdout,stderr)):
readonly_cap = unicode(uri.from_string(upload_dircap).get_readonly().to_string(), 'utf-8')
return d
def get_caps_from_files(self, client_num):
- collective_dircap = fileutil.read(os.path.join(self.get_clientdir(i=client_num), "private/collective_dircap"))
- upload_dircap = fileutil.read(os.path.join(self.get_clientdir(i=client_num), "private/magic_folder_dircap"))
+ collective_dircap = fileutil.read(os.path.join(self.get_clientdir(i=client_num),
+ u"private", u"collective_dircap"))
+ upload_dircap = fileutil.read(os.path.join(self.get_clientdir(i=client_num),
+ u"private", u"magic_folder_dircap"))
self.failIf(collective_dircap is None or upload_dircap is None)
return collective_dircap, upload_dircap
def check_config(self, client_num, local_dir):
- precondition_abspath(local_dir)
client_config = fileutil.read(os.path.join(self.get_clientdir(i=client_num), "tahoe.cfg"))
local_dir_utf8 = local_dir.encode('utf-8')
- ret = re.search("\[magic_folder\]\nenabled = True\nlocal.directory = %s" % (local_dir_utf8,), client_config)
- self.failIf(ret is None)
+ magic_folder_config = "[magic_folder]\nenabled = True\nlocal.directory = %s" % (local_dir_utf8,)
+ self.failUnlessIn(magic_folder_config, client_config)
def create_invite_join_magic_folder(self, nickname, local_dir):
nickname_arg = unicode_to_argv(nickname)
local_dir_arg = unicode_to_argv(local_dir)
d = self.do_cli("magic-folder", "create", "magic:", nickname_arg, local_dir_arg)
def _done((rc, stdout, stderr)):
- self.failUnless(rc == 0)
+ self.failUnlessEqual(rc, 0)
client = self.get_client()
self.collective_dircap, self.upload_dircap = self.get_caps_from_files(0)
# invites herself then and joins.
d = self.do_create_magic_folder(0)
d.addCallback(lambda ign: self.do_invite(0, u"Alice\u00F8"))
- def get_invitecode(result):
- self.invitecode = result[1].strip()
- d.addCallback(get_invitecode)
- d.addCallback(lambda ign: self.do_join(0, alice_magic_dir, self.invitecode))
+ def get_invite_code(result):
+ self.invite_code = result[1].strip()
+ d.addCallback(get_invite_code)
+ d.addCallback(lambda ign: self.do_join(0, alice_magic_dir, self.invite_code))
def get_alice_caps(ign):
self.alice_collective_dircap, self.alice_upload_dircap = self.get_caps_from_files(0)
d.addCallback(get_alice_caps)
# Alice invites Bob. Bob joins.
d.addCallback(lambda ign: self.do_invite(0, u"Bob\u00F8"))
- def get_invitecode(result):
- self.invitecode = result[1].strip()
- d.addCallback(get_invitecode)
- d.addCallback(lambda ign: self.do_join(1, bob_magic_dir, self.invitecode))
+ def get_invite_code(result):
+ self.invite_code = result[1].strip()
+ d.addCallback(get_invite_code)
+ d.addCallback(lambda ign: self.do_join(1, bob_magic_dir, self.invite_code))
def get_bob_caps(ign):
self.bob_collective_dircap, self.bob_upload_dircap = self.get_caps_from_files(1)
d.addCallback(get_bob_caps)
self.basedir = "cli/MagicFolder/create-and-then-invite-join"
self.set_up_grid()
local_dir = os.path.join(self.basedir, "magic")
+ abs_local_dir_u = abspath_expanduser_unicode(unicode(local_dir), long_path=False)
d = self.do_create_magic_folder(0)
d.addCallback(lambda ign: self.do_invite(0, u"Alice"))
def get_invite_code_and_join((rc, stdout, stderr)):
invite_code = stdout.strip()
- return self.do_join(0, local_dir, invite_code)
+ return self.do_join(0, unicode(local_dir), invite_code)
d.addCallback(get_invite_code_and_join)
def get_caps(ign):
self.collective_dircap, self.upload_dircap = self.get_caps_from_files(0)
d.addCallback(get_caps)
d.addCallback(lambda ign: self.check_joined_config(0, self.upload_dircap))
- d.addCallback(lambda ign: self.check_config(0, abspath_expanduser_unicode(unicode(local_dir))))
+ d.addCallback(lambda ign: self.check_config(0, abs_local_dir_u))
return d
def test_create_error(self):
self.basedir = "cli/MagicFolder/create-invite-join"
self.set_up_grid()
local_dir = os.path.join(self.basedir, "magic")
+ abs_local_dir_u = abspath_expanduser_unicode(unicode(local_dir), long_path=False)
d = self.do_cli("magic-folder", "create", "magic:", "Alice", local_dir)
def _done((rc, stdout, stderr)):
- self.failUnless(rc == 0)
- d.addCallback(_done)
- def get_caps(ign):
+ self.failUnlessEqual(rc, 0)
self.collective_dircap, self.upload_dircap = self.get_caps_from_files(0)
- d.addCallback(get_caps)
+ d.addCallback(_done)
d.addCallback(lambda ign: self.check_joined_config(0, self.upload_dircap))
- d.addCallback(lambda ign: self.check_config(0, abspath_expanduser_unicode(unicode(local_dir))))
+ d.addCallback(lambda ign: self.check_config(0, abs_local_dir_u))
return d
def test_create_invite_join_failure(self):