Coverage for src/tagmania/delete_volumes.py: 98%

1import argparse 

2 

3from tagmania.iac_tools.clusterset import ClusterSet 

4from tagmania.iac_tools.tagset import TagSet 

5 

6 

7def main(): 

8 parser = argparse.ArgumentParser( 

9 description="AWS cluster volume deletion.", 

10 epilog='This tool relies on the "Cluster" and "Owner" tags on instances and ' 

11 "volumes. IAC automation puts this in place" 

12 "This tool deletes all volumes belonging to a cluster.", 

13 ) 

14 group = parser.add_mutually_exclusive_group(required=True) 

15 group.add_argument( 

16 "-D", 

17 "--delete", 

18 action="store_const", 

19 dest="delete", 

20 const=True, 

21 default=False, 

22 help="Delete snapshots from cluster CLUSTER.", 

23 ) 

24 group.add_argument( 

25 "-l", 

26 "--list", 

27 action="store_const", 

28 dest="list", 

29 const=True, 

30 default=False, 

31 help="List snapshots with a given label or all if none specified.", 

32 ) 

33 parser.add_argument( 

34 "cluster", 

35 help=""" 

36 the name CLUSTER of the cluster in question. This can be found by 

37 looking at any node in the AWS console and looking for the "Cluster" 

38 tag.""", 

39 ) 

40 

41 parser.add_argument("--profile", "-p", help="the AWS profile to use", default=None) 

42 args = parser.parse_args() 

43 

44 cluster = ClusterSet(args.cluster, profile=args.profile) 

45 volumes = cluster.get_volumes() 

46 kubernetes_volumes = cluster.get_kubernetes_volumes() 

47 

48 if args.list: 

49 print(f"Listing volumes for cluster {args.cluster}:") 

50 for volume in volumes: 

51 print(volume.id) 

52 

53 if len(kubernetes_volumes) != 0: 

54 print(f"\nKubernetes dynamically created volumes for cluster {args.cluster}:") 

55 for volume in kubernetes_volumes: 

56 ts = TagSet(volume.tags) 

57 volume_name = ts.get("Name") 

58 print(volume.id + " (" + volume_name + ")") 

59 

60 if len(volumes) == 0 and len(kubernetes_volumes) == 0: 

61 print("No volumes found.") 

62 

63 if args.delete: 

64 print(f"Listing all volumes for cluster {args.cluster}:") 

65 for volume in volumes: 

66 print(volume.id) 

67 

68 for volume in kubernetes_volumes: 

69 ts = TagSet(volume.tags) 

70 volume_name = ts.get("Name") 

71 print(volume.id + " (" + volume_name + ")") 

72 

73 if len(volumes) != 0 or len(kubernetes_volumes) != 0: 

74 confirm = input('Do you want to delete these volumes? (only "yes" is accepted): ') 

75 if confirm != "yes": 

76 print("Command Aborted.") 

77 else: 

78 if len(volumes) != 0: 

79 cluster.delete_volumes() 

80 if len(kubernetes_volumes) != 0: 

81 cluster.delete_kubernetes_volumes() 

82 else: 

83 print("No volumes found.") 

84 

85 

86if __name__ == "__main__": 

87 main()