Coverage for frappe_manager / docker / docker_exceptions.py: 47%

1from frappe_manager.docker.subprocess_output import SubprocessOutput 

2 

3 

4class DockerException(Exception): 

5 """Exception raised when Docker command execution fails.""" 

6 

7 def __init__( 

8 self, 

9 command_launched: list[str], 

10 output: SubprocessOutput, 

11 ): 

12 self.docker_command: list[str] = command_launched 

13 self.output = output 

14 

15 command_launched_str = " ".join(command_launched) 

16 

17 error_msg = ( 

18 f"The docker command executed was `{command_launched_str}`.\n" 

19 f"It returned with code {self.output.exit_code}\n" 

20 ) 

21 

22 if self.output.stdout: 

23 stdout_output = "\n".join(self.output.stdout) 

24 error_msg += f"The content of stdout is \n{'--' * 10}\n'{stdout_output}'\n" 

25 else: 

26 error_msg += "The content of stdout can be found above the stacktrace (it wasn't captured).\n" 

27 

28 if self.output.stderr: 

29 stderr_output = "\n".join(self.output.stderr) 

30 error_msg += f"The content of stderr is \n{'--' * 10}\n'{stderr_output}'\n" 

31 else: 

32 error_msg += "The content of stderr can be found above the stacktrace (it wasn't captured)." 

33 

34 super().__init__(error_msg) 

35 

36 

37def is_port_conflict_error(exception: DockerException) -> tuple[bool, list[int]]: 

38 """ 

39 Detect if DockerException was caused by port binding conflict. 

40 

41 Returns: 

42 Tuple of (is_conflict, list_of_conflicting_ports) 

43 

44 Examples: 

45 >>> exc = DockerException(["compose", "up"], output_with_port_error) 

46 >>> is_conflict, ports = is_port_conflict_error(exc) 

47 >>> is_conflict 

48 True 

49 >>> ports 

50 [80, 443] 

51 """ 

52 import re 

53 

54 stderr_text = "\n".join(exception.output.stderr) if exception.output.stderr else "" 

55 

56 if not ("port is already allocated" in stderr_text or "address already in use" in stderr_text): 

57 return False, [] 

58 

59 ports = [] 

60 port_patterns = [ 

61 r"Bind for (?:0\.0\.0\.0|:::):(\d+)", 

62 r"port (\d+) is already allocated", 

63 r"address already in use.*:(\d+)", 

64 ] 

65 

66 for pattern in port_patterns: 

67 matches = re.findall(pattern, stderr_text) 

68 for match in matches: 

69 port = int(match) 

70 if port not in ports: 

71 ports.append(port) 

72 

73 return len(ports) > 0, sorted(ports)