Coverage for seedboxsync/core/utils.py: 100%

1# 

2# Copyright (C) 2015-2026 Guillaume Kulakowski <guillaume@kulakowski.fr> 

3# 

4# For the full copyright and license information, please view the LICENSE 

5# file that was distributed with this source code. 

6# 

7"""A collection of utility functions for SeedboxSync.""" 

8 

9import os 

10from os import PathLike 

11from pathlib import Path 

12from typing import Any 

13from urllib.parse import urlparse 

14from bcoding import bdecode 

15from flask import current_app as app 

16 

17 

18def byte_to_gi(bytes_value: float, suffix: str = "B") -> str: 

19 """ 

20 Convert in human readable units. 

21 

22 Args: 

23 bytes_value (integer): Value not human readable. 

24 suffix (str): Suffix for value given to (default: B). 

25 

26 Returns: 

27 str: human readable value in Gi. 

28 """ 

29 gib = bytes_value / (1024**3) 

30 return f"{gib:.1f}Gi{suffix}" 

31 

32 

33def ensure_dir_exists(path: str | PathLike[str]) -> None: 

34 """ 

35 Ensure the directory ``path`` exists, and if not create it. 

36 

37 Args: 

38 path (str): The filesystem path of a directory. 

39 

40 Raises: 

41 AssertionError: If the directory ``path`` exists, but is not a directory. 

42 

43 """ 

44 path = Path(path).expanduser().resolve() 

45 

46 if path.exists() and not path.is_dir(): 

47 raise AssertionError(f"Path `{path}` exists but is not a directory!") 

48 if not path.exists(): 

49 path.mkdir() 

50 

51 

52def get_torrent_infos(torrent_path: str | PathLike[str]) -> Any | None: 

53 """ 

54 Extracts information from a torrent file. 

55 

56 Args: 

57 torrent_path (str | PathLike[str]): Path to the torrent file. 

58 

59 Returns: 

60 str: Decoded torrent information. 

61 

62 Raises: 

63 Exception: If the file is not a valid torrent. 

64 """ 

65 with Path(torrent_path).open("rb") as torrent: 

66 torrent_info = None 

67 

68 try: 

69 torrent_info = bdecode(torrent.read()) 

70 except Exception: 

71 app.logger.exception("Not valid torrent") 

72 finally: 

73 torrent.close() 

74 

75 return torrent_info 

76 

77 

78def is_running_in_docker() -> bool: 

79 """ 

80 Return whether the current process appears to run inside Docker. 

81 

82 Returns: 

83 bool: True if in docker envoronment. 

84 

85 """ 

86 # Test mountinfo 

87 if Path("/proc/self/mountinfo").exists(): 

88 with Path("/proc/self/mountinfo").open() as f: 

89 if "docker" in f.read() or "overlay" in f.read(): 

90 return True 

91 

92 # Test du cgroup 

93 if Path("/proc/1/cgroup").exists(): 

94 with Path("/proc/1/cgroup").open() as f: 

95 lines = f.read() 

96 if "docker" in lines or "kubepods" in lines: 

97 return True 

98 

99 # Test /.dockerenv but not on podman 

100 return Path("/.dockerenv").exists() 

101 

102 

103def get_web_healthcheck_url() -> str: 

104 """ 

105 Return the URL used to check the local Flask application. 

106 

107 Returns: 

108 str: The healthcheck URL. 

109 """ 

110 explicit_url = os.getenv("HEALTHCHECK_URL") 

111 if explicit_url: 

112 return explicit_url.rstrip("/") + "/healthcheck" 

113 

114 bind = os.getenv("BIND") 

115 if bind: 

116 return _healthcheck_url_from_bind(bind) 

117 

118 port = 8000 if is_running_in_docker() else 5000 

119 return f"http://127.0.0.1:{port}/healthcheck" 

120 

121 

122def _healthcheck_url_from_bind(bind: str) -> str: 

123 """ 

124 Build a local healthcheck URL from a Gunicorn bind value. 

125 

126 Returns: 

127 str: The healthcheck URL from BIND. 

128 """ 

129 bind = bind.strip() 

130 

131 if bind.startswith("unix:"): 

132 raise ValueError("A Unix socket bind cannot be checked with a standard HTTP URL.") 

133 

134 # urlparse requires a scheme to correctly parse host and port. 

135 parsed = urlparse(f"//{bind}") 

136 

137 if parsed.port is None: 

138 raise ValueError(f"Invalid BIND value: {bind}") 

139 

140 host = parsed.hostname or "127.0.0.1" 

141 

142 if host in {"0.0.0.0", "::", "[::]"}: 

143 host = "127.0.0.1" 

144 

145 return f"http://{host}:{parsed.port}/healthcheck"