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

1from dataclasses import dataclass 

2 

3from jumpstarter_driver_composite.client import CompositeClient 

4from jumpstarter_driver_opendal.common import PathBuf 

5from opendal import Operator 

6from yarl import URL 

7 

8 

9@dataclass(kw_only=True) 

10class HttpServerClient(CompositeClient): 

11 """Client for the HTTP server driver""" 

12 

13 def start(self): 

14 """ 

15 Start the HTTP server. 

16 

17 Initializes and starts the HTTP server if it's not already running. 

18 The server will listen on the configured host and port. 

19 """ 

20 self.call("start") 

21 

22 def stop(self): 

23 """ 

24 Stop the HTTP server. 

25 

26 Stops the running HTTP server and releases associated resources. 

27 Raises: 

28 ServerNotRunning: If the server is not currently running 

29 """ 

30 self.call("stop") 

31 

32 def get_host(self) -> str: 

33 """ 

34 Get the host IP address the HTTP server is listening on. 

35 

36 Returns: 

37 str: The IP address or hostname the server is bound to 

38 """ 

39 return self.call("get_host") 

40 

41 def get_port(self) -> int: 

42 """ 

43 Get the port number the HTTP server is listening on. 

44 

45 Returns: 

46 int: The port number (default is 8080) 

47 """ 

48 return self.call("get_port") 

49 

50 def get_url(self) -> str: 

51 """ 

52 Get the base URL of the HTTP server. 

53 

54 Returns: 

55 str: The base URL of the server 

56 """ 

57 return self.call("get_url") 

58 

59 def put_file(self, dst: PathBuf, src: PathBuf, operator: Operator | None = None) -> str: 

60 """ 

61 Upload a file to the HTTP server using a opendal operator as source. 

62 

63 Args: 

64 dst (PathBuf): Name to save the file as on the server. 

65 src (PathBuf): Name to read the file from opendal operator. 

66 operator (Operator): opendal operator to read the file from, defaults to local fs. 

67 

68 Returns: 

69 str: URL of the uploaded file 

70 """ 

71 self.storage.write_from_path(dst, src, operator) 

72 

73 return str(URL(self.get_url()).joinpath(dst))