Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/airflow/utils/platform.py: 0%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17"""Platform and system specific function.""" 

18 

19from __future__ import annotations 

20 

21import getpass 

22import logging 

23import os 

24import pkgutil 

25import platform 

26import sys 

27from functools import cache 

28 

29IS_WINDOWS = platform.system() == "Windows" 

30 

31log = logging.getLogger(__name__) 

32 

33 

34def is_tty(): 

35 """Check if stdout is connected (is associated with a terminal device) to a tty(-like) device.""" 

36 if not hasattr(sys.stdout, "isatty"): 

37 return False 

38 return sys.stdout.isatty() 

39 

40 

41def is_terminal_support_colors() -> bool: 

42 """Try to determine if the current terminal supports colors.""" 

43 if sys.platform == "win32": 

44 return False 

45 if not is_tty(): 

46 return False 

47 if "COLORTERM" in os.environ: 

48 return True 

49 term = os.environ.get("TERM", "dumb").lower() 

50 if term in ("xterm", "linux") or "color" in term: 

51 return True 

52 return False 

53 

54 

55def get_airflow_git_version(): 

56 """Return the git commit hash representing the current version of the application.""" 

57 git_version = None 

58 try: 

59 git_version = str(pkgutil.get_data("airflow", "git_version"), encoding="UTF-8") 

60 except Exception as e: 

61 log.debug(e) 

62 

63 return git_version 

64 

65 

66@cache 

67def getuser() -> str: 

68 """ 

69 Get the username of the current user, or error with a nice error message if there's no current user. 

70 

71 We don't want to fall back to os.getuid() because not having a username 

72 probably means the rest of the user environment is wrong (e.g. no $HOME). 

73 Explicit failure is better than silently trying to work badly. 

74 """ 

75 try: 

76 return getpass.getuser() 

77 except KeyError: 

78 # Inner import to avoid circular import 

79 from airflow.exceptions import AirflowConfigException 

80 

81 raise AirflowConfigException( 

82 "The user that Airflow is running as has no username; you must run" 

83 "Airflow as a full user, with a username and home directory, " 

84 "in order for it to function properly." 

85 )