Coverage for mfutil/cli_tools/mfprogress.py: 0%

1#!/usr/bin/env python3 

2 

3import argparse 

4import threading 

5import time 

6import sys 

7import os 

8from mfutil.cli import MFProgress 

9from mfutil.misc import kill_process_and_children 

10from mfutil.bash_wrapper import BashWrapper 

11import rich 

12import psutil 

13from rich.panel import Panel 

14 

15DESCRIPTION = "execute a command with a nice progressbar" 

16TIMEOUT_FLAG = False 

17STOP_FLAG = False 

18 

19 

20def thread_advance(progress, tid, timeout): 

21 global TIMEOUT_FLAG, STOP_FLAG 

22 i = 1 

23 while i <= timeout and not STOP_FLAG: 

24 if i < timeout: 

25 progress.update(tid, advance=1) 

26 time.sleep(1) 

27 i = i + 1 

28 if not STOP_FLAG: 

29 # timeout 

30 TIMEOUT_FLAG = True 

31 current_pid = os.getpid() 

32 process = psutil.Process(current_pid) 

33 children = process.children(recursive=False) 

34 [kill_process_and_children(x.pid) for x in children] 

35 

36 

37def main(): 

38 parser = argparse.ArgumentParser(description=DESCRIPTION) 

39 parser.add_argument("COMMAND", help="command to execute") 

40 parser.add_argument("COMMAND_ARG", nargs='*', 

41 help="command arg") 

42 parser.add_argument("--timeout", 

43 help="timeout (in seconds)", type=int, 

44 default=180) 

45 parser.add_argument("--title", 

46 help="title of the command", type=str, 

47 default="title of the command") 

48 parser.add_argument("--silent", action="store_true", 

49 help="if set, we don't add a debug output in case of " 

50 "errors") 

51 args = parser.parse_args() 

52 

53 command = " ".join([args.COMMAND] + args.COMMAND_ARG) 

54 

55 status = True 

56 timeout = False 

57 with MFProgress() as progress: 

58 t = progress.add_task(args.title, total=args.timeout) 

59 x = threading.Thread(target=thread_advance, args=(progress, t, 

60 args.timeout), 

61 daemon=True) 

62 x.start() 

63 bw = BashWrapper(command) 

64 STOP_FLAG = True # noqa: 

65 if bw: 

66 progress.complete_task(t) 

67 else: 

68 if TIMEOUT_FLAG: 

69 # timeout 

70 progress.complete_task_nok(t, "timeout") 

71 timeout = True 

72 else: 

73 progress.complete_task_nok(t, "bad exit code") 

74 status = False 

75 if not status: 

76 if not args.silent and not timeout: 

77 rich.print(Panel("[bold]Error details:[/bold]\n%s" % # noqa: E999 

78 str(bw))) 

79 sys.exit(1) 

80 else: 

81 sys.exit(0) 

82 

83 

84if __name__ == '__main__': 

85 main()