Coverage for collectra/gui/updates.py: 95.00%

1"""Check and fast-forward the two source repositories used by this GUI.""" 

2 

3import os 

4import subprocess 

5from pathlib import Path 

6 

7 

8def git(root, *args): 

9 env = dict(os.environ, GIT_TERMINAL_PROMPT="0", GCM_INTERACTIVE="never") 

10 env.setdefault("GIT_SSH_COMMAND", "ssh -oBatchMode=yes") 

11 result = subprocess.run( 

12 ["git", "-C", str(root), *args], capture_output=True, text=True, 

13 timeout=120, env=env, 

14 ) 

15 if result.returncode: 

16 raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "Git command failed") 

17 return result.stdout.strip() 

18 

19 

20def collectra_repository(): 

21 import collectra 

22 

23 source = Path(collectra.__file__).resolve() 

24 for root in source.parents: 

25 if (root / ".git").exists(): 

26 try: 

27 git(root, "ls-files", "--error-unmatch", "--", str(source.relative_to(root))) 

28 return root 

29 except (RuntimeError, OSError, subprocess.TimeoutExpired): 

30 return None 

31 return None 

32 

33 

34def inspect_repository(root, label, fetch=True): 

35 result = {"label": label, "path": str(root) if root else "", "can_update": False} 

36 if root is None or not (Path(root) / ".git").exists(): 

37 return {**result, "status": "unavailable", "message": "Not installed from a Git checkout."} 

38 root = Path(root) 

39 try: 

40 branch = git(root, "rev-parse", "--abbrev-ref", "HEAD") 

41 if branch == "HEAD": 

42 raise RuntimeError("This checkout has a detached HEAD. Switch to a tracking branch before updating.") 

43 remotes = git(root, "remote").splitlines() 

44 if not remotes: 

45 return {**result, "status": "unavailable", "message": "No Git remote configured."} 

46 remote = git(root, "for-each-ref", "--format=%(upstream:remotename)", f"refs/heads/{branch}") 

47 if remote not in remotes: 

48 raise RuntimeError("The current branch does not track a remote branch.") 

49 if fetch: 

50 git(root, "fetch", "--prune", "--", remote) 

51 upstream = git(root, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}") 

52 ahead, behind = map(int, git(root, "rev-list", "--left-right", "--count", "HEAD...@{upstream}").split()) 

53 dirty = bool(git(root, "status", "--porcelain")) 

54 result.update(branch=branch, upstream=upstream, ahead=ahead, behind=behind) 

55 if dirty: 

56 return {**result, "status": "blocked", "message": "Local changes are present. Commit or stash them before updating."} 

57 if ahead and behind: 

58 return {**result, "status": "blocked", "message": "Local and remote branches have diverged. Resolve this in Git before updating."} 

59 if behind: 

60 return {**result, "status": "available", "can_update": True, 

61 "message": f"{behind} new commit(s) available from {upstream}."} 

62 return {**result, "status": "current", "message": "Up to date." if not ahead else "No incoming updates; local commits are ahead of the remote."} 

63 except (RuntimeError, OSError, subprocess.TimeoutExpired) as error: 

64 return {**result, "status": "error", "message": str(error)} 

65 

66 

67def update_repository(root, label): 

68 # Recheck immediately before modifying a checkout: status may have changed 

69 # since the dialog opened. Never merge or reset local work. 

70 result = inspect_repository(root, label) 

71 if not result["can_update"]: 

72 return result 

73 try: 

74 git(root, "merge", "--ff-only", "@{upstream}") 

75 return {**result, "can_update": False, "status": "updated", 

76 "message": "Updated successfully. Restart the GUI to use the new code."} 

77 except (RuntimeError, OSError, subprocess.TimeoutExpired) as error: 

78 return {**result, "can_update": False, "status": "error", "message": str(error)}