()
| 849 | |
| 850 | @pytest.fixture |
| 851 | def messages_client() -> TestClient: |
| 852 | def handler(request: httpx.Request) -> httpx.Response: |
| 853 | body = json.loads(request.content.decode("utf-8")) |
| 854 | return httpx.Response( |
| 855 | 200, |
| 856 | json={ |
| 857 | "id": "msg_fixture", |
| 858 | "type": "message", |
| 859 | "role": "assistant", |
| 860 | "model": body.get("model", "fixture-model"), |
| 861 | "content": [{"type": "text", "text": "ok"}], |
| 862 | "stop_reason": "end_turn", |
| 863 | "stop_sequence": None, |
| 864 | "usage": {"input_tokens": 10, "output_tokens": 1}, |
| 865 | }, |
| 866 | headers={"content-type": "application/json"}, |
| 867 | ) |
| 868 | |
| 869 | spend_control = SpendControl(storage=InMemorySpendControlStorage()) |
| 870 | async_client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) |
| 871 | old_get_client = __import__("uncommon_route.proxy", fromlist=["_get_client"])._get_client |
| 872 | __import__("uncommon_route.proxy", fromlist=["_get_client"])._get_client = lambda: async_client |
| 873 | app = create_app( |
| 874 | upstream="http://127.0.0.1:1/fake", |
| 875 | spend_control=spend_control, |
| 876 | connections_store=ConnectionsStore(storage=InMemoryConnectionsStorage()), |
| 877 | ) |
| 878 | client = TestClient(app, raise_server_exceptions=False) |
| 879 | try: |
| 880 | yield client |
| 881 | finally: |
| 882 | __import__("uncommon_route.proxy", fromlist=["_get_client"])._get_client = old_get_client |
| 883 | asyncio.run(async_client.aclose()) |
| 884 | |
| 885 | |
| 886 | def _build_seed_mapper(*model_ids: str) -> ModelMapper: |
nothing calls this directly
no test coverage detected