commit - 55f151dd04897f8d20a8c7e3a5a4ff89d6f1cfc4
commit + bf03e2e836549cebc2bea9f9909649885ceffaa4
blob - ffb09bf2e31af7558226378fc160a1243da68c67
blob + db78a5d8d4be44d580f33820ffdfa580ce23efd5
--- NEWS
+++ NEWS
* 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 - 3e42df1110827c2d294370cb448591d31dc71c55
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,
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 + 30b67fb3173e2adf104c10ee6f23c8b697b8b6d4
--- dulwich/config.py
+++ dulwich/config.py
if len(section) > 1:
try:
- return self._values[section][name]
+ return self._values[section].get_all(name)
except KeyError:
pass
pass
raise KeyError(name)
+ def get_multivar(self, section, name):
+ 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, name, value):
if self.writable is None:
raise NotImplementedError(self.set)
return self.writable.set(section, name, value)
+ def sections(self):
+ 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.
blob - 767fe94a49c684c5c481bf515a8b19c4892dfe1c
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,
]
),
)
+
+
+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/'))
