Spaces:
Sleeping
Sleeping
| """ | |
| Shared fixtures and mock builders for the Beacon test suite. | |
| Factory functions (not fixtures) build Anthropic mock objects so they can be | |
| called from any scope without fixture injection. | |
| """ | |
| from __future__ import annotations | |
| from typing import Iterator | |
| from unittest.mock import MagicMock | |
| import pytest | |
| from models import PatientProfile | |
| # --------------------------------------------------------------------------- | |
| # Anthropic mock-object factories | |
| # --------------------------------------------------------------------------- | |
| def make_text_block(text: str) -> MagicMock: | |
| """Return a mock ContentBlock with type='text' and .text=text.""" | |
| block = MagicMock() | |
| block.type = "text" | |
| block.text = text | |
| return block | |
| def make_tool_use_block(name: str, input_data: dict, tool_use_id: str = "tu_001") -> MagicMock: | |
| """Return a mock ContentBlock with type='tool_use'.""" | |
| block = MagicMock() | |
| block.type = "tool_use" | |
| block.name = name | |
| block.id = tool_use_id | |
| block.input = input_data | |
| return block | |
| def make_message( | |
| content: list, | |
| stop_reason: str = "end_turn", | |
| model: str = "claude-sonnet-4-6", | |
| ) -> MagicMock: | |
| """Return a mock Message with .content, .stop_reason, and .model.""" | |
| msg = MagicMock() | |
| msg.content = content | |
| msg.stop_reason = stop_reason | |
| msg.model = model | |
| return msg | |
| class FakeStream: | |
| """ | |
| Minimal stand-in for the object returned by ``client.messages.stream()``. | |
| Usage:: | |
| fake = FakeStream(tokens=["Hello", " world"], final_message=make_message([...])) | |
| with fake as stream: | |
| for chunk in stream.text_stream: | |
| ... | |
| msg = stream.get_final_message() | |
| """ | |
| def __init__(self, tokens: list[str], final_message: MagicMock) -> None: | |
| self._tokens = tokens | |
| self._final_message = final_message | |
| def __enter__(self) -> "FakeStream": | |
| return self | |
| def __exit__(self, *args: object) -> None: | |
| pass | |
| def text_stream(self) -> Iterator[str]: | |
| yield from self._tokens | |
| def get_final_message(self) -> MagicMock: | |
| return self._final_message | |
| # --------------------------------------------------------------------------- | |
| # Fixtures | |
| # --------------------------------------------------------------------------- | |
| def als_patient() -> PatientProfile: | |
| """ALS patient fixture used across most tests.""" | |
| return PatientProfile( | |
| disease="Amyotrophic Lateral Sclerosis", | |
| age=52, | |
| onset_months=18, | |
| diagnosis_months=12, | |
| benchmarks={"alsfrs_r": "38", "forced_vital_capacity_percent": "72"}, | |
| zip_code="02115", | |
| country_code="US", | |
| lat=42.3370, | |
| lon=-71.1061, | |
| radius_miles=100, | |
| phases=["2", "3"], | |
| include_eap=False, | |
| include_observational=False, | |
| lang="en", | |
| ) | |
| def mock_client() -> MagicMock: | |
| """Bare MagicMock Anthropic client — callers set side_effects as needed.""" | |
| return MagicMock() | |