commit - 568cc3367f4e298b55a2f1d0ac5e6756a7093b68
commit + 2cdd2e71c18488c534a7b3df37c37eba31d3811f
blob - 4263bf409b25caea88ee8477d865f9009b26ea69
blob + 956ea765820765587c72d393dd007aa4fb95592a
--- NEWS
+++ NEWS
+0.20.44 2022-06-30
+
+ * Fix reading of chunks in server. (Jelmer Vernooij, #977)
+
+ * Support applying of URL rewriting using ``insteadOf`` / ``pushInsteadOf``.
+ (Jelmer Vernooij, #706)
+
0.20.43 2022-06-07
* Lazily import url2pathname.
blob - 5daca8bf089b001f523873510c0fc3ee5bffe1ac
blob + 3de0b5b9f6f458c21bdbe15bb6a586b4bbcc5588
--- PKG-INFO
+++ PKG-INFO
Metadata-Version: 2.1
Name: dulwich
-Version: 0.20.43
+Version: 0.20.44
Summary: Python Git Library
Home-page: https://www.dulwich.io/
Author: Jelmer Vernooij
blob - 308456db2c3def6e69af28207a56e12c0ada98da
blob + 46d32c43fb08c7311dc9994ee088f36c3aee1b8e
--- debian/changelog
+++ debian/changelog
+dulwich (0.20.44-1) UNRELEASED; urgency=low
+
+ * New upstream release.
+
+ -- Jelmer Vernooij <jelmer@debian.org> Thu, 30 Jun 2022 19:37:46 -0000
+
dulwich (0.20.43-1) unstable; urgency=low
* New upstream release.
blob - 1645488ae85bca0627e86cebf0a6397d1fcda0f0
blob + 4544902ed0a6ad3a3d904cd6c64d39e9cb72860f
--- dulwich/__init__.py
+++ dulwich/__init__.py
"""Python implementation of the Git file formats and protocols."""
-__version__ = (0, 20, 43)
+__version__ = (0, 20, 44)
blob - 168e28f0da1198afd1fbb89e87ebedda13f17406
blob + 19f540e69893a800afb90122765f0e02d003faf1
--- dulwich/client.py
+++ dulwich/client.py
import socket
import subprocess
import sys
-from typing import Any, Callable, Dict, List, Optional, Set, Tuple, IO
+from typing import Any, Callable, Dict, List, Optional, Set, Tuple, IO, Iterable
from urllib.parse import (
quote as urlquote,
import dulwich
-from dulwich.config import get_xdg_config_home_path
+from dulwich.config import get_xdg_config_home_path, Config
from dulwich.errors import (
GitProtocolError,
NotGitRepository,
if username is not None:
# No escaping needed: ":" is not allowed in username:
# https://tools.ietf.org/html/rfc2617#section-2
- credentials = "%s:%s" % (username, password)
+ credentials = f"{username}:{password or ''}"
import urllib3.util
basic_auth = urllib3.util.make_headers(basic_auth=credentials)
return url2pathname(netloc + path) # type: ignore
-def get_transport_and_path_from_url(url, config=None, **kwargs):
+def iter_instead_of(config: Config, push: bool = False) -> Iterable[Tuple[str, str]]:
+ """Iterate over insteadOf / pushInsteadOf values.
+ """
+ for section in config.sections():
+ if section[0] != b'url':
+ continue
+ replacement = section[1]
+ try:
+ needles = list(config.get_multivar(section, "insteadOf"))
+ except KeyError:
+ needles = []
+ if push:
+ try:
+ needles += list(config.get_multivar(section, "pushInsteadOf"))
+ except KeyError:
+ pass
+ for needle in needles:
+ yield needle.decode('utf-8'), replacement.decode('utf-8')
+
+
+def apply_instead_of(config: Config, orig_url: str, push: bool = False) -> str:
+ """Apply insteadOf / pushInsteadOf to a URL.
+ """
+ longest_needle = ""
+ updated_url = orig_url
+ for needle, replacement in iter_instead_of(config, push):
+ if not orig_url.startswith(needle):
+ continue
+ if len(longest_needle) < len(needle):
+ longest_needle = needle
+ updated_url = replacement + orig_url[len(needle):]
+ return updated_url
+
+
+def get_transport_and_path_from_url(
+ url: str, config: Optional[Config] = None,
+ operation: Optional[str] = None, **kwargs) -> Tuple[GitClient, str]:
"""Obtain a git client from a URL.
Args:
url: URL to open (a unicode string)
config: Optional config object
+ operation: Kind of operation that'll be performed; "pull" or "push"
thin_packs: Whether or not thin packs should be retrieved
report_activity: Optional callback for reporting transport
activity.
Tuple with client instance and relative path.
"""
+ if config is not None:
+ url = apply_instead_of(config, url, push=(operation == "push"))
parsed = urlparse(url)
if parsed.scheme == "git":
return (TCPGitClient.from_parsedurl(parsed, **kwargs), parsed.path)
raise ValueError("unknown scheme '%s'" % parsed.scheme)
-def parse_rsync_url(location):
+def parse_rsync_url(location: str) -> Tuple[Optional[str], str, str]:
"""Parse a rsync-style URL."""
if ":" in location and "@" not in location:
# SSH with no user@, zero or one leading slash.
def get_transport_and_path(
location: str,
+ operation: Optional[str] = None,
**kwargs: Any
) -> Tuple[GitClient, str]:
"""Obtain a git client from a URL.
Args:
location: URL or path (a string)
config: Optional config object
+ operation: Kind of operation that'll be performed; "pull" or "push"
thin_packs: Whether or not thin packs should be retrieved
report_activity: Optional callback for reporting transport
activity.
"""
# First, try to parse it as a URL
try:
- return get_transport_and_path_from_url(location, **kwargs)
+ return get_transport_and_path_from_url(location, operation=operation, **kwargs)
except ValueError:
pass
blob - 4ceecab182fd7d58ab8dab0dd4d0d18371d75afc
blob + e52d468e3bc66f6fa295218061a2656e207ac800
--- dulwich/config.py
+++ dulwich/config.py
import os
import sys
import warnings
-
from typing import (
BinaryIO,
Iterable,
Iterator,
KeysView,
+ List,
MutableMapping,
Optional,
Tuple,
Union,
+ overload,
)
from dulwich.file import GitFile
return self[key]
+Name = bytes
+NameLike = Union[bytes, str]
+Section = Tuple[bytes, ...]
+SectionLike = Union[bytes, str, Tuple[Union[bytes, str], ...]]
+Value = bytes
+ValueLike = Union[bytes, str]
+
+
class Config(object):
"""A Git configuration."""
- def get(self, section, name):
+ def get(self, section: SectionLike, name: NameLike) -> Value:
"""Retrieve the contents of a configuration setting.
Args:
- section: Tuple with section name and optional subsection namee
+ section: Tuple with section name and optional subsection name
name: Variable name
Returns:
Contents of the setting
"""
raise NotImplementedError(self.get)
- def get_multivar(self, section, name):
+ def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
"""Retrieve the contents of a multivar configuration setting.
Args:
"""
raise NotImplementedError(self.get_multivar)
- def get_boolean(self, section, name, default=None):
+ @overload
+ def get_boolean(self, section: SectionLike, name: NameLike, default: bool) -> bool:
+ ...
+
+ @overload
+ def get_boolean(self, section: SectionLike, name: NameLike) -> Optional[bool]:
+ ...
+
+ def get_boolean(
+ self, section: SectionLike, name: NameLike, default: Optional[bool] = None
+ ) -> Optional[bool]:
"""Retrieve a configuration setting as boolean.
Args:
subsection.
Returns:
Contents of the setting
- Raises:
- KeyError: if the value is not set
"""
try:
value = self.get(section, name)
return False
raise ValueError("not a valid boolean string: %r" % value)
- def set(self, section, name, value):
+ def set(
+ self,
+ section: SectionLike,
+ name: NameLike,
+ value: Union[ValueLike, bool]
+ ) -> None:
"""Set a configuration value.
Args:
"""
raise NotImplementedError(self.set)
- def items(self, section):
+ def items(self, section: SectionLike) -> Iterator[Tuple[Name, Value]]:
"""Iterate over the configuration pairs for a specific section.
Args:
"""
raise NotImplementedError(self.items)
- def iteritems(self, section):
+ def iteritems(self, section: SectionLike) -> Iterator[Tuple[Name, Value]]:
"""Iterate over the configuration pairs for a specific section.
Args:
)
return self.items(section)
- def itersections(self):
+ def itersections(self) -> Iterator[Section]:
warnings.warn(
"Use %s.items instead." % type(self).__name__,
DeprecationWarning,
)
return self.sections()
- def sections(self):
+ def sections(self) -> Iterator[Section]:
"""Iterate over the sections.
Returns: Iterator over section tuples
"""
raise NotImplementedError(self.sections)
- def has_section(self, name: Tuple[bytes, ...]) -> bool:
+ def has_section(self, name: Section) -> bool:
"""Check if a specified section exists.
Args:
return name in self.sections()
-BytesLike = Union[bytes, str]
-Key = Tuple[bytes, ...]
-KeyLike = Union[bytes, str, Tuple[BytesLike, ...]]
-Value = Union[bytes, bool]
-ValueLike = Union[bytes, str, bool]
-
-
-class ConfigDict(Config, MutableMapping[Key, MutableMapping[bytes, Value]]):
+class ConfigDict(Config, MutableMapping[Section, MutableMapping[Name, Value]]):
"""Git configuration stored in a dictionary."""
def __init__(
self,
values: Union[
- MutableMapping[Key, MutableMapping[bytes, Value]], None
+ MutableMapping[Section, MutableMapping[Name, Value]], None
] = None,
encoding: Union[str, None] = None
) -> None:
def __eq__(self, other: object) -> bool:
return isinstance(other, self.__class__) and other._values == self._values
- def __getitem__(self, key: Key) -> MutableMapping[bytes, Value]:
+ def __getitem__(self, key: Section) -> MutableMapping[Name, Value]:
return self._values.__getitem__(key)
def __setitem__(
self,
- key: Key,
- value: MutableMapping[bytes, Value]
+ key: Section,
+ value: MutableMapping[Name, Value]
) -> None:
return self._values.__setitem__(key, value)
- def __delitem__(self, key: Key) -> None:
+ def __delitem__(self, key: Section) -> None:
return self._values.__delitem__(key)
- def __iter__(self) -> Iterator[Key]:
+ def __iter__(self) -> Iterator[Section]:
return self._values.__iter__()
def __len__(self) -> int:
return self._values.__len__()
@classmethod
- def _parse_setting(cls, name):
+ def _parse_setting(cls, name: str):
parts = name.split(".")
if len(parts) == 3:
return (parts[0], parts[1], parts[2])
def _check_section_and_name(
self,
- section: KeyLike,
- name: BytesLike
- ) -> Tuple[Key, bytes]:
+ section: SectionLike,
+ name: NameLike
+ ) -> Tuple[Section, Name]:
if not isinstance(section, tuple):
section = (section,)
def get_multivar(
self,
- section: KeyLike,
- name: BytesLike
+ section: SectionLike,
+ name: NameLike
) -> Iterator[Value]:
section, name = self._check_section_and_name(section, name)
if len(section) > 1:
try:
- return self._values[section][name]
+ return self._values[section].get_all(name)
except KeyError:
pass
def get( # type: ignore[override]
self,
- section: KeyLike,
- name: BytesLike,
- ) -> Optional[Value]:
+ section: SectionLike,
+ name: NameLike,
+ ) -> Value:
section, name = self._check_section_and_name(section, name)
if len(section) > 1:
def set(
self,
- section: KeyLike,
- name: BytesLike,
- value: ValueLike,
+ section: SectionLike,
+ name: NameLike,
+ value: Union[ValueLike, bool],
) -> None:
section, name = self._check_section_and_name(section, name)
- if not isinstance(value, (bytes, bool)):
+ if isinstance(value, bool):
+ value = b"true" if value else b"false"
+
+ if not isinstance(value, bytes):
value = value.encode(self.encoding)
self._values.setdefault(section)[name] = value
def items( # type: ignore[override]
self,
- section: Key
- ) -> Iterator[Value]:
+ section: Section
+ ) -> Iterator[Tuple[Name, Value]]:
return self._values.get(section).items()
- def sections(self) -> Iterator[Key]:
+ def sections(self) -> Iterator[Section]:
return self._values.keys()
-def _format_string(value):
+def _format_string(value: bytes) -> bytes:
if (
value.startswith(b" ")
or value.startswith(b"\t")
_WHITESPACE_CHARS = [ord(b"\t"), ord(b" ")]
-def _parse_string(value):
+def _parse_string(value: bytes) -> bytes:
value = bytearray(value.strip())
ret = bytearray()
whitespace = bytearray()
return bytes(ret)
-def _escape_value(value):
+def _escape_value(value: bytes) -> bytes:
"""Escape a value."""
value = value.replace(b"\\", b"\\\\")
value = value.replace(b"\n", b"\\n")
return value
-def _check_variable_name(name):
+def _check_variable_name(name: bytes) -> bool:
for i in range(len(name)):
c = name[i : i + 1]
if not c.isalnum() and c != b"-":
return True
-def _check_section_name(name):
+def _check_section_name(name: bytes) -> bool:
for i in range(len(name)):
c = name[i : i + 1]
if not c.isalnum() and c not in (b"-", b"."):
return True
-def _strip_comments(line):
+def _strip_comments(line: bytes) -> bytes:
comment_bytes = {ord(b"#"), ord(b";")}
quote = ord(b'"')
string_open = False
class ConfigFile(ConfigDict):
"""A Git configuration file, like .git/config or ~/.gitconfig."""
- def __init__(self, values=None, encoding=None):
+ def __init__(
+ self,
+ values: Union[
+ MutableMapping[Section, MutableMapping[Name, Value]], None
+ ] = None,
+ encoding: Union[str, None] = None
+ ) -> None:
super(ConfigFile, self).__init__(values=values, encoding=encoding)
- self.path = None
+ self.path: Optional[str] = None
@classmethod # noqa: C901
def from_file(cls, f: BinaryIO) -> "ConfigFile": # noqa: C901
"""Read configuration from a file-like object."""
ret = cls()
- section = None # type: Optional[Tuple[bytes, ...]]
+ section: Optional[Section] = None
setting = None
continuation = None
for lineno, line in enumerate(f.readlines()):
return ret
@classmethod
- def from_path(cls, path) -> "ConfigFile":
+ def from_path(cls, path: str) -> "ConfigFile":
"""Read configuration from a file on disk."""
with GitFile(path, "rb") as f:
ret = cls.from_file(f)
ret.path = path
return ret
- def write_to_path(self, path=None) -> None:
+ def write_to_path(self, path: Optional[str] = None) -> None:
"""Write configuration to a file on disk."""
if path is None:
path = self.path
else:
f.write(b"[" + section_name + b' "' + subsection_name + b'"]\n')
for key, value in values.items():
- if value is True:
- value = b"true"
- elif value is False:
- value = b"false"
- else:
- value = _format_string(value)
+ value = _format_string(value)
f.write(b"\t" + key + b" = " + value + b"\n")
class StackedConfig(Config):
"""Configuration which reads from multiple config files.."""
- def __init__(self, backends, writable=None):
+ def __init__(
+ self, backends: List[ConfigFile], writable: Optional[ConfigFile] = None
+ ):
self.backends = backends
self.writable = writable
- def __repr__(self):
+ def __repr__(self) -> str:
return "<%s for %r>" % (self.__class__.__name__, self.backends)
@classmethod
- def default(cls):
+ def default(cls) -> "StackedConfig":
return cls(cls.default_backends())
@classmethod
- def default_backends(cls):
+ def default_backends(cls) -> List[ConfigFile]:
"""Retrieve the default configuration.
See git-config(1) for details on the files searched.
backends.append(cf)
return backends
- def get(self, section, name):
+ def get(self, section: SectionLike, name: NameLike) -> Value:
if not isinstance(section, tuple):
section = (section,)
for backend in self.backends:
pass
raise KeyError(name)
- def set(self, section, name, value):
+ def get_multivar(self, section: SectionLike, name: NameLike) -> Iterator[Value]:
+ if not isinstance(section, tuple):
+ section = (section,)
+ for backend in self.backends:
+ try:
+ yield from backend.get_multivar(section, name)
+ except KeyError:
+ pass
+
+ def set(
+ self,
+ section: SectionLike,
+ name: NameLike,
+ value: Union[ValueLike, bool]
+ ) -> None:
if self.writable is None:
raise NotImplementedError(self.set)
return self.writable.set(section, name, value)
+ def sections(self) -> Iterator[Section]:
+ seen = set()
+ for backend in self.backends:
+ for section in backend.sections():
+ if section not in seen:
+ seen.add(section)
+ yield section
+
def parse_submodules(config: ConfigFile) -> Iterator[Tuple[bytes, bytes, bytes]]:
"""Parse a gitmodules GitConfig file, returning submodules.
section_kind, section_name = section
if section_kind == b"submodule":
sm_path = config.get(section, b"path")
- assert isinstance(sm_path, bytes)
- assert sm_path is not None
sm_url = config.get(section, b"url")
- assert sm_url is not None
- assert isinstance(sm_url, bytes)
yield (sm_path, sm_url, section_name)
blob - 55a821f46c14bf7e8392d1d7d4ca85dd77cf3551
blob + 1831fe7d0266f3e7414db8ddae86af302fa42e1b
--- dulwich/ignore.py
+++ dulwich/ignore.py
Path to a global ignore file
"""
try:
- return config.get((b"core",), b"excludesFile")
+ value = config.get((b"core",), b"excludesFile")
+ assert isinstance(value, bytes)
+ return value.decode(encoding="utf-8")
except KeyError:
pass
blob - 34e8dda184436f34544f1badd646cde24bcf7851
blob + d34dac7e8a6f3a6113e2b30e97f9f4f27dbe1a1a
--- dulwich/porcelain.py
+++ dulwich/porcelain.py
if config.has_section(section):
remote_name = encoded_location.decode()
- url = config.get(section, "url")
- assert url is not None
- assert isinstance(url, bytes)
- encoded_location = url
+ encoded_location = config.get(section, "url")
else:
remote_name = None
path, r, progress=errstream.write, determine_wants=determine_wants
)
for (lh, rh, force_ref) in selected_refs:
- try:
- check_diverged(r, r.refs[rh], fetch_result.refs[lh])
- except DivergedBranches:
- if fast_forward:
- raise
- else:
- raise NotImplementedError("merge is not yet supported")
+ if not force_ref and rh in r.refs:
+ try:
+ check_diverged(r, r.refs.follow(rh)[1], fetch_result.refs[lh])
+ except DivergedBranches:
+ if fast_forward:
+ raise
+ else:
+ raise NotImplementedError("merge is not yet supported")
r.refs[rh] = fetch_result.refs[lh]
if selected_refs:
r[b"HEAD"] = fetch_result.refs[selected_refs[0][1]]
blob - 1fd839da6664491e553d2413929d81faeadb24d2
blob + 1309ecf8cfb5dd6c274538d8b5c94b1fb54b04b2
--- dulwich/tests/__init__.py
+++ dulwich/tests/__init__.py
"""Tests for Dulwich."""
+__all__ = [
+ 'SkipTest',
+ 'TestCase',
+ 'BlackboxTestCase',
+ 'skipIf',
+ 'expectedFailure',
+]
+
import doctest
import os
import shutil
blob - 12f62d15e3ef2f408dc60a906e06e474720ff45a
blob + d8c00b557f47b03af641d3e2624c4142616b8c12
--- dulwich/tests/test_client.py
+++ dulwich/tests/test_client.py
PLinkSSHVendor,
HangupException,
GitProtocolError,
+ apply_instead_of,
check_wants,
default_urllib3_manager,
get_credentials_from_store,
auth_string = "%s:%s" % ("user", "passwd")
b64_credentials = base64.b64encode(auth_string.encode("latin1"))
expected_basic_auth = "Basic %s" % b64_credentials.decode("latin1")
+ self.assertEqual(basic_auth, expected_basic_auth)
+
+ def test_init_username_set_no_password(self):
+ url = "https://github.com/jelmer/dulwich"
+
+ c = HttpGitClient(url, config=None, username="user")
+ self.assertEqual("user", c._username)
+ self.assertIs(c._password, None)
+
+ basic_auth = c.pool_manager.headers["authorization"]
+ auth_string = b"user:"
+ b64_credentials = base64.b64encode(auth_string)
+ expected_basic_auth = f"Basic {b64_credentials.decode('ascii')}"
self.assertEqual(basic_auth, expected_basic_auth)
def test_init_no_username_passwd(self):
self.assertIs(None, c._username)
self.assertIs(None, c._password)
self.assertNotIn("authorization", c.pool_manager.headers)
+
+ def test_from_parsedurl_username_only(self):
+ username = "user"
+ url = f"https://{username}@github.com/jelmer/dulwich"
+
+ c = HttpGitClient.from_parsedurl(urlparse(url))
+ self.assertEqual(c._username, username)
+ self.assertEqual(c._password, None)
+
+ basic_auth = c.pool_manager.headers["authorization"]
+ auth_string = username.encode('ascii') + b":"
+ b64_credentials = base64.b64encode(auth_string)
+ expected_basic_auth = f"Basic {b64_credentials.decode('ascii')}"
+ self.assertEqual(basic_auth, expected_basic_auth)
def test_from_parsedurl_on_url_with_quoted_credentials(self):
original_username = "john|the|first"
]
),
)
+
+
+class ApplyInsteadOfTests(TestCase):
+ def test_none(self):
+ config = ConfigDict()
+ self.assertEqual(
+ 'https://example.com/', apply_instead_of(config, 'https://example.com/'))
+
+ def test_apply(self):
+ config = ConfigDict()
+ config.set(
+ ('url', 'https://samba.org/'), 'insteadOf', 'https://example.com/')
+ self.assertEqual(
+ 'https://samba.org/',
+ apply_instead_of(config, 'https://example.com/'))
+
+ def test_apply_multiple(self):
+ config = ConfigDict()
+ config.set(
+ ('url', 'https://samba.org/'), 'insteadOf', 'https://blah.com/')
+ config.set(
+ ('url', 'https://samba.org/'), 'insteadOf', 'https://example.com/')
+ self.assertEqual(
+ [b'https://blah.com/', b'https://example.com/'],
+ list(config.get_multivar(('url', 'https://samba.org/'), 'insteadOf')))
+ self.assertEqual(
+ 'https://samba.org/',
+ apply_instead_of(config, 'https://example.com/'))
blob - 201669ebff9d296ddd2ea89404ca02dfa5debb6e
blob + 6ed9cac2c3b9e595b3eba070fd0ccef979c56148
--- dulwich/tests/test_porcelain.py
+++ dulwich/tests/test_porcelain.py
"""Tests for dulwich.porcelain."""
+import contextlib
from io import BytesIO, StringIO
import os
import platform
import sys
import tarfile
import tempfile
+import threading
import time
from unittest import skipIf
NoIndexPresent,
Repo,
)
+from dulwich.server import (
+ DictBackend,
+)
from dulwich.tests import (
TestCase,
)
build_commit_graph,
make_commit,
make_object,
+)
+from dulwich.web import (
+ make_server,
+ make_wsgi_chain,
)
self.assertEqual(
c1.id.decode('ascii')[:7],
porcelain.find_unique_abbrev(self.repo.object_store, c1.id))
+
+
+class ServerTests(PorcelainTestCase):
+ @contextlib.contextmanager
+ def _serving(self):
+ with make_server('localhost', 0, self.app) as server:
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ thread.start()
+
+ try:
+ yield f"http://localhost:{server.server_port}"
+
+ finally:
+ server.shutdown()
+ thread.join(10)
+
+ def setUp(self):
+ super().setUp()
+
+ self.served_repo_path = os.path.join(self.test_dir, "served_repo.git")
+ self.served_repo = Repo.init_bare(self.served_repo_path, mkdir=True)
+ self.addCleanup(self.served_repo.close)
+
+ backend = DictBackend({"/": self.served_repo})
+ self.app = make_wsgi_chain(backend)
+
+ def test_pull(self):
+ c1, = build_commit_graph(self.served_repo.object_store, [[1]])
+ self.served_repo.refs[b"refs/heads/master"] = c1.id
+
+ with self._serving() as url:
+ porcelain.pull(self.repo, url, "master")
+
+ def test_push(self):
+ c1, = build_commit_graph(self.repo.object_store, [[1]])
+ self.repo.refs[b"refs/heads/master"] = c1.id
+
+ with self._serving() as url:
+ porcelain.push(self.repo, url, "master")
blob - 27ac8577ad6eb73a7fb204437a432ad9c37402e9
blob + daa728944f853c304a7de8e628b34f70b1c6f4fa
--- dulwich/web.py
+++ dulwich/web.py
req.respond(HTTP_OK, "text/plain")
logger.info("Emulating dumb info/packs")
return generate_objects_info_packs(get_repo(backend, mat))
+
+
+def _chunk_iter(f):
+ while True:
+ line = f.readline()
+ length = int(line.rstrip(), 16)
+ chunk = f.read(length + 2)
+ if length == 0:
+ break
+ yield chunk[:-2]
+
+
+class ChunkReader(object):
+
+ def __init__(self, f):
+ self._iter = _chunk_iter(f)
+ self._buffer = []
+
+ def read(self, n):
+ while sum(map(len, self._buffer)) < n:
+ try:
+ self._buffer.append(next(self._iter))
+ except StopIteration:
+ break
+ f = b''.join(self._buffer)
+ ret = f[:n]
+ self._buffer = [f[n:]]
+ return ret
class _LengthLimitedFile(object):
return
req.nocache()
write = req.respond(HTTP_OK, "application/x-%s-result" % service)
- proto = ReceivableProtocol(req.environ["wsgi.input"].read, write)
+ if req.environ.get('HTTP_TRANSFER_ENCODING') == 'chunked':
+ read = ChunkReader(req.environ["wsgi.input"]).read
+ else:
+ read = req.environ["wsgi.input"].read
+ proto = ReceivableProtocol(read, write)
# TODO(jelmer): Find a way to pass in repo, rather than having handler_cls
# reopen.
handler = handler_cls(backend, [url_prefix(mat)], proto, stateless_rpc=req)
blob - 5daca8bf089b001f523873510c0fc3ee5bffe1ac
blob + 3de0b5b9f6f458c21bdbe15bb6a586b4bbcc5588
--- dulwich.egg-info/PKG-INFO
+++ dulwich.egg-info/PKG-INFO
Metadata-Version: 2.1
Name: dulwich
-Version: 0.20.43
+Version: 0.20.44
Summary: Python Git Library
Home-page: https://www.dulwich.io/
Author: Jelmer Vernooij
blob - a2983826a60d21673c6e8b02788cf1c6cb4a5842
blob + 3c557ab7db8b44fb11b1ad1d677fee8ec3467d63
--- setup.py
+++ setup.py
'For 2.7 support, please install a version prior to 0.20')
-dulwich_version_string = '0.20.43'
+dulwich_version_string = '0.20.44'
class DulwichDistribution(Distribution):
