[PATCH v2 6/6] tests: hwsim: add hwsim relay code

[Johannes Berg]

On Sat, 2026-09-12 at 00:59 +0200, Johannes Berg wrote:
> 
> +class RelayLayer:
> +    """
> +    Base layer for frame interception - each layer has its own
> +    .intercept() like the HwsimRelay() itself.
> +
> +    Used as "async with XyzLayer(...) as layer:" just like
> +    the relay itself.
> +    """
> +
> +    def __init__(self, upstream):
> +        self._upstream = upstream
> +        self._downstream = self._default
> +
> +    def __enter__(self):
> +        self._prev = self._upstream.intercept
> +        self._upstream.intercept = self._run
> +        return self
> +
> +    def __exit__(self, exc_type, exc_value, traceback):
> +        self._upstream.intercept = self._prev
> +
> +    @property
> +    def intercept(self):
> +        return self._downstream
> +
> +    @intercept.setter
> +    def intercept(self, fn):
> +        self._downstream = fn
> +
> +    def _default(self, src, freq, frame):
> +        return frame
> +
> +    def _run(self, src, freq, frame):
> +        raise NotImplementedError
> +

...

OK I've been revisiting this (in part due to some bugs), and I'm really
not happy with the API .. it works, but it's messy, and difficult to
understand.

So I think I'm going to change it, so that you can either use the
intercept function here or install a layer later, but you have to
perfectly stack layers.

This supported kind of (temporary) forks in the frame flow graph here
and I thought that was more powerful (and it probably _is_), but I've
myself lost track of it a few times now, so ... no good.

So handwaving over the HwsimRelay changes, which will allow only the
intercept function _or_ a layer attached, we'll get:

class RelayLayer:
    """
    Base implementation for frame interception layers.
    Implement subclasses of this that have ._run().

    Used as

        with XyzLayer(prev, ...) as layer:
            ...

    where where prev is either another layer or the relay.
    """

    def __init__(self, upstream):
        assert isinstance(upstream, (RelayLayer, HwsimRelay))
        self._upstream = upstream
        self._next = None

    def __enter__(self):
        assert self._upstream._next is None
        self._upstream._next = self
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self._upstream._next = None

    def _call_next(self, src, freq, frame):
        if self._next:
            frame = self._next._run(src, freq, frame)
        return frame

    def _run(self, src, freq, frame):
        """
        Example:

        if should_drop_frame:
            return None

        # could modify / drop frame here

        frame = self._call_next(src, freq, frame)

        # could also modify / drop frame here, but
        # note it might have been dropped (be None)
        # by the next layer

        return frame
        """
        raise NotImplementedError


Yeah this way you can't build arbitrary layer graphs, but that's a
feature, not a bug ...

It also means the tests do things in the right order, i.e. layers
constructed first are passed through first:

  with Layer1(relay, ...) as l1, \
       Layer2(l1, ...) as l2:
    # in this block relay passes to l1, passes to l2

johannes