Coverage for collectra/gui/source_control.py: 77.78%

1"""Run the GUI's fixed Git operations in the active results repository.""" 

2 

3import os 

4import subprocess 

5from pathlib import Path 

6 

7 

8def run_git( 

9 directory: Path, action: str, message: str = "", repository: str = "" 

10) -> dict: 

11 commands = { 

12 "status": ["status"], 

13 "pull": ["pull", "--no-edit"], 

14 "add": ["add", "."], 

15 "commit": ["commit", "-m", message], 

16 "push": ["push"], 

17 } 

18 if action not in commands: 

19 return {"success": False, "error": "Unknown Git action"} 

20 if action == "commit" and not message.strip(): 

21 return {"success": False, "error": "Enter a commit message"} 

22 environment = dict(os.environ, GIT_TERMINAL_PROMPT="0", GCM_INTERACTIVE="never") 

23 environment.setdefault("GIT_SSH_COMMAND", "ssh -oBatchMode=yes") 

24 try: 

25 root_result = subprocess.run( 

26 ["git", "-C", str(directory), "rev-parse", "--show-toplevel"], 

27 capture_output=True, 

28 text=True, 

29 timeout=10, 

30 env=environment, 

31 ) 

32 if root_result.returncode: 

33 return { 

34 "success": False, 

35 "error": root_result.stderr.strip() 

36 or "This folder is not in a Git repository", 

37 } 

38 root = Path(root_result.stdout.strip()).resolve() 

39 if repository and root != Path(repository).resolve(): 

40 return { 

41 "success": False, 

42 "error": "The active repository changed. Reopen source control before continuing.", 

43 } 

44 result = subprocess.run( 

45 ["git", "-C", str(root), *commands[action]], 

46 capture_output=True, 

47 text=True, 

48 timeout=120, 

49 env=environment, 

50 ) 

51 output = "\n".join( 

52 part.strip() for part in [result.stdout, result.stderr] if part.strip() 

53 ) 

54 return { 

55 "success": result.returncode == 0, 

56 "repository": str(root), 

57 "action": action, 

58 "output": output or ("Files staged." if action == "add" else "Done."), 

59 "error": output if result.returncode else "", 

60 } 

61 except FileNotFoundError: 

62 return {"success": False, "error": "Git is not installed or cannot be found"} 

63 except subprocess.TimeoutExpired: 

64 return { 

65 "success": False, 

66 "error": "Git timed out. Check the repository status before retrying.", 

67 } 

68 except OSError as error: 

69 return {"success": False, "error": str(error)}