Skip to content

Commit

Permalink
[Test Proxy] Fix error with recorded_test fixture (Azure#26073)
Browse files Browse the repository at this point in the history
In the case of live testing with recording disabled,
the 'recorded_test` pytest fixture tries to unpack a NoneType
object. Here, some small refactoring is done in order
to handle this case.

Signed-off-by: Paul Van Eck <[email protected]>
  • Loading branch information
pvaneck authored Sep 14, 2022
1 parent 3e08990 commit 4760e34
Showing 1 changed file with 18 additions and 18 deletions.
36 changes: 18 additions & 18 deletions tools/azure-sdk-tools/devtools_testutils/proxy_fixtures.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ def environment_variables(test_proxy: None) -> EnvironmentVariableSanitizer:


@pytest.fixture
async def recorded_test(test_proxy: None, request: "FixtureRequest") -> "Dict[str, Any]":
async def recorded_test(test_proxy: None, request: "FixtureRequest") -> "Optional[Dict[str, Any]]":
"""Fixture that redirects network requests to target the azure-sdk-tools test proxy.
Use with recorded tests. For more details and usage examples, refer to
Expand All @@ -125,21 +125,24 @@ async def recorded_test(test_proxy: None, request: "FixtureRequest") -> "Dict[st
:type request: ~pytest.FixtureRequest
:yields: A dictionary containing information relevant to the currently executing test.
If the current test session is live but recording is disabled, yields None.
"""

test_id, recording_id, variables = start_proxy_session()

# True if the function requesting the fixture is an async test
if iscoroutinefunction(request._pyfuncitem.function):
original_transport_func = await redirect_async_traffic(recording_id)
yield {"variables": variables} # yield relevant test info and allow tests to run
restore_async_traffic(original_transport_func, request)
if is_live_and_not_recording():
yield
else:
original_transport_func = redirect_traffic(recording_id)
yield {"variables": variables} # yield relevant test info and allow tests to run
restore_traffic(original_transport_func, request)
test_id, recording_id, variables = start_proxy_session()

stop_record_or_playback(test_id, recording_id, variables)
# True if the function requesting the fixture is an async test
if iscoroutinefunction(request._pyfuncitem.function):
original_transport_func = await redirect_async_traffic(recording_id)
yield {"variables": variables} # yield relevant test info and allow tests to run
restore_async_traffic(original_transport_func, request)
else:
original_transport_func = redirect_traffic(recording_id)
yield {"variables": variables} # yield relevant test info and allow tests to run
restore_traffic(original_transport_func, request)

stop_record_or_playback(test_id, recording_id, variables)


@pytest.fixture
Expand All @@ -162,16 +165,13 @@ def variable_recorder(recorded_test: "Dict[str, Any]") -> VariableRecorder:
# ----------HELPERS----------


def start_proxy_session() -> "Optional[Tuple[str, str, Dict[str, str]]]":
def start_proxy_session() -> "Tuple[str, str, Dict[str, str]]":
"""Begins a playback or recording session and returns the current test ID, recording ID, and recorded variables.
:returns: A tuple, (a, b, c), where a is the test ID, b is the recording ID, and c is the `variables` dictionary
that maps test variables to string values. If no variable dictionary was stored when the test was recorded, c is
an empty dictionary. If the current test session is live but recording is disabled, this returns None.
an empty dictionary.
"""
if is_live_and_not_recording():
return

test_id = get_test_id()
recording_id, variables = start_record_or_playback(test_id)
return (test_id, recording_id, variables)
Expand Down

0 comments on commit 4760e34

Please sign in to comment.