Coverage for crunch/client/diagnostics.py: 100.00%

1import platform, socket, re, uuid, psutil, os 

2import getpass 

3 

4import os 

5from pathlib import Path 

6import subprocess 

7 

8def version() -> str: 

9 """ 

10 Gets the version number of the django-crunch module. 

11 

12 Returns: 

13 str: The current version. 

14 """ 

15 import importlib.metadata 

16 version = importlib.metadata.version("django-crunch") 

17 return version 

18 

19 

20# Return the git revision as a string 

21def git_revision(directory:Path=None) -> str: 

22 """ 

23 Gets the git revision hash for a directory. 

24 

25 Adapted from https://stackoverflow.com/a/40170206 which was taken from NumPy. 

26 

27 Args: 

28 directory (Path, optional): The directory we are interested in. Defaults to None in which case it uses the directory of the current source file. 

29 

30 Returns: 

31 str: The git hash for the current revision. 

32 """ 

33 directory = directory or Path(__file__).parent 

34 def _minimal_ext_cmd(cmd): 

35 # construct minimal environment 

36 env = {} 

37 for k in ['SYSTEMROOT', 'PATH']: 

38 v = os.environ.get(k) 

39 if v is not None: 

40 env[k] = v 

41 # LANGUAGE is used on win32 

42 env['LANGUAGE'] = 'C' 

43 env['LANG'] = 'C' 

44 env['LC_ALL'] = 'C' 

45 out = subprocess.Popen(cmd, stdout = subprocess.PIPE, env=env, cwd=directory).communicate()[0] 

46 return out 

47 

48 try: 

49 out = _minimal_ext_cmd(['git', 'rev-parse', 'HEAD']) 

50 GIT_REVISION = out.strip().decode('ascii') 

51 except OSError: 

52 GIT_REVISION = "Unknown" 

53 

54 return GIT_REVISION 

55 

56def get_diagnostic(diagnostics:dict, key:str, func, default=""): 

57 try: 

58 result = func() 

59 except Exception: 

60 result = default 

61 diagnostics[key] = result 

62 return result 

63 

64def get_diagnostics() -> dict: 

65 """ 

66 Gets diagnostic information about the current environment. 

67 

68 Used when sending status updates to a crunch hosted site. 

69 

70 Returns: 

71 dict: A dictionary with the diagnostic information. 

72 """ 

73 diagnostics = dict() 

74 

75 # adapts code from here: https://stackoverflow.com/questions/3103178/how-to-get-the-system-info-with-python 

76 get_diagnostic(diagnostics, 'agent_user', lambda: getpass.getuser()) 

77 get_diagnostic(diagnostics, 'version', lambda: version()) 

78 get_diagnostic(diagnostics, 'revision', lambda: git_revision()) 

79 get_diagnostic(diagnostics, 'system', lambda: platform.system()) 

80 get_diagnostic(diagnostics, 'system_release', lambda: platform.release()) 

81 get_diagnostic(diagnostics, 'system_version', lambda: platform.version()) 

82 get_diagnostic(diagnostics, 'machine', lambda: platform.machine()) 

83 get_diagnostic(diagnostics, 'hostname', lambda: socket.gethostname()) 

84 get_diagnostic(diagnostics, 'ip_address', lambda: socket.gethostbyname(socket.gethostname())) # this could use the hostname from above 

85 get_diagnostic(diagnostics, 'mac_address', lambda: ':'.join(re.findall('..', '%012x' % uuid.getnode()))) 

86 get_diagnostic(diagnostics, 'memory_total', lambda: psutil.virtual_memory().total, default=None) 

87 get_diagnostic(diagnostics, 'memory_free', lambda: psutil.virtual_memory().available, default=None) 

88 get_diagnostic(diagnostics, 'disk_total', lambda: psutil.disk_usage(os.getcwd()).total, default=None) 

89 get_diagnostic(diagnostics, 'disk_free', lambda: psutil.disk_usage(os.getcwd()).free, default=None) 

90 

91 return diagnostics