def test_sync_generator_task_error_handling(exporter):
    """Test error handling in synchronous generator task"""
    error_message = "Intentional generator test error"

    @task(name="failing_generator_task")
    def failing_generator_task() -> Generator[int, None, None]:
        yield 1
        yield 2
        raise ValueError(error_message)

    results = []
    with pytest.raises(ValueError, match=error_message):
        for num in failing_generator_task():
            results.append(num)

    spans = exporter.get_finished_spans()
    assert len(spans) == 1

    task_span = spans[0]
    assert task_span.name == "failing_generator_task.task"
    assert task_span.status.status_code == StatusCode.ERROR
    assert error_message in task_span.status.description
    assert results == [1, 2]
Test error handling in synchronous generator task
packages/traceloop-sdk/tests/test_tasks.py