with repo.Repo(srcpath) as src:
sendrefs = dict(src.get_refs())
del sendrefs[b'HEAD']
- c.send_pack(self._build_path(b'/dest'), lambda _: sendrefs,
+ c.send_pack(self._build_path('/dest'), lambda _: sendrefs,
src.object_store.generate_pack_contents)
def test_send_pack(self):
with repo.Repo(srcpath) as src:
sendrefs = dict(src.get_refs())
del sendrefs[b'HEAD']
- c.send_pack(self._build_path(b'/dest'), lambda _: sendrefs,
+ c.send_pack(self._build_path('/dest'), lambda _: sendrefs,
src.object_store.generate_pack_contents)
self.assertDestEqualsSrc()
sendrefs, gen_pack = self.compute_send(src)
c = self._client()
try:
- c.send_pack(self._build_path(b'/dest'),
+ c.send_pack(self._build_path('/dest'),
lambda _: sendrefs, gen_pack)
except errors.UpdateRefsError as e:
self.assertEqual('refs/heads/master failed to update',
sendrefs, gen_pack = self.compute_send(src)
c = self._client()
try:
- c.send_pack(self._build_path(b'/dest'), lambda _: sendrefs, gen_pack)
+ c.send_pack(self._build_path('/dest'), lambda _: sendrefs, gen_pack)
except errors.UpdateRefsError as e:
self.assertIn(str(e),
['{0}, {1} failed to update'.format(
def test_archive(self):
c = self._client()
f = BytesIO()
- c.archive(self._build_path(b'/server_new.export'), b'HEAD', f.write)
+ c.archive(self._build_path('/server_new.export'), b'HEAD', f.write)
f.seek(0)
tf = tarfile.open(fileobj=f)
self.assertEqual(['baz', 'foo'], tf.getnames())
def test_fetch_pack(self):
c = self._client()
with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
- refs = c.fetch(self._build_path(b'/server_new.export'), dest)
+ refs = c.fetch(self._build_path('/server_new.export'), dest)
for r in refs.items():
dest.refs.set_if_equals(r[0], None, r[1])
self.assertDestEqualsSrc()
c = self._client()
repo_dir = os.path.join(self.gitroot, 'server_new.export')
with repo.Repo(repo_dir) as dest:
- refs = c.fetch(self._build_path(b'/dest'), dest)
+ refs = c.fetch(self._build_path('/dest'), dest)
for r in refs.items():
dest.refs.set_if_equals(r[0], None, r[1])
self.assertDestEqualsSrc()
c = self._client()
c._fetch_capabilities.remove(b'side-band-64k')
with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
- refs = c.fetch(self._build_path(b'/server_new.export'), dest)
+ refs = c.fetch(self._build_path('/server_new.export'), dest)
for r in refs.items():
dest.refs.set_if_equals(r[0], None, r[1])
self.assertDestEqualsSrc()
c = self._client()
with repo.Repo(os.path.join(self.gitroot, 'dest')) as dest:
refs = c.fetch(
- self._build_path(b'/server_new.export'), dest,
+ self._build_path('/server_new.export'), dest,
lambda refs: [protocol.ZERO_SHA])
for r in refs.items():
dest.refs.set_if_equals(r[0], None, r[1])
c = self._client()
self.assertEqual(dest.refs[b"refs/heads/abranch"], dummy_commit)
c.send_pack(
- self._build_path(b'/dest'), lambda _: sendrefs, gen_pack)
+ self._build_path('/dest'), lambda _: sendrefs, gen_pack)
self.assertFalse(b"refs/heads/abranch" in dest.refs)
def test_get_refs(self):
c = self._client()
- refs = c.get_refs(self._build_path(b'/server_new.export'))
+ refs = c.get_refs(self._build_path('/server_new.export'))
repo_dir = os.path.join(self.gitroot, 'server_new.export')
with repo.Repo(repo_dir) as dest:
return client.SSHGitClient('localhost')
def _build_path(self, path):
- return self.gitroot.encode(sys.getfilesystemencoding()) + path
+ return self.gitroot + path
class DulwichSubprocessClientTest(CompatTestCase, DulwichClientTestBase):
return client.SubprocessGitClient(stderr=subprocess.PIPE)
def _build_path(self, path):
- return self.gitroot.encode(sys.getfilesystemencoding()) + path
+ return self.gitroot + path
class GitHTTPRequestHandler(SimpleHTTPServer.SimpleHTTPRequestHandler):
return client.HttpGitClient(self._httpd.get_url())
def _build_path(self, path):
- if sys.version_info[0] == 3:
- return path.decode('ascii')
- else:
- return path
+ return path
def test_archive(self):
raise SkipTest("exporting archives not supported over http")