[azure-core/corehttp] Add JSONL and SSE streaming support - #48077

request = <SubRequest 'recorded_test' for <Function test_system_assigned[get_token]>>
kwargs = {'test_proxy': None}, func = <function recorded_test at 0x7fd5b87404c0>
event_loop_fixture_id = 'event_loop'
setup = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup at 0x7fd5b55a5c60>
finalizer = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer at 0x7fd5b55a5a20>

    @functools.wraps(fixture)
    def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
        func = _perhaps_rebind_fixture_func(fixture, request.instance)
        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
            request, func
        )
        event_loop = request.getfixturevalue(event_loop_fixture_id)
        kwargs.pop(event_loop_fixture_id, None)
        gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
    
        async def setup():
            res = await gen_obj.__anext__()  # type: ignore[union-attr]
            return res
    
        def finalizer() -> None:
            """Yield again, to finalize."""
    
            async def async_finalizer() -> None:
                try:
                    await gen_obj.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    pass
                else:
                    msg = "Async generator fixture didn't stop."
                    msg += "Yield only once."
                    raise ValueError(msg)
    
            event_loop.run_until_complete(async_finalizer())
    
>       result = event_loop.run_until_complete(setup())

../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/asyncio/base_events.py:649: in run_until_complete
    return future.result()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:325: in setup
    res = await gen_obj.__anext__()  # type: ignore[union-attr]
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:157: in recorded_test
    test_id, recording_id, variables = start_proxy_session()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:200: in start_proxy_session
    recording_id, variables = start_record_or_playback(test_id)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

test_id = 'sdk/identity/azure-identity/tests/recordings/test_imds_credential.pyTestImdstest_system_assigned[get_token]'

    def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
        """Sends a request to begin recording or playing back the provided test.
    
        This returns a tuple, (a, b), where a is the recording ID of the test and b is the `variables` dictionary that maps
        test variables to values. If no variable dictionary was stored when the test was recorded, b is an empty dictionary.
        """
        variables = {}  # this stores a dictionary of test variable values that could have been stored with a recording
    
        json_payload = {"x-recording-file": test_id}
        assets_json = get_recording_assets(test_id)
        if assets_json:
            json_payload["x-recording-assets-file"] = assets_json
    
        encoded_payload = json.dumps(json_payload).encode("utf-8")
        http_client = get_http_client()
    
        if is_live():
            result = http_client.request(
                method="POST",
                url=RECORDING_START_URL,
                body=encoded_payload,
            )
            if result
... [The stack trace has been truncated as it exceeded the maximum allowed size. Please refer to the complete log available in the Test Run attachments for full details.]

request = <SubRequest 'recorded_test' for <Function test_system_assigned[get_token_info]>>
kwargs = {'test_proxy': None}, func = <function recorded_test at 0x7fd5b87404c0>
event_loop_fixture_id = 'event_loop'
setup = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup at 0x7fd5b55a65f0>
finalizer = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer at 0x7fd5b55a6e60>

    @functools.wraps(fixture)
    def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
        func = _perhaps_rebind_fixture_func(fixture, request.instance)
        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
            request, func
        )
        event_loop = request.getfixturevalue(event_loop_fixture_id)
        kwargs.pop(event_loop_fixture_id, None)
        gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
    
        async def setup():
            res = await gen_obj.__anext__()  # type: ignore[union-attr]
            return res
    
        def finalizer() -> None:
            """Yield again, to finalize."""
    
            async def async_finalizer() -> None:
                try:
                    await gen_obj.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    pass
                else:
                    msg = "Async generator fixture didn't stop."
                    msg += "Yield only once."
                    raise ValueError(msg)
    
            event_loop.run_until_complete(async_finalizer())
    
>       result = event_loop.run_until_complete(setup())

../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/asyncio/base_events.py:649: in run_until_complete
    return future.result()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:325: in setup
    res = await gen_obj.__anext__()  # type: ignore[union-attr]
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:157: in recorded_test
    test_id, recording_id, variables = start_proxy_session()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:200: in start_proxy_session
    recording_id, variables = start_record_or_playback(test_id)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

test_id = 'sdk/identity/azure-identity/tests/recordings/test_imds_credential.pyTestImdstest_system_assigned[get_token_info]'

    def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
        """Sends a request to begin recording or playing back the provided test.
    
        This returns a tuple, (a, b), where a is the recording ID of the test and b is the `variables` dictionary that maps
        test variables to values. If no variable dictionary was stored when the test was recorded, b is an empty dictionary.
        """
        variables = {}  # this stores a dictionary of test variable values that could have been stored with a recording
    
        json_payload = {"x-recording-file": test_id}
        assets_json = get_recording_assets(test_id)
        if assets_json:
            json_payload["x-recording-assets-file"] = assets_json
    
        encoded_payload = json.dumps(json_payload).encode("utf-8")
        http_client = get_http_client()
    
        if is_live():
            result = http_client.request(
                method="POST",
                url=RECORDING_START_URL,
                body=encoded_payload,
            )
           
... [The stack trace has been truncated as it exceeded the maximum allowed size. Please refer to the complete log available in the Test Run attachments for full details.]

request = <SubRequest 'recorded_test' for <Function test_system_assigned_tenant_id[get_token]>>
kwargs = {'test_proxy': None}, func = <function recorded_test at 0x7fd5b87404c0>
event_loop_fixture_id = 'event_loop'
setup = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup at 0x7fd5b55a6ef0>
finalizer = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer at 0x7fd5b69b0d30>

    @functools.wraps(fixture)
    def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
        func = _perhaps_rebind_fixture_func(fixture, request.instance)
        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
            request, func
        )
        event_loop = request.getfixturevalue(event_loop_fixture_id)
        kwargs.pop(event_loop_fixture_id, None)
        gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
    
        async def setup():
            res = await gen_obj.__anext__()  # type: ignore[union-attr]
            return res
    
        def finalizer() -> None:
            """Yield again, to finalize."""
    
            async def async_finalizer() -> None:
                try:
                    await gen_obj.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    pass
                else:
                    msg = "Async generator fixture didn't stop."
                    msg += "Yield only once."
                    raise ValueError(msg)
    
            event_loop.run_until_complete(async_finalizer())
    
>       result = event_loop.run_until_complete(setup())

../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/asyncio/base_events.py:649: in run_until_complete
    return future.result()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:325: in setup
    res = await gen_obj.__anext__()  # type: ignore[union-attr]
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:157: in recorded_test
    test_id, recording_id, variables = start_proxy_session()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:200: in start_proxy_session
    recording_id, variables = start_record_or_playback(test_id)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

test_id = 'sdk/identity/azure-identity/tests/recordings/test_imds_credential.pyTestImdstest_system_assigned_tenant_id[get_token]'

    def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
        """Sends a request to begin recording or playing back the provided test.
    
        This returns a tuple, (a, b), where a is the recording ID of the test and b is the `variables` dictionary that maps
        test variables to values. If no variable dictionary was stored when the test was recorded, b is an empty dictionary.
        """
        variables = {}  # this stores a dictionary of test variable values that could have been stored with a recording
    
        json_payload = {"x-recording-file": test_id}
        assets_json = get_recording_assets(test_id)
        if assets_json:
            json_payload["x-recording-assets-file"] = assets_json
    
        encoded_payload = json.dumps(json_payload).encode("utf-8")
        http_client = get_http_client()
    
        if is_live():
            result = http_client.request(
                method="POST",
                url=RECORDING_START_URL,
                body=encoded_payload,
            )
 
... [The stack trace has been truncated as it exceeded the maximum allowed size. Please refer to the complete log available in the Test Run attachments for full details.]

request = <SubRequest 'recorded_test' for <Function test_system_assigned_tenant_id[get_token_info]>>
kwargs = {'test_proxy': None}, func = <function recorded_test at 0x7fd5b87404c0>
event_loop_fixture_id = 'event_loop'
setup = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.setup at 0x7fd5b69b3640>
finalizer = <function _wrap_asyncgen_fixture.<locals>._asyncgen_fixture_wrapper.<locals>.finalizer at 0x7fd5b55a6950>

    @functools.wraps(fixture)
    def _asyncgen_fixture_wrapper(request: FixtureRequest, **kwargs: Any):
        func = _perhaps_rebind_fixture_func(fixture, request.instance)
        event_loop_fixture_id = _get_event_loop_fixture_id_for_async_fixture(
            request, func
        )
        event_loop = request.getfixturevalue(event_loop_fixture_id)
        kwargs.pop(event_loop_fixture_id, None)
        gen_obj = func(**_add_kwargs(func, kwargs, event_loop, request))
    
        async def setup():
            res = await gen_obj.__anext__()  # type: ignore[union-attr]
            return res
    
        def finalizer() -> None:
            """Yield again, to finalize."""
    
            async def async_finalizer() -> None:
                try:
                    await gen_obj.__anext__()  # type: ignore[union-attr]
                except StopAsyncIteration:
                    pass
                else:
                    msg = "Async generator fixture didn't stop."
                    msg += "Yield only once."
                    raise ValueError(msg)
    
            event_loop.run_until_complete(async_finalizer())
    
>       result = event_loop.run_until_complete(setup())

../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:343: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/opt/hostedtoolcache/Python/3.10.20/x64/lib/python3.10/asyncio/base_events.py:649: in run_until_complete
    return future.result()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/pytest_asyncio/plugin.py:325: in setup
    res = await gen_obj.__anext__()  # type: ignore[union-attr]
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:157: in recorded_test
    test_id, recording_id, variables = start_proxy_session()
../../../.venv/azure-identity/.venv_mindependency/lib/python3.10/site-packages/devtools_testutils/proxy_fixtures.py:200: in start_proxy_session
    recording_id, variables = start_record_or_playback(test_id)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

test_id = 'sdk/identity/azure-identity/tests/recordings/test_imds_credential.pyTestImdstest_system_assigned_tenant_id[get_token_info]'

    def start_record_or_playback(test_id: str) -> "Tuple[str, Dict[str, str]]":
        """Sends a request to begin recording or playing back the provided test.
    
        This returns a tuple, (a, b), where a is the recording ID of the test and b is the `variables` dictionary that maps
        test variables to values. If no variable dictionary was stored when the test was recorded, b is an empty dictionary.
        """
        variables = {}  # this stores a dictionary of test variable values that could have been stored with a recording
    
        json_payload = {"x-recording-file": test_id}
        assets_json = get_recording_assets(test_id)
        if assets_json:
            json_payload["x-recording-assets-file"] = assets_json
    
        encoded_payload = json.dumps(json_payload).encode("utf-8")
        http_client = get_http_client()
    
        if is_live():
            result = http_client.request(
                method="POST",
                url=RECORDING_START_URL,
                body=encoded_payload,
     
... [The stack trace has been truncated as it exceeded the maximum allowed size. Please refer to the complete log available in the Test Run attachments for full details.]