Coverage for bookbuilderpy/resources/_resource_server.py: 24%

1"""An internal web server for serving persistent resources.""" 

2from http.server import BaseHTTPRequestHandler, HTTPServer 

3from importlib import resources # nosem 

4from threading import Thread 

5 

6 

7def _get_file(name: str) -> bytes | None: 

8 """ 

9 Get a file from the resource server. 

10 

11 :param name: the file name 

12 :return: the file contents 

13 """ 

14 if not name: 

15 return None 

16 

17 while name[0] == "/": 

18 name = name[1:] 

19 if not name: 

20 return None 

21 

22 i = name.rfind("?") 

23 if i >= 0: 

24 name = name[:i] 

25 i = name.rfind("#") 

26 if i >= 0: 

27 name = name[:i] 

28 i = name.rfind("/") 

29 if i >= 0: 

30 name = name[i + 1:] 

31 fn = name.strip() 

32 if not name: 

33 return None 

34 

35 pack = str(__package__) 

36 if resources.is_resource(package=pack, name=fn): 

37 with resources.open_binary(package=pack, 

38 resource=fn) as stream: 

39 return stream.read() 

40 

41 return None 

42 

43 

44class _SimpleHTTPRequestHandler(BaseHTTPRequestHandler): 

45 """The internal server for obtaining resources.""" 

46 

47 # noinspection PyPep8Naming 

48 def do_GET(self) -> None: # noqa 

49 """Get the resource.""" 

50 # noinspection PyUnresolvedReferences 

51 res = _get_file(self.path) 

52 if res: 

53 self.send_response(200) 

54 self.end_headers() 

55 self.wfile.write(res) 

56 else: 

57 self.send_response(404) 

58 self.end_headers() 

59 

60 

61class ResourceServer: 

62 """The resource server.""" 

63 

64 def __init__(self): 

65 """Initialize the server.""" 

66 self.__httpd = HTTPServer(("localhost", 0), 

67 _SimpleHTTPRequestHandler) 

68 self.__thread = Thread(target=self.__serve) 

69 

70 def __serve(self) -> None: 

71 """Start the server and serve.""" 

72 self.__httpd.serve_forever() 

73 

74 def get_server(self) -> str: 

75 """ 

76 Get the server address. 

77 

78 :return: the server address 

79 :rtype: str 

80 """ 

81 return f"http://localhost:{self.__httpd.socket.getsockname()[1]}/" 

82 

83 def get_mathjax_url(self) -> str: 

84 """ 

85 Get the mathjax url. 

86 

87 :return: the mathjax url 

88 :rtype: str 

89 """ 

90 return f"{self.get_server()}mathjax.js" 

91 

92 def __enter__(self): 

93 """Start the resource server.""" 

94 self.__thread.start() 

95 return self 

96 

97 def __exit__(self, exception_type, exception_value, traceback) -> bool: 

98 """ 

99 Close and exist the server. 

100 

101 :param exception_type: ignored 

102 :param exception_value: ignored 

103 :param traceback: ignored 

104 :returns: `True` to suppress an exception, `False` to rethrow it 

105 """ 

106 self.__httpd.shutdown() 

107 del self.__httpd 

108 self.__thread.join() 

109 del self.__thread 

110 return exception_type is None