| 222 | |
| 223 | |
| 224 | class TestServer(Server): |
| 225 | @property |
| 226 | def url(self) -> httpx.URL: |
| 227 | protocol = "https" if self.config.is_ssl else "http" |
| 228 | return httpx.URL(f"{protocol}://{self.config.host}:{self.config.port}/") |
| 229 | |
| 230 | def install_signal_handlers(self) -> None: |
| 231 | # Disable the default installation of handlers for signals such as SIGTERM, |
| 232 | # because it can only be done in the main thread. |
| 233 | pass # pragma: nocover |
| 234 | |
| 235 | async def serve(self, sockets=None): |
| 236 | self.restart_requested = asyncio.Event() |
| 237 | |
| 238 | loop = asyncio.get_event_loop() |
| 239 | tasks = { |
| 240 | loop.create_task(super().serve(sockets=sockets)), |
| 241 | loop.create_task(self.watch_restarts()), |
| 242 | } |
| 243 | await asyncio.wait(tasks) |
| 244 | |
| 245 | async def restart(self) -> None: # pragma: no cover |
| 246 | # This coroutine may be called from a different thread than the one the |
| 247 | # server is running on, and from an async environment that's not asyncio. |
| 248 | # For this reason, we use an event to coordinate with the server |
| 249 | # instead of calling shutdown()/startup() directly, and should not make |
| 250 | # any asyncio-specific operations. |
| 251 | self.started = False |
| 252 | self.restart_requested.set() |
| 253 | while not self.started: |
| 254 | await sleep(0.2) |
| 255 | |
| 256 | async def watch_restarts(self) -> None: # pragma: no cover |
| 257 | while True: |
| 258 | if self.should_exit: |
| 259 | return |
| 260 | |
| 261 | try: |
| 262 | await asyncio.wait_for(self.restart_requested.wait(), timeout=0.1) |
| 263 | except asyncio.TimeoutError: |
| 264 | continue |
| 265 | |
| 266 | self.restart_requested.clear() |
| 267 | await self.shutdown() |
| 268 | await self.startup() |
| 269 | |
| 270 | |
| 271 | def serve_in_thread(server: TestServer) -> typing.Iterator[TestServer]: |