Coverage for src/lpip/validation.py: 88%

1"""Catalog and config schema helpers (no I/O).""" 

2 

3from __future__ import annotations 

4 

5import re 

6from collections.abc import Mapping 

7from pathlib import Path, PurePosixPath 

8from typing import Any 

9from urllib.parse import urlparse 

10 

11PACKAGE_FIELD_KEYS = frozenset({"url", "revision", "subdirectory"}) 

12CATALOG_TOP_KEYS = frozenset({"packages"}) 

13CONFIG_TOP_KEYS = frozenset({"source", "sources", "policy", "runtime"}) 

14SOURCE_KEYS = frozenset({"type", "url", "revision", "path", "refresh"}) 

15SOURCE_ITEM_KEYS = SOURCE_KEYS | {"name"} 

16SOURCE_NAME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_-]*$") 

17POLICY_KEYS = frozenset({"allowed_schemes", "require_pinned_revision"}) 

18RUNTIME_KEYS = frozenset({"python"}) 

19 

20 

21def first_unknown_key(data: Mapping[str, Any], allowed: frozenset[str]) -> str | None: 

22 extra = set(data) - allowed 

23 if not extra: 

24 return None 

25 return sorted(extra)[0] 

26 

27 

28def optional_str(value: object) -> str | None: 

29 """Treat missing or blank strings as omitted (ADR-0010 empty revision).""" 

30 if value is None: 

31 return None 

32 if not isinstance(value, str): 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 msg = f"Expected a string, got {type(value).__name__}" 

34 raise TypeError(msg) 

35 stripped = value.strip() 

36 return stripped or None 

37 

38 

39def subdirectory_escapes_repo(subdirectory: str) -> bool: 

40 """True if subdirectory is absolute or walks above the repo root.""" 

41 if not subdirectory or subdirectory.strip() != subdirectory: 41 ↛ 43line 41 didn't jump to line 43 because the condition on line 41 was never true

42 # leading/trailing space is still the path; strip for check only 

43 subdirectory = subdirectory.strip() 

44 posix = PurePosixPath(subdirectory.replace("\\", "/")) 

45 if posix.is_absolute() or posix.anchor: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return True 

47 depth = 0 

48 for part in posix.parts: 

49 if part in ("", "."): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 continue 

51 if part == "..": 

52 depth -= 1 

53 if depth < 0: 53 ↛ 48line 53 didn't jump to line 48 because the condition on line 53 was always true

54 return True 

55 else: 

56 depth += 1 

57 return False 

58 

59 

60def url_scheme(url: str) -> str: 

61 return urlparse(url).scheme.lower() 

62 

63 

64_AUTHORING_TO_STORED = { 

65 "ssh": "git+ssh", 

66 "https": "git+https", 

67 "git+ssh": "git+ssh", 

68 "git+https": "git+https", 

69 "git+file": "git+file", 

70} 

71_SCP_STYLE = re.compile(r"^(?P<user>[^@\s]+)@(?P<host>[^:/\s]+):(?P<path>.+)$") 

72 

73 

74def _scp_to_git_ssh(match: re.Match[str]) -> str: 

75 path = match.group("path").replace("\\", "/").lstrip("/") 

76 return f"git+ssh://{match.group('user')}@{match.group('host')}/{path}" 

77 

78 

79def normalize_authoring_git_url(url: str) -> str: 

80 """Clone-style URL → pip direct URL stored in the catalog (ADR-0015).""" 

81 cleaned = url.strip() 

82 if "://" not in cleaned: 

83 scp = _SCP_STYLE.match(cleaned) 

84 if scp is not None: 84 ↛ 86line 84 didn't jump to line 86 because the condition on line 84 was always true

85 return _scp_to_git_ssh(scp) 

86 scheme = url_scheme(cleaned) 

87 stored = _AUTHORING_TO_STORED.get(scheme) 

88 if stored is None: 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true

89 return cleaned 

90 if scheme == "git+file": 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true

91 return normalize_git_file_url(cleaned) 

92 if scheme in {"git+ssh", "git+https"}: 

93 return cleaned.replace("\\", "/") 

94 sep = cleaned.find("://") 

95 rest = cleaned[sep:] if sep != -1 else f"://{cleaned}" 

96 return f"{stored}{rest}".replace("\\", "/") 

97 

98 

99def toml_posix_path(path: Path | str) -> str: 

100 """Single TOML path spelling: POSIX separators (``D:/foo`` or ``/home/foo``).""" 

101 return str(path).replace("\\", "/") 

102 

103 

104_LOCALHOST_PREFIX = "localhost/" 

105 

106 

107def normalize_git_file_url(url: str) -> str: 

108 """Canonical ``git+file`` URL for pip (POSIX separators, required netloc). 

109 

110 Windows drive letters use ``git+file://D:/repo`` (pip rejects 

111 ``git+file:///D:/repo``). POSIX paths use ``git+file://localhost/abs/path`` 

112 because ``packaging`` rejects ``git+file`` URLs with an empty host. 

113 """ 

114 cleaned = url.replace("\\", "/") 

115 if not cleaned.lower().startswith("git+file:"): 

116 return cleaned 

117 path = cleaned.split(":", 1)[1].lstrip("/") 

118 if len(path) >= 2 and path[1] == ":": 

119 return f"git+file://{path}" 

120 if path.lower().startswith(_LOCALHOST_PREFIX): 

121 path = path[len(_LOCALHOST_PREFIX) :] 

122 return f"git+file://localhost/{path}" 

123 

124 

125def to_git_remote_url(url: str) -> str: 

126 """Translate a catalog/package VCS URL into a URL ``git clone`` accepts.""" 

127 cleaned = normalize_git_file_url(url) 

128 if cleaned.lower().startswith("git+file:"): 

129 remainder = cleaned.split(":", 1)[1] 

130 if remainder.lower().startswith("//localhost/"): 

131 return "file:///" + remainder[len("//localhost/") :] 

132 return "file:" + remainder 

133 if cleaned.lower().startswith("git+"): 

134 return cleaned[4:] 

135 return cleaned