Coverage for tests / cli / conftest.py: 93%

1import os (empty)

2import select (empty)

3import signal (empty)

4import subprocess (empty)

5import threading (empty)

6import time (empty)

7from collections.abc import Generator (empty)

8from contextlib import contextmanager, suppress (empty)

9from pathlib import Path (empty)

10from textwrap import dedent (empty)

11 

12import pytest (empty)

13 

14from faststream import FastStream (empty)

15from faststream._internal._compat import IS_WINDOWS (empty)

16from tests.cli import interfaces (empty)

17 

18 

19@pytest.fixture() (empty)

20def broker(): (empty)

21 # separate import from e2e tests 

22 from faststream.rabbit import RabbitBroker (empty)

23 

24 return RabbitBroker() (empty)

25 

26 

27@pytest.fixture() (empty)

28def app_without_logger(broker) -> FastStream: (empty)

29 return FastStream(broker, logger=None) (empty)

30 

31 

32@pytest.fixture() (empty)

33def app_without_broker() -> FastStream: (empty)

34 return FastStream() (empty)

35 

36 

37@pytest.fixture() (empty)

38def app(broker) -> FastStream: (empty)

39 return FastStream(broker) (empty)

40 

41 

42@pytest.fixture() (empty)

43def faststream_tmp_path(tmp_path: "Path"): (empty)

44 faststream_tmp = tmp_path / "faststream_templates" (empty)

45 faststream_tmp.mkdir(exist_ok=True) (empty)

46 return faststream_tmp (empty)

47 

48 

49@pytest.fixture() (empty)

50def generate_template( (empty)

51 faststream_tmp_path: "Path", 

52) -> interfaces.GenerateTemplateFactory: 

53 @contextmanager (empty)

54 def factory( (empty)

55 code: str, 

56 filename: str = "temp_app.py", 

57 ) -> Generator["Path", None, None]: 

58 file_path: Path = faststream_tmp_path / filename (empty)

59 cleaned_code = dedent(code).strip() (empty)

60 

61 file_path.write_text(cleaned_code, encoding="utf-8") (empty)

62 

63 try: (empty)

64 yield file_path (empty)

65 finally: 

66 file_path.unlink(missing_ok=True) (empty)

67 

68 return factory (empty)

69 

70 

71class CLIThread: (empty)

72 def __init__( (empty)

73 self, 

74 command: tuple[str, ...], (empty)

75 env: dict[str, str], (empty)

76 ) -> None: (empty)

77 self.process = subprocess.Popen( (empty)

78 command, (empty)

79 stderr=subprocess.PIPE, (empty)

80 text=True, (empty)

81 shell=False, (empty)

82 env=env, (empty)

83 ) 

84 self.running = True (empty)

85 self.started = False (empty)

86 

87 self.stderr = "" (empty)

88 

89 self.__std_poll_thread = threading.Thread(target=self._poll_std) (empty)

90 self.__std_poll_thread.start() (empty)

91 

92 def _poll_std(self) -> None: (empty)

93 assert self.process.stderr (empty)

94 

95 if IS_WINDOWS: (empty)

96 return (empty)

97 

98 while self.running: (empty)

99 rlist, _, _ = select.select([self.process.stderr], [], [], 0.1) (empty)

100 if rlist: (empty)

101 self.started = True (empty)

102 

103 if line := self.process.stderr.readline(): (empty)

104 self.stderr += line.strip() (empty)

105 

106 else: 

107 break (empty)

108 

109 elif self.process.poll() is not None: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true(empty)

110 break 

111 

112 def wait_for_stderr(self, message: str, timeout: float = 2.0) -> bool: (empty)

113 assert self.process.stderr (empty)

114 

115 if message in self.stderr: (empty)

116 return True (empty)

117 

118 expiration_time = time.time() + timeout (empty)

119 

120 while time.time() < expiration_time: (empty)

121 time.sleep(0.1) (empty)

122 if message in self.stderr: (empty)

123 return True (empty)

124 

125 if self.process.returncode is not None: 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true(empty)

126 return message in self.process.stderr.read() 

127 

128 return False (empty)

129 

130 def wait(self, timeout: float) -> None: (empty)

131 self.process.wait(timeout) (empty)

132 

133 def signint(self) -> None: (empty)

134 if IS_WINDOWS: (empty)

135 self.process.terminate() (empty)

136 else: 

137 self.process.send_signal(signal.SIGINT) (empty)

138 

139 def stop(self) -> None: (empty)

140 self.process.terminate() (empty)

141 

142 self.running = False (empty)

143 with suppress(Exception): (empty)

144 self.__std_poll_thread.join() (empty)

145 

146 try: (empty)

147 self.wait(5) (empty)

148 

149 except subprocess.TimeoutExpired: 

150 self.process.kill() 

151 

152 

153@pytest.fixture() (empty)

154def faststream_cli( (empty)

155 faststream_tmp_path: Path, 

156) -> interfaces.FastStreamCLIFactory: 

157 @contextmanager (empty)

158 def cli_factory( (empty)

159 *cmd: str, 

160 wait_time: float = 2.0, 

161 extra_env: dict[str, str] | None = None, 

162 ) -> Generator[CLIThread, None, None]: 

163 env = ( (empty)

164 os.environ.copy() 

165 | { 

166 "PATH": f"{faststream_tmp_path}:{os.environ['PATH']}", 

167 "PYTHONPATH": str(faststream_tmp_path), 

168 } 

169 | (extra_env or {}) 

170 ) 

171 

172 cli = CLIThread(cmd, env) (empty)

173 

174 wait_for_startup(cli, wait_time) (empty)

175 

176 try: (empty)

177 yield cli (empty)

178 finally: 

179 cli.stop() (empty)

180 

181 return cli_factory (empty)

182 

183 

184def wait_for_startup( (empty)

185 cli: CLIThread, 

186 timeout: float = 10, 

187 check_interval: float = 0.1, 

188) -> None: 

189 start_time = time.time() (empty)

190 

191 while time.time() - start_time < timeout: (empty)

192 if cli.started: (empty)

193 return (empty)

194 

195 if cli.process.poll() is not None: 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true(empty)

196 return 

197 

198 time.sleep(check_interval) (empty)

199 

200 return (empty)