Provide `test-proxy` redirect fixture that can apply to `httpx` transport

[scbedd]

Right now, we can support redirect of AiohttpTransport, TrioTransport (naming*), and RequestsTransport. We need to add a fourth one for HttpxTransport in general.

I had a chat with @mccoyp to discuss how we wanted to resolve this going forward. Effectively, this is what we have right now:

• recorded_test pytest fixture
  • If run from sync function, patches RequestsTransport.send
  • If run from async function, patches AiohttpTransport.send
• @recorded_by_proxy decorator
  • patches RequestsTransport.send
• @recorded_by_proxy_async decorator
  • patches AiohttpTransport.Send

The thing is, our implementations across these locations is not shared. We definitely think that we should support individual, customizable patches as stacks, eg:

@recorded_by_proxy
@recorded_by_proxy_httpx
def test_function_that_calls_both_requests_and_httpx_sync()
   ...

Before we do that, we need to refactor to share implementation. That way only the very top level definition of the name of the decorator or feature is passed, and the same common implementation will be used regardless of which one is activated.

[github-actions]

@kashifkhan @xiangyan99

[scbedd]

@mccoyp Darren tried to use stacked decorators and I'm confirming my repro of that not working. Because the decorators apply outside in, and we're expecting a variables dictionary being passed into the called test func, the stacking goes all wonky.

    @servicePreparer()
    @recorded_by_proxy
    @recorded_by_proxy_httpx
   def test_something()
     pass

This melts. I'm imagining something like @recorded_by(Requests, Httpx) or something like that, but I don't think decorator syntax really allows this.

[mccoyp]

Got it... not an unexpected outcome. I guess our options are to either make these decorators compatible with stacking by adjusting their expected inputs/outputs, or have something like a dedicated decorator that patches all transports. That's a good point though about potentially passing inputs through the decorator -- it might be possible, but we'd have to investigate that

[scbedd]

Got it... not an unexpected outcome. I guess our options are to either make these decorators compatible with stacking by adjusting their expected inputs/outputs, or have something like a dedicated decorator that patches all transports. That's a good point though about potentially passing inputs through the decorator -- it might be possible, but we'd have to investigate that

Half-copilot half me slamming face into keyboard:

from functools import wraps
from contextlib import ExitStack

# helper: turn decorator args into a normalized list of (owner, attr_name)
def _normalize_transport_specs(specs):
    norm = []
    for spec in specs:
        if isinstance(spec, tuple) and len(spec) == 2:
            owner, attr = spec
            norm.append((owner, attr))
        else:
            # If a class was provided, assume attribute name "send"
            # (Safer than accepting a bare function object like RequestsTransport.send,
            # because assignment requires knowing the owner and attribute name.)
            norm.append((spec, "send"))
    return norm

def recorded_by_proxy(*maybe_specs):
    """
    Usages:
      @recorded_by_proxy
      def test(...): ...

      @recorded_by_proxy((RequestsTransport, "send"), (HttpxTransport, "send"))
      def test(...): ...
    """

    # Bare decorator usage: @recorded_by_proxy
    if len(maybe_specs) == 1 and callable(maybe_specs[0]):
        test_func = maybe_specs[0]
        transports = _normalize_transport_specs([(RequestsTransport, "send")])
        return _make_proxy_decorator(transports)(test_func)

    # Parameterized decorator usage: @recorded_by_proxy(...)
    transports = _normalize_transport_specs(maybe_specs or [(RequestsTransport, "send")])
    return _make_proxy_decorator(transports)

def _make_proxy_decorator(transports):
    def _decorator(test_func: "Callable"):
        @wraps(test_func)
        def record_wrap(*args, **kwargs):
            # ---- your existing trimming/early-exit logic ----
            trimmed_kwargs = {k: v for k, v in kwargs.items()}
            trim_kwargs_from_test_function(test_func, trimmed_kwargs)

            if is_live_and_not_recording():
                return test_func(*args, **trimmed_kwargs)

            test_id = get_test_id()
            recording_id, variables = start_record_or_playback(test_id)

            def transform_args_inner(*call_args, **call_kwargs):
                copied_positional_args = list(call_args)
                request = copied_positional_args[1]
                transform_request(request, recording_id)
                return tuple(copied_positional_args), call_kwargs

            # Build a wrapper factory so each patched method closes over its own original
            def make_combined_call(original_transport_func):
                def combined_call(*call_args, **call_kwargs):
                    adjusted_args, adjusted_kwargs = transform_args_inner(*call_args, **call_kwargs)
                    result = original_transport_func(*adjusted_args, **adjusted_kwargs)

                    # rewrite request.url to the original upstream for LROs, etc.
                    parsed_result = url_parse.urlparse(result.request.url)
                    upstream_uri = url_parse.urlparse(result.request.headers["x-recording-upstream-base-uri"])
                    upstream_uri_dict = {"scheme": upstream_uri.scheme, "netloc": upstream_uri.netloc}
                    original_target = parsed_result._replace(**upstream_uri_dict).geturl()
                    result.request.url = original_target
                    return result
                return combined_call

            # Patch multiple transports and ensure restoration
            test_variables = None
            test_run = False
            with ExitStack() as stack:
                originals = []
                # monkeypatch all requested transports
                for owner, name in transports:
                    original = getattr(owner, name)
                    setattr(owner, name, make_combined_call(original))
                    originals.append((owner, name, original))

                try:
                    try:
                        test_variables = test_func(*args, variables=variables, **trimmed_kwargs)
                        test_run = True
                    except TypeError as error:
                        if "unexpected keyword argument" in str(error) and "variables" in str(error):
                            logger = logging.getLogger()
                            logger.info(
                                "This test can't accept variables as input. "
                                "Accept `**kwargs` and/or a `variables` parameter to use recorded variables."
                            )
                        else:
                            raise

                    if not test_run:
                        test_variables = test_func(*args, **trimmed_kwargs)

                except ResourceNotFoundError as error:
                    error_body = ContentDecodePolicy.deserialize_from_http_generics(error.response)
                    troubleshoot = (
                        "Playback failure -- for help resolving, see https://aka.ms/azsdk/python/test-proxy/troubleshoot."
                    )
                    message = error_body.get("message") or error_body.get("Message")
                    error_with_message = ResourceNotFoundError(
                        message=f"{troubleshoot} Error details:\n{message}",
                        response=error.response,
                    )
                    raise error_with_message from error

                finally:
                    # restore in reverse order
                    for owner, name, original in reversed(originals):
                        setattr(owner, name, original)
                    stop_record_or_playback(test_id, recording_id, test_variables)

            return test_variables
        return record_wrap
    return _decorator

something like this