Coverage for src/nats_contrib/micro/testing.py: 100%

1from __future__ import annotations 

2 

3from nats_contrib.micro.request import Request 

4 

5 

6class NoResponseError(Exception): 

7 """Raised when the response is not available. 

8 

9 This exception is never raised during normal operation. 

10 

11 It is only used during testing to detect when the response 

12 has not been set by the micro handler and test attempts to 

13 access the response data or headers. 

14 """ 

15 

16 

17def make_request( 

18 subject: str, 

19 data: bytes | None = None, 

20 headers: dict[str, str] | None = None, 

21) -> RequestStub: 

22 """Create a request for testing. 

23 

24 Args: 

25 subject: The subject of the request. 

26 data: The data of the request. 

27 headers: The headers of the request. 

28 

29 Returns: 

30 A request stub for testing. 

31 

32 Note: 

33 A request stub can be used to call a micro handler directly 

34 during test and to check the response data and headers. 

35 """ 

36 return RequestStub(subject, data, headers) 

37 

38 

39class RequestStub(Request): 

40 """A request stub for testing.""" 

41 

42 def __init__( 

43 self, 

44 subject: str, 

45 data: bytes | None = None, 

46 headers: dict[str, str] | None = None, 

47 ) -> None: 

48 # Because this is a testing request, we don't care about some overhead 

49 # due to asserting types. We can just use isinstance() to check the types. 

50 # This will allow small mistakes in tests to be caught early. 

51 if not isinstance(subject, str): # pyright: ignore[reportUnnecessaryIsInstance] 

52 raise TypeError(f"subject must be a string, not {type(subject).__name__}") 

53 if not isinstance(data, (bytes, type(None))): 

54 raise TypeError(f"data must be bytes, not {type(data).__name__}") 

55 if not isinstance(headers, (dict, type(None))): 

56 raise TypeError(f"headers must be a dict, not {type(headers).__name__}") 

57 self._subject = subject 

58 self._data = data or b"" 

59 self._headers = headers or {} 

60 self._response_headers: dict[str, str] | None = None 

61 self._response_data: bytes | None = None 

62 

63 def subject(self) -> str: 

64 """The subject on which request was received.""" 

65 return self._subject 

66 

67 def headers(self) -> dict[str, str]: 

68 """The headers of the request.""" 

69 return self._headers 

70 

71 def data(self) -> bytes: 

72 """The data of the request.""" 

73 return self._data 

74 

75 async def respond(self, data: bytes, headers: dict[str, str] | None = None) -> None: 

76 """Send a success response to the request. 

77 

78 Args: 

79 data: The response data. 

80 headers: Additional response headers. 

81 """ 

82 # Detect errors early during testing 

83 if not isinstance(data, bytes): # pyright: ignore[reportUnnecessaryIsInstance] 

84 raise TypeError(f"data must be bytes, not {type(data).__name__}") 

85 if not isinstance(headers, (dict, type(None))): 

86 raise TypeError(f"headers must be a dict, not {type(headers).__name__}") 

87 # Save the response data and headers for testing 

88 # Make sure there cannot be a None value when method is called 

89 self._response_data = data 

90 self._response_headers = headers or {} 

91 

92 def response_data(self) -> bytes: 

93 """Use this method durign tests to get the response data.""" 

94 if self._response_data is None: 

95 raise NoResponseError("No response has been set") 

96 return self._response_data 

97 

98 def response_headers(self) -> dict[str, str]: 

99 """Use this method during tests to get the response headers.""" 

100 if self._response_headers is None: 

101 raise NoResponseError("No response has been set") 

102 return self._response_headers