Mercurial > code > home > repos > rdferry
annotate examples/_run_server_child.py @ 12:ba73d8ba81dc default tip
refactor
| author | drewp@bigasterisk.com |
|---|---|
| date | Mon, 18 Mar 2024 16:51:44 -0700 |
| parents | 0bc06da6bf74 |
| children |
| rev | line source |
|---|---|
| 3 | 1 import asyncio |
| 11 | 2 import contextlib |
| 3 import logging | |
| 3 | 4 from dataclasses import dataclass |
| 11 | 5 from datetime import timedelta |
| 3 | 6 from pathlib import Path |
| 7 | |
| 8 import aiohttp | |
| 11 | 9 from aiohttp_sse_client import client as sse_client |
| 10 | |
| 11 log = logging.getLogger('chil') | |
| 3 | 12 |
| 13 | |
| 14 @dataclass | |
| 8 | 15 class RunHttpServerChildProcess: |
| 3 | 16 server_path: Path |
| 17 | |
| 18 async def __aenter__(self): | |
| 19 self.subprocess = await asyncio.create_subprocess_exec( | |
| 20 'pdm', 'run', 'python', self.server_path) | |
| 21 self._session = await aiohttp.ClientSession().__aenter__() | |
| 22 return self | |
| 23 | |
| 24 async def __aexit__(self, exc_type, exc, tb): | |
|
6
d885fce5e4e7
try to reduce warnings about leaked sessions
drewp@bigasterisk.com
parents:
5
diff
changeset
|
25 await self._session.close() |
| 3 | 26 self.subprocess.terminate() |
|
5
b34cd6619316
new test. server has to be able to exit between tests.
drewp@bigasterisk.com
parents:
3
diff
changeset
|
27 await self.subprocess.wait() |
|
6
d885fce5e4e7
try to reduce warnings about leaked sessions
drewp@bigasterisk.com
parents:
5
diff
changeset
|
28 |
| 10 | 29 async def get(self, url: str, headers=None) -> aiohttp.ClientResponse: |
| 3 | 30 while True: |
| 31 try: | |
| 10 | 32 return await self._session.get(url, headers=headers) |
| 3 | 33 except aiohttp.ClientConnectorError: |
| 34 await asyncio.sleep(0.05) | |
| 11 | 35 |
| 36 @contextlib.asynccontextmanager | |
| 37 async def eventSource(self, url: str): | |
| 38 async with sse_client.EventSource( | |
| 39 url, reconnection_time=timedelta(seconds=.05)) as es: | |
| 40 yield es | |
| 41 | |
| 42 | |
| 43 async def assert_event_stream_starts_with(http_server, url, expected_events): | |
| 44 events_left = expected_events[:] | |
| 45 async with http_server.eventSource(url) as es: | |
| 46 async for event in es: | |
| 47 assert (event.message, event.data) == events_left[0] | |
| 48 events_left.pop(0) | |
| 49 if not events_left: | |
| 50 break |