Coverage for /home/fedora/jumpstarter/packages/jumpstarter-driver-http/jumpstarter_driver_http/driver.py: 59%

1import os 

2from dataclasses import dataclass, field 

3from typing import Optional 

4 

5import anyio 

6from aiohttp import web 

7from jumpstarter_driver_opendal.driver import Opendal 

8 

9from jumpstarter.driver import Driver, export 

10 

11 

12class HttpServerError(Exception): 

13 """Base exception for HTTP server errors""" 

14 

15 

16class FileWriteError(HttpServerError): 

17 """Exception raised when file writing fails""" 

18 

19 

20@dataclass(kw_only=True) 

21class HttpServer(Driver): 

22 """HTTP Server driver for Jumpstarter""" 

23 

24 root_dir: str = "/var/www" 

25 host: str | None = field(default=None) 

26 port: int = 8080 

27 timeout: int = field(default=600) 

28 app: web.Application = field(init=False, default_factory=web.Application) 

29 runner: Optional[web.AppRunner] = field(init=False, default=None) 

30 

31 def __post_init__(self): 

32 if hasattr(super(), "__post_init__"): 

33 super().__post_init__() 

34 

35 os.makedirs(self.root_dir, exist_ok=True) 

36 

37 self.children["storage"] = Opendal(scheme="fs", kwargs={"root": self.root_dir}) 

38 self.app.router.add_routes( 

39 [ 

40 web.static("/", self.root_dir), 

41 ] 

42 ) 

43 if self.host is None: 

44 self.host = self.get_default_ip() 

45 

46 def get_default_ip(self): 

47 try: 

48 import socket 

49 

50 with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: 

51 s.connect(("8.8.8.8", 80)) 

52 return s.getsockname()[0] 

53 except Exception: 

54 self.logger.warning("Could not determine default IP address, falling back to 0.0.0.0") 

55 return "0.0.0.0" 

56 

57 @classmethod 

58 def client(cls) -> str: 

59 """Return the import path of the corresponding client""" 

60 return "jumpstarter_driver_http.client.HttpServerClient" 

61 

62 @export 

63 async def start(self): 

64 """ 

65 Start the HTTP server. 

66 

67 Raises: 

68 HttpServerError: If the server fails to start. 

69 """ 

70 if self.runner is not None: 

71 self.logger.warning("HTTP server is already running.") 

72 return 

73 

74 self.runner = web.AppRunner(self.app) 

75 if self.runner: 

76 await self.runner.setup() 

77 

78 site = web.TCPSite(self.runner, self.host, self.port) 

79 await site.start() 

80 self.logger.info(f"HTTP server started at http://{self.host}:{self.port}") 

81 

82 @export 

83 async def stop(self): 

84 """ 

85 Stop the HTTP server. 

86 

87 Raises: 

88 HttpServerError: If the server fails to stop. 

89 """ 

90 if self.runner is None: 

91 self.logger.warning("HTTP server is not running.") 

92 return 

93 

94 await self.runner.cleanup() 

95 self.logger.info("HTTP server stopped.") 

96 self.runner = None 

97 

98 @export 

99 def get_url(self) -> str: 

100 """ 

101 Get the base URL of the HTTP server. 

102 

103 Returns: 

104 str: Base URL of the HTTP server. 

105 """ 

106 return f"http://{self.host}:{self.port}" 

107 

108 @export 

109 def get_host(self) -> str | None: 

110 """ 

111 Get the host IP address of the HTTP server. 

112 

113 Returns: 

114 str: Host IP address. 

115 """ 

116 return self.host 

117 

118 @export 

119 def get_port(self) -> int: 

120 """ 

121 Get the port number of the HTTP server. 

122 

123 Returns: 

124 int: Port number. 

125 """ 

126 return self.port 

127 

128 def close(self): 

129 if self.runner: 

130 try: 

131 if anyio.get_current_task(): 

132 anyio.from_thread.run(self._async_cleanup) 

133 except Exception as e: 

134 self.logger.warning(f"HTTP server cleanup failed synchronously: {e}") 

135 self.runner = None 

136 super().close() 

137 

138 async def _async_cleanup(self): 

139 try: 

140 if self.runner: 

141 await self.runner.shutdown() 

142 await self.runner.cleanup() 

143 self.logger.info("HTTP server cleanup completed asynchronously.") 

144 except Exception as e: 

145 self.logger.error(f"HTTP server cleanup failed asynchronously: {e}")