Test normal completion.
(self)
| 212 | assert state.finish_reason == "stop" |
| 213 | |
| 214 | def test_complete_normal(self): |
| 215 | """Test normal completion.""" |
| 216 | state = StreamingState() |
| 217 | state.add_chunk(StreamingChunk(content="Test")) |
| 218 | |
| 219 | state.complete() |
| 220 | |
| 221 | assert state.phase == StreamingPhase.COMPLETED |
| 222 | assert not state.interrupted |
| 223 | assert state.is_complete |
| 224 | assert not state.is_active |
| 225 | assert state.end_time is not None |
| 226 | |
| 227 | def test_complete_interrupted(self): |
| 228 | """Test interrupted completion.""" |
nothing calls this directly
no test coverage detected