Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_internal/vcs/mercurial.py: 40%

1import configparser 

2import logging 

3import os 

4from typing import List, Optional, Tuple 

5 

6from pip._internal.exceptions import BadCommand, InstallationError 

7from pip._internal.utils.misc import HiddenText, display_path 

8from pip._internal.utils.subprocess import make_command 

9from pip._internal.utils.urls import path_to_url 

10from pip._internal.vcs.versioncontrol import ( 

11 RevOptions, 

12 VersionControl, 

13 find_path_to_project_root_from_repo_root, 

14 vcs, 

15) 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class Mercurial(VersionControl): 

21 name = "hg" 

22 dirname = ".hg" 

23 repo_name = "clone" 

24 schemes = ( 

25 "hg+file", 

26 "hg+http", 

27 "hg+https", 

28 "hg+ssh", 

29 "hg+static-http", 

30 ) 

31 

32 @staticmethod 

33 def get_base_rev_args(rev: str) -> List[str]: 

34 return [rev] 

35 

36 def fetch_new( 

37 self, dest: str, url: HiddenText, rev_options: RevOptions, verbosity: int 

38 ) -> None: 

39 rev_display = rev_options.to_display() 

40 logger.info( 

41 "Cloning hg %s%s to %s", 

42 url, 

43 rev_display, 

44 display_path(dest), 

45 ) 

46 if verbosity <= 0: 

47 flags: Tuple[str, ...] = ("--quiet",) 

48 elif verbosity == 1: 

49 flags = () 

50 elif verbosity == 2: 

51 flags = ("--verbose",) 

52 else: 

53 flags = ("--verbose", "--debug") 

54 self.run_command(make_command("clone", "--noupdate", *flags, url, dest)) 

55 self.run_command( 

56 make_command("update", *flags, rev_options.to_args()), 

57 cwd=dest, 

58 ) 

59 

60 def switch(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None: 

61 repo_config = os.path.join(dest, self.dirname, "hgrc") 

62 config = configparser.RawConfigParser() 

63 try: 

64 config.read(repo_config) 

65 config.set("paths", "default", url.secret) 

66 with open(repo_config, "w") as config_file: 

67 config.write(config_file) 

68 except (OSError, configparser.NoSectionError) as exc: 

69 logger.warning("Could not switch Mercurial repository to %s: %s", url, exc) 

70 else: 

71 cmd_args = make_command("update", "-q", rev_options.to_args()) 

72 self.run_command(cmd_args, cwd=dest) 

73 

74 def update(self, dest: str, url: HiddenText, rev_options: RevOptions) -> None: 

75 self.run_command(["pull", "-q"], cwd=dest) 

76 cmd_args = make_command("update", "-q", rev_options.to_args()) 

77 self.run_command(cmd_args, cwd=dest) 

78 

79 @classmethod 

80 def get_remote_url(cls, location: str) -> str: 

81 url = cls.run_command( 

82 ["showconfig", "paths.default"], 

83 show_stdout=False, 

84 stdout_only=True, 

85 cwd=location, 

86 ).strip() 

87 if cls._is_local_repository(url): 

88 url = path_to_url(url) 

89 return url.strip() 

90 

91 @classmethod 

92 def get_revision(cls, location: str) -> str: 

93 """ 

94 Return the repository-local changeset revision number, as an integer. 

95 """ 

96 current_revision = cls.run_command( 

97 ["parents", "--template={rev}"], 

98 show_stdout=False, 

99 stdout_only=True, 

100 cwd=location, 

101 ).strip() 

102 return current_revision 

103 

104 @classmethod 

105 def get_requirement_revision(cls, location: str) -> str: 

106 """ 

107 Return the changeset identification hash, as a 40-character 

108 hexadecimal string 

109 """ 

110 current_rev_hash = cls.run_command( 

111 ["parents", "--template={node}"], 

112 show_stdout=False, 

113 stdout_only=True, 

114 cwd=location, 

115 ).strip() 

116 return current_rev_hash 

117 

118 @classmethod 

119 def is_commit_id_equal(cls, dest: str, name: Optional[str]) -> bool: 

120 """Always assume the versions don't match""" 

121 return False 

122 

123 @classmethod 

124 def get_subdirectory(cls, location: str) -> Optional[str]: 

125 """ 

126 Return the path to Python project root, relative to the repo root. 

127 Return None if the project root is in the repo root. 

128 """ 

129 # find the repo root 

130 repo_root = cls.run_command( 

131 ["root"], show_stdout=False, stdout_only=True, cwd=location 

132 ).strip() 

133 if not os.path.isabs(repo_root): 

134 repo_root = os.path.abspath(os.path.join(location, repo_root)) 

135 return find_path_to_project_root_from_repo_root(location, repo_root) 

136 

137 @classmethod 

138 def get_repository_root(cls, location: str) -> Optional[str]: 

139 loc = super().get_repository_root(location) 

140 if loc: 

141 return loc 

142 try: 

143 r = cls.run_command( 

144 ["root"], 

145 cwd=location, 

146 show_stdout=False, 

147 stdout_only=True, 

148 on_returncode="raise", 

149 log_failed_cmd=False, 

150 ) 

151 except BadCommand: 

152 logger.debug( 

153 "could not determine if %s is under hg control " 

154 "because hg is not available", 

155 location, 

156 ) 

157 return None 

158 except InstallationError: 

159 return None 

160 return os.path.normpath(r.rstrip("\r\n")) 

161 

162 

163vcs.register(Mercurial)