Coverage for src / updates2mqtt / integrations / git_utils.py: 81%

1import datetime 

2import re 

3import subprocess 

4from pathlib import Path 

5from re import Match 

6 

7import structlog 

8 

9log = structlog.get_logger() 

10 

11 

12def git_trust(repo_path: Path, git_path: Path) -> bool: 

13 try: 

14 subprocess.run(f"{git_path} config --global --add safe.directory {repo_path}", check=True, shell=True, cwd=repo_path) 

15 return True 

16 except Exception as e: 

17 log.warn("GIT Unable to trust repo at %s: %s", repo_path, e, action="git_trust") 

18 return False 

19 

20 

21def git_iso_timestamp(repo_path: Path, git_path: Path) -> str | None: 

22 result = None 

23 try: 

24 result = subprocess.run( 

25 str(git_path) + r" log -1 --format=%cI --no-show-signature", 

26 cwd=repo_path, 

27 shell=True, 

28 text=True, 

29 capture_output=True, 

30 check=True, 

31 ) 

32 # round-trip the iso format for pythony consistency 

33 return datetime.datetime.fromisoformat(result.stdout.strip()).isoformat() 

34 except subprocess.CalledProcessError as cpe: 

35 log.warn("GIT No result from git log at %s: %s", repo_path, cpe, action="git_iso_timestamp") 

36 except Exception as e: 

37 log.error( 

38 "GIT Unable to parse timestamp at %s - %s: %s", 

39 repo_path, 

40 result.stdout if result else "<NO RESULT>", 

41 e, 

42 action="git_iso_timestamp", 

43 ) 

44 return None 

45 

46 

47def git_local_version(repo_path: Path, git_path: Path) -> str | None: 

48 result = None 

49 try: 

50 result = subprocess.run( 

51 f"{git_path} rev-parse HEAD", 

52 cwd=repo_path, 

53 shell=True, 

54 text=True, 

55 capture_output=True, 

56 check=True, 

57 ) 

58 if result.returncode == 0: 58 ↛ 71line 58 didn't jump to line 71 because the condition on line 58 was always true

59 log.debug("Local git rev-parse", action="git_local_version", path=repo_path, version=result.stdout.strip()) 

60 return f"git:{result.stdout.strip()}"[:19] 

61 except subprocess.CalledProcessError as cpe: 

62 log.warn("GIT No result from git rev-parse at %s: %s", repo_path, cpe, action="git_local_version") 

63 except Exception as e: 

64 log.error( 

65 "GIT Unable to retrieve version at %s - %s: %s", 

66 repo_path, 

67 result.stdout if result else "<NO RESULT>", 

68 e, 

69 action="git_local_version", 

70 ) 

71 return None 

72 

73 

74def git_check_update_available(repo_path: Path, git_path: Path, timeout: int = 120) -> int: 

75 result = None 

76 try: 

77 # check if remote repo ahead 

78 result = subprocess.run( 

79 f"{git_path} fetch;{git_path} status -uno", 

80 capture_output=True, 

81 text=True, 

82 shell=True, 

83 check=True, 

84 cwd=repo_path, 

85 timeout=timeout, 

86 ) 

87 if result.returncode == 0: 87 ↛ 103line 87 didn't jump to line 103 because the condition on line 87 was always true

88 count_match: Match[str] | None = re.search( 

89 r"Your branch is behind.*by (\d+) commit", result.stdout, flags=re.MULTILINE 

90 ) 

91 if count_match and count_match.groups(): 

92 log.debug( 

93 "Local git repo update available: %s (%s)", 

94 count_match.group(1), 

95 result.stdout.strip(), 

96 action="git_check", 

97 path=repo_path, 

98 ) 

99 return int(count_match.group(1)) 

100 log.debug("Local git repo no update available", action="git_check", path=repo_path, status=result.stdout.strip()) 

101 return 0 

102 

103 log.debug( 

104 "No git update available", 

105 action="git_check", 

106 path=repo_path, 

107 returncode=result.returncode, 

108 stdout=result.stdout, 

109 stderr=result.stderr, 

110 ) 

111 except Exception as e: 

112 log.warn("GIT Unable to check status %s: %s", result.stdout if result else "<NO RESULT>", e, action="git_check") 

113 return 0 

114 

115 

116def git_pull(repo_path: Path, git_path: Path) -> bool: 

117 log.info("GIT Pulling git at %s", repo_path, action="git_pull") 

118 proc = subprocess.run(f"{git_path} pull", shell=True, check=False, cwd=repo_path, timeout=300) 

119 if proc.returncode == 0: 

120 log.info("GIT pull at %s successful", repo_path, action="git_pull") 

121 return True 

122 log.warn("GIT pull at %s failed: %s", repo_path, proc.returncode, action="git_pull", stdout=proc.stdout, stderr=proc.stderr) 

123 return False