# test_client.py -- Tests for the git protocol, client side
# Copyright (C) 2009 Jelmer Vernooij <jelmer@jelmer.uk>
#
# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
# General Public License as public by the Free Software Foundation; version 2.0
# or (at your option) any later version. You can redistribute it and/or
# modify it under the terms of either of these two licenses.
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# You should have received a copy of the licenses; if not, see
# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
# License, Version 2.0.
#
from io import BytesIO
import base64
import sys
import shutil
import tempfile
import warnings
try:
from urllib import quote as urlquote
except ImportError:
from urllib.parse import quote as urlquote
try:
import urlparse
except ImportError:
import urllib.parse as urlparse
import urllib3
import dulwich
from dulwich import (
client,
)
from dulwich.client import (
InvalidWants,
LocalGitClient,
TraditionalGitClient,
TCPGitClient,
SSHGitClient,
HttpGitClient,
ReportStatusParser,
SendPackError,
StrangeHostname,
SubprocessSSHVendor,
PLinkSSHVendor,
UpdateRefsError,
check_wants,
default_urllib3_manager,
get_transport_and_path,
get_transport_and_path_from_url,
parse_rsync_url,
)
from dulwich.config import (
ConfigDict,
)
from dulwich.tests import (
TestCase,
)
from dulwich.protocol import (
TCP_GIT_PORT,
Protocol,
)
from dulwich.pack import (
pack_objects_to_data,
write_pack_data,
write_pack_objects,
)
from dulwich.objects import (
Commit,
Tree
)
from dulwich.repo import (
MemoryRepo,
Repo,
)
from dulwich.tests import skipIf
from dulwich.tests.utils import (
open_repo,
tear_down_repo,
setup_warning_catcher,
)
class DummyClient(TraditionalGitClient):
def __init__(self, can_read, read, write):
self.can_read = can_read
self.read = read
self.write = write
TraditionalGitClient.__init__(self)
def _connect(self, service, path):
return Protocol(self.read, self.write), self.can_read, None
class DummyPopen():
def __init__(self, *args, **kwards):
self.stdin = BytesIO(b"stdin")
self.stdout = BytesIO(b"stdout")
self.stderr = BytesIO(b"stderr")
self.returncode = 0
self.args = args
self.kwargs = kwards
def communicate(self, *args, **kwards):
return ('Running', '')
def wait(self, *args, **kwards):
return False
# TODO(durin42): add unit-level tests of GitClient
class GitClientTests(TestCase):
def setUp(self):
super(GitClientTests, self).setUp()
self.rout = BytesIO()
self.rin = BytesIO()
self.client = DummyClient(lambda x: True, self.rin.read,
self.rout.write)
def test_caps(self):
agent_cap = (
'agent=dulwich/%d.%d.%d' % dulwich.__version__).encode('ascii')
self.assertEqual(set([b'multi_ack', b'side-band-64k', b'ofs-delta',
b'thin-pack', b'multi_ack_detailed', b'shallow',
agent_cap]),
set(self.client._fetch_capabilities))
self.assertEqual(set([b'ofs-delta', b'report-status', b'side-band-64k',
agent_cap]),
set(self.client._send_capabilities))
def test_archive_ack(self):
self.rin.write(
b'0009NACK\n'
b'0000')
self.rin.seek(0)
self.client.archive(b'bla', b'HEAD', None, None)
self.assertEqual(self.rout.getvalue(), b'0011argument HEAD0000')
def test_fetch_empty(self):
self.rin.write(b'0000')
self.rin.seek(0)
def check_heads(heads):
self.assertEqual(heads, {})
return []
ret = self.client.fetch_pack(b'/', check_heads, None, None)
self.assertEqual({}, ret.refs)
self.assertEqual({}, ret.symrefs)
def test_fetch_pack_ignores_magic_ref(self):
self.rin.write(
b'00000000000000000000000000000000000000000000 capabilities^{}'
b'\x00 multi_ack '
b'thin-pack side-band side-band-64k ofs-delta shallow no-progress '
b'include-tag\n'
b'0000')
self.rin.seek(0)
def check_heads(heads):
self.assertEqual({}, heads)
return []
ret = self.client.fetch_pack(b'bla', check_heads, None, None, None)
self.assertEqual({}, ret.refs)
self.assertEqual({}, ret.symrefs)
self.assertEqual(self.rout.getvalue(), b'0000')
def test_fetch_pack_none(self):
self.rin.write(
b'008855dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7 HEAD\x00multi_ack '
b'thin-pack side-band side-band-64k ofs-delta shallow no-progress '
b'include-tag\n'
b'0000')
self.rin.seek(0)
ret = self.client.fetch_pack(
b'bla', lambda heads: [], None, None, None)
self.assertEqual(
{b'HEAD': b'55dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7'},
ret.refs)
self.assertEqual({}, ret.symrefs)
self.assertEqual(self.rout.getvalue(), b'0000')
def test_fetch_pack_sha_not_in_ref(self):
self.rin.write(
b'008855dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7 HEAD\x00multi_ack '
b'thin-pack side-band side-band-64k ofs-delta shallow no-progress '
b'include-tag\n'
b'0000')
self.rin.seek(0)
self.assertRaises(
InvalidWants, self.client.fetch_pack,
b'bla',
lambda heads: ['aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'],
None, None,
None)
def test_send_pack_no_sideband64k_with_update_ref_error(self):
# No side-bank-64k reported by server shouldn't try to parse
# side band data
pkts = [b'55dcc6bf963f922e1ed5c4bbaaefcfacef57b1d7 capabilities^{}'
b'\x00 report-status delete-refs ofs-delta\n',
b'',
b"unpack ok",
b"ng refs/foo/bar pre-receive hook declined",
b'']
for pkt in pkts:
if pkt == b'':
self.rin.write(b"0000")
else:
self.rin.write(("%04x" % (len(pkt)+4)).encode('ascii') + pkt)
self.rin.seek(0)
tree = Tree()
commit = Commit()
commit.tree = tree
commit.parents = []
commit.author = commit.committer = b'test user'
commit.commit_time = commit.author_time = 1174773719
commit.commit_timezone = commit.author_timezone = 0
commit.encoding = b'UTF-8'
commit.message = b'test message'
def update_refs(refs):
return {b'refs/foo/bar': commit.id, }
def generate_pack_data(have, want, ofs_delta=False):
return pack_objects_to_data([(commit, None), (tree, ''), ])
self.assertRaises(UpdateRefsError,
self.client.send_pack, "blah",
update_refs, generate_pack_data)
def test_send_pack_none(self):
self.rin.write(
b'0078310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'refs/heads/master\x00 report-status delete-refs '
b'side-band-64k quiet ofs-delta\n'
b'0000')
self.rin.seek(0)
def update_refs(refs):
return {
b'refs/heads/master':
b'310ca9477129b8586fa2afc779c1f57cf64bba6c'
}
def generate_pack_data(have, want, ofs_delta=False):
return 0, []
self.client.send_pack(b'/', update_refs, generate_pack_data)
self.assertEqual(self.rout.getvalue(), b'0000')
def test_send_pack_keep_and_delete(self):
self.rin.write(
b'0063310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'refs/heads/master\x00report-status delete-refs ofs-delta\n'
b'003f310ca9477129b8586fa2afc779c1f57cf64bba6c refs/heads/keepme\n'
b'0000000eunpack ok\n'
b'0019ok refs/heads/master\n'
b'0000')
self.rin.seek(0)
def update_refs(refs):
return {b'refs/heads/master': b'0' * 40}
def generate_pack_data(have, want, ofs_delta=False):
return 0, []
self.client.send_pack(b'/', update_refs, generate_pack_data)
self.assertIn(
self.rout.getvalue(),
[b'007f310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'0000000000000000000000000000000000000000 '
b'refs/heads/master\x00report-status ofs-delta0000',
b'007f310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'0000000000000000000000000000000000000000 '
b'refs/heads/master\x00ofs-delta report-status0000'])
def test_send_pack_delete_only(self):
self.rin.write(
b'0063310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'refs/heads/master\x00report-status delete-refs ofs-delta\n'
b'0000000eunpack ok\n'
b'0019ok refs/heads/master\n'
b'0000')
self.rin.seek(0)
def update_refs(refs):
return {b'refs/heads/master': b'0' * 40}
def generate_pack_data(have, want, ofs_delta=False):
return 0, []
self.client.send_pack(b'/', update_refs, generate_pack_data)
self.assertIn(
self.rout.getvalue(),
[b'007f310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'0000000000000000000000000000000000000000 '
b'refs/heads/master\x00report-status ofs-delta0000',
b'007f310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'0000000000000000000000000000000000000000 '
b'refs/heads/master\x00ofs-delta report-status0000'])
def test_send_pack_new_ref_only(self):
self.rin.write(
b'0063310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'refs/heads/master\x00report-status delete-refs ofs-delta\n'
b'0000000eunpack ok\n'
b'0019ok refs/heads/blah12\n'
b'0000')
self.rin.seek(0)
def update_refs(refs):
return {
b'refs/heads/blah12':
b'310ca9477129b8586fa2afc779c1f57cf64bba6c',
b'refs/heads/master':
b'310ca9477129b8586fa2afc779c1f57cf64bba6c'
}
def generate_pack_data(have, want, ofs_delta=False):
return 0, []
f = BytesIO()
write_pack_objects(f, {})
self.client.send_pack('/', update_refs, generate_pack_data)
self.assertIn(
self.rout.getvalue(),
[b'007f0000000000000000000000000000000000000000 '
b'310ca9477129b8586fa2afc779c1f57cf64bba6c '
b'refs/heads/blah12\x00report-status ofs-delta0000' +
f.getvalue(),
b'007f0000000000000000000000000000000000000000 '
Loading ...