mirror of https://github.com/pypa/pip
Add DirectUrl model, implementing PEP 610
This commit is contained in:
Stéphane Bidoul
parent
657cf2515b
commit
6b7f4ce81b
|
|
@ -0,0 +1,245 @@
|
|||
""" PEP 610 """
|
||||
import json
|
||||
import re
|
||||
|
||||
from pip._vendor import six
|
||||
from pip._vendor.six.moves.urllib import parse as urllib_parse
|
||||
|
||||
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
|
||||
|
||||
if MYPY_CHECK_RUNNING:
|
||||
from typing import (
|
||||
Any, Dict, Iterable, Optional, Type, TypeVar, Union
|
||||
)
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
DIRECT_URL_METADATA_NAME = "direct_url.json"
|
||||
ENV_VAR_RE = re.compile(r"^\$\{[A-Za-z0-9-_]+\}(:\$\{[A-Za-z0-9-_]+\})?$")
|
||||
|
||||
__all__ = [
|
||||
"DirectUrl",
|
||||
"DirectUrlValidationError",
|
||||
"DirInfo",
|
||||
"ArchiveInfo",
|
||||
"VcsInfo",
|
||||
]
|
||||
|
||||
|
||||
class DirectUrlValidationError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def _get(d, expected_type, key, default=None):
|
||||
# type: (Dict[str, Any], Type[T], str, Optional[T]) -> Optional[T]
|
||||
"""Get value from dictionary and verify expected type."""
|
||||
if key not in d:
|
||||
return default
|
||||
value = d[key]
|
||||
if six.PY2 and expected_type is str:
|
||||
expected_type = six.string_types # type: ignore
|
||||
if not isinstance(value, expected_type):
|
||||
raise DirectUrlValidationError(
|
||||
"{!r} has unexpected type for {} (expected {})".format(
|
||||
value, key, expected_type
|
||||
)
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def _get_required(d, expected_type, key, default=None):
|
||||
# type: (Dict[str, Any], Type[T], str, Optional[T]) -> T
|
||||
value = _get(d, expected_type, key, default)
|
||||
if value is None:
|
||||
raise DirectUrlValidationError("{} must have a value".format(key))
|
||||
return value
|
||||
|
||||
|
||||
def _exactly_one_of(infos):
|
||||
# type: (Iterable[Optional[InfoType]]) -> InfoType
|
||||
infos = [info for info in infos if info is not None]
|
||||
if not infos:
|
||||
raise DirectUrlValidationError(
|
||||
"missing one of archive_info, dir_info, vcs_info"
|
||||
)
|
||||
if len(infos) > 1:
|
||||
raise DirectUrlValidationError(
|
||||
"more than one of archive_info, dir_info, vcs_info"
|
||||
)
|
||||
assert infos[0] is not None
|
||||
return infos[0]
|
||||
|
||||
|
||||
def _filter_none(**kwargs):
|
||||
# type: (Any) -> Dict[str, Any]
|
||||
"""Make dict excluding None values."""
|
||||
return {k: v for k, v in kwargs.items() if v is not None}
|
||||
|
||||
|
||||
class VcsInfo(object):
|
||||
name = "vcs_info"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
vcs, # type: str
|
||||
commit_id, # type: str
|
||||
requested_revision=None, # type: Optional[str]
|
||||
resolved_revision=None, # type: Optional[str]
|
||||
resolved_revision_type=None, # type: Optional[str]
|
||||
):
|
||||
self.vcs = vcs
|
||||
self.requested_revision = requested_revision
|
||||
self.commit_id = commit_id
|
||||
self.resolved_revision = resolved_revision
|
||||
self.resolved_revision_type = resolved_revision_type
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d):
|
||||
# type: (Optional[Dict[str, Any]]) -> Optional[VcsInfo]
|
||||
if d is None:
|
||||
return None
|
||||
return cls(
|
||||
vcs=_get_required(d, str, "vcs"),
|
||||
commit_id=_get_required(d, str, "commit_id"),
|
||||
requested_revision=_get(d, str, "requested_revision"),
|
||||
resolved_revision=_get(d, str, "resolved_revision"),
|
||||
resolved_revision_type=_get(d, str, "resolved_revision_type"),
|
||||
)
|
||||
|
||||
def _to_dict(self):
|
||||
# type: () -> Dict[str, Any]
|
||||
return _filter_none(
|
||||
vcs=self.vcs,
|
||||
requested_revision=self.requested_revision,
|
||||
commit_id=self.commit_id,
|
||||
resolved_revision=self.resolved_revision,
|
||||
resolved_revision_type=self.resolved_revision_type,
|
||||
)
|
||||
|
||||
|
||||
class ArchiveInfo(object):
|
||||
name = "archive_info"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hash=None, # type: Optional[str]
|
||||
):
|
||||
self.hash = hash
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d):
|
||||
# type: (Optional[Dict[str, Any]]) -> Optional[ArchiveInfo]
|
||||
if d is None:
|
||||
return None
|
||||
return cls(hash=_get(d, str, "hash"))
|
||||
|
||||
def _to_dict(self):
|
||||
# type: () -> Dict[str, Any]
|
||||
return _filter_none(hash=self.hash)
|
||||
|
||||
|
||||
class DirInfo(object):
|
||||
name = "dir_info"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
editable=False, # type: bool
|
||||
):
|
||||
self.editable = editable
|
||||
|
||||
@classmethod
|
||||
def _from_dict(cls, d):
|
||||
# type: (Optional[Dict[str, Any]]) -> Optional[DirInfo]
|
||||
if d is None:
|
||||
return None
|
||||
return cls(
|
||||
editable=_get_required(d, bool, "editable", default=False)
|
||||
)
|
||||
|
||||
def _to_dict(self):
|
||||
# type: () -> Dict[str, Any]
|
||||
return _filter_none(editable=self.editable or None)
|
||||
|
||||
|
||||
if MYPY_CHECK_RUNNING:
|
||||
InfoType = Union[ArchiveInfo, DirInfo, VcsInfo]
|
||||
|
||||
|
||||
class DirectUrl(object):
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
url, # type: str
|
||||
info, # type: InfoType
|
||||
subdirectory=None, # type: Optional[str]
|
||||
):
|
||||
self.url = url
|
||||
self.info = info
|
||||
self.subdirectory = subdirectory
|
||||
|
||||
def _remove_auth_from_netloc(self, netloc):
|
||||
# type: (str) -> str
|
||||
if "@" not in netloc:
|
||||
return netloc
|
||||
user_pass, netloc_no_user_pass = netloc.split("@", 1)
|
||||
if (
|
||||
isinstance(self.info, VcsInfo) and
|
||||
self.info.vcs == "git" and
|
||||
user_pass == "git"
|
||||
):
|
||||
return netloc
|
||||
if ENV_VAR_RE.match(user_pass):
|
||||
return netloc
|
||||
return netloc_no_user_pass
|
||||
|
||||
@property
|
||||
def redacted_url(self):
|
||||
# type: () -> str
|
||||
"""url with user:password part removed unless it is formed with
|
||||
environment variables as specified in PEP 610, or it is ``git``
|
||||
in the case of a git URL.
|
||||
"""
|
||||
purl = urllib_parse.urlsplit(self.url)
|
||||
netloc = self._remove_auth_from_netloc(purl.netloc)
|
||||
surl = urllib_parse.urlunsplit(
|
||||
(purl.scheme, netloc, purl.path, purl.query, purl.fragment)
|
||||
)
|
||||
return surl
|
||||
|
||||
def validate(self):
|
||||
# type: () -> None
|
||||
self.from_dict(self.to_dict())
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, d):
|
||||
# type: (Dict[str, Any]) -> DirectUrl
|
||||
return DirectUrl(
|
||||
url=_get_required(d, str, "url"),
|
||||
subdirectory=_get(d, str, "subdirectory"),
|
||||
info=_exactly_one_of(
|
||||
[
|
||||
ArchiveInfo._from_dict(_get(d, dict, "archive_info")),
|
||||
DirInfo._from_dict(_get(d, dict, "dir_info")),
|
||||
VcsInfo._from_dict(_get(d, dict, "vcs_info")),
|
||||
]
|
||||
),
|
||||
)
|
||||
|
||||
def to_dict(self):
|
||||
# type: () -> Dict[str, Any]
|
||||
res = _filter_none(
|
||||
url=self.redacted_url,
|
||||
subdirectory=self.subdirectory,
|
||||
)
|
||||
res[self.info.name] = self.info._to_dict()
|
||||
return res
|
||||
|
||||
@classmethod
|
||||
def from_json(cls, s):
|
||||
# type: (str) -> DirectUrl
|
||||
return cls.from_dict(json.loads(s))
|
||||
|
||||
def to_json(self):
|
||||
# type: () -> str
|
||||
return json.dumps(self.to_dict(), sort_keys=True)
|
||||
|
|
@ -0,0 +1,151 @@
|
|||
import pytest
|
||||
|
||||
from pip._internal.models.direct_url import (
|
||||
ArchiveInfo,
|
||||
DirectUrl,
|
||||
DirectUrlValidationError,
|
||||
DirInfo,
|
||||
VcsInfo,
|
||||
)
|
||||
|
||||
|
||||
def test_from_json():
|
||||
json = '{"url": "file:///home/user/project", "dir_info": {}}'
|
||||
direct_url = DirectUrl.from_json(json)
|
||||
assert direct_url.url == "file:///home/user/project"
|
||||
assert direct_url.info.editable is False
|
||||
|
||||
|
||||
def test_to_json():
|
||||
direct_url = DirectUrl(
|
||||
url="file:///home/user/archive.tgz",
|
||||
info=ArchiveInfo(),
|
||||
)
|
||||
direct_url.validate()
|
||||
assert direct_url.to_json() == (
|
||||
'{"archive_info": {}, "url": "file:///home/user/archive.tgz"}'
|
||||
)
|
||||
|
||||
|
||||
def test_archive_info():
|
||||
direct_url_dict = {
|
||||
"url": "file:///home/user/archive.tgz",
|
||||
"archive_info": {
|
||||
"hash": "sha1=1b8c5bc61a86f377fea47b4276c8c8a5842d2220"
|
||||
},
|
||||
}
|
||||
direct_url = DirectUrl.from_dict(direct_url_dict)
|
||||
assert isinstance(direct_url.info, ArchiveInfo)
|
||||
assert direct_url.url == direct_url_dict["url"]
|
||||
assert direct_url.info.hash == direct_url_dict["archive_info"]["hash"]
|
||||
assert direct_url.to_dict() == direct_url_dict
|
||||
|
||||
|
||||
def test_dir_info():
|
||||
direct_url_dict = {
|
||||
"url": "file:///home/user/project",
|
||||
"dir_info": {"editable": True},
|
||||
}
|
||||
direct_url = DirectUrl.from_dict(direct_url_dict)
|
||||
assert isinstance(direct_url.info, DirInfo)
|
||||
assert direct_url.url == direct_url_dict["url"]
|
||||
assert direct_url.info.editable is True
|
||||
assert direct_url.to_dict() == direct_url_dict
|
||||
# test editable default to False
|
||||
direct_url_dict = {"url": "file:///home/user/project", "dir_info": {}}
|
||||
direct_url = DirectUrl.from_dict(direct_url_dict)
|
||||
assert direct_url.info.editable is False
|
||||
|
||||
|
||||
def test_vcs_info():
|
||||
direct_url_dict = {
|
||||
"url": "https:///g.c/u/p.git",
|
||||
"vcs_info": {
|
||||
"vcs": "git",
|
||||
"requested_revision": "master",
|
||||
"commit_id": "1b8c5bc61a86f377fea47b4276c8c8a5842d2220",
|
||||
},
|
||||
}
|
||||
direct_url = DirectUrl.from_dict(direct_url_dict)
|
||||
assert isinstance(direct_url.info, VcsInfo)
|
||||
assert direct_url.url == direct_url_dict["url"]
|
||||
assert direct_url.info.vcs == "git"
|
||||
assert direct_url.info.requested_revision == "master"
|
||||
assert (
|
||||
direct_url.info.commit_id == "1b8c5bc61a86f377fea47b4276c8c8a5842d2220"
|
||||
)
|
||||
assert direct_url.to_dict() == direct_url_dict
|
||||
|
||||
|
||||
def test_parsing_validation():
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError, match="url must have a value"
|
||||
):
|
||||
DirectUrl.from_dict({"dir_info": {}})
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError,
|
||||
match="missing one of archive_info, dir_info, vcs_info",
|
||||
):
|
||||
DirectUrl.from_dict({"url": "http://..."})
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError, match="unexpected type for editable"
|
||||
):
|
||||
DirectUrl.from_dict(
|
||||
{"url": "http://...", "dir_info": {"editable": "false"}}
|
||||
)
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError, match="unexpected type for hash"
|
||||
):
|
||||
DirectUrl.from_dict({"url": "http://...", "archive_info": {"hash": 1}})
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError, match="unexpected type for vcs"
|
||||
):
|
||||
DirectUrl.from_dict({"url": "http://...", "vcs_info": {"vcs": None}})
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError, match="commit_id must have a value"
|
||||
):
|
||||
DirectUrl.from_dict({"url": "http://...", "vcs_info": {"vcs": "git"}})
|
||||
with pytest.raises(
|
||||
DirectUrlValidationError,
|
||||
match="more than one of archive_info, dir_info, vcs_info",
|
||||
):
|
||||
DirectUrl.from_dict(
|
||||
{"url": "http://...", "dir_info": {}, "archive_info": {}}
|
||||
)
|
||||
|
||||
|
||||
def test_redact_url():
|
||||
def _redact_git(url):
|
||||
direct_url = DirectUrl(
|
||||
url=url,
|
||||
info=VcsInfo(vcs="git", commit_id="1"),
|
||||
)
|
||||
return direct_url.redacted_url
|
||||
|
||||
def _redact_archive(url):
|
||||
direct_url = DirectUrl(
|
||||
url=url,
|
||||
info=ArchiveInfo(),
|
||||
)
|
||||
return direct_url.redacted_url
|
||||
|
||||
assert (
|
||||
_redact_git("https://user:password@g.c/u/p.git@branch#egg=pkg") ==
|
||||
"https://g.c/u/p.git@branch#egg=pkg"
|
||||
)
|
||||
assert (
|
||||
_redact_git("https://${USER}:password@g.c/u/p.git") ==
|
||||
"https://g.c/u/p.git"
|
||||
)
|
||||
assert (
|
||||
_redact_archive("file://${U}:${PIP_PASSWORD}@g.c/u/p.tgz") ==
|
||||
"file://${U}:${PIP_PASSWORD}@g.c/u/p.tgz"
|
||||
)
|
||||
assert (
|
||||
_redact_git("https://${PIP_TOKEN}@g.c/u/p.git") ==
|
||||
"https://${PIP_TOKEN}@g.c/u/p.git"
|
||||
)
|
||||
assert (
|
||||
_redact_git("ssh://git@g.c/u/p.git") ==
|
||||
"ssh://git@g.c/u/p.git"
|
||||
)
|
||||
Loading…
Reference in New Issue