Class reply

test/mitmproxy/proxy/tutils.py:307–359

class reply(events.Event):
    args: tuple[Any, ...]
    to: commands.Command | type[commands.Command] | int
    side_effect: Callable[[Any], Any]

    def __init__(
        self,
        *args,
        to: commands.Command | type[commands.Command] | int = -1,
        side_effect: Callable[[Any], None] = lambda x: None,
    ):
        """Utility method to reply to the latest hook in playbooks."""
        assert not args or not isinstance(args[0], commands.Command)
        self.args = args
        self.to = to
        self.side_effect = side_effect

    def playbook_eval(self, playbook: Playbook) -> events.CommandCompleted:
        if isinstance(self.to, int):
            expected = playbook.expected[: playbook.expected.index(self)]
            assert abs(self.to) < len(expected)
            to = expected[self.to]
            if not isinstance(to, commands.Command):
                raise AssertionError(f"There is no command at offset {self.to}: {to}")
            else:
                self.to = to
        elif isinstance(self.to, type):
            for cmd in reversed(playbook.actual):
                if isinstance(cmd, self.to):
                    assert isinstance(cmd, commands.Command)
                    self.to = cmd
                    break
            else:
                raise AssertionError(f"There is no command of type {self.to}.")
        for cmd in reversed(playbook.actual):
            if eq(self.to, cmd):
                self.to = cmd
                break
        else:
            raise AssertionError(f"Expected command {self.to} did not occur.")

        assert isinstance(self.to, commands.Command)
        if isinstance(self.to, commands.StartHook):
            self.side_effect(*self.to.args())
            reply_cls = events.HookCompleted
        else:
            self.side_effect(self.to)
            reply_cls = command_reply_subclasses[type(self.to)]
        try:
            inst = reply_cls(self.to, *self.args)
        except TypeError as e:
            raise ValueError(f"Cannot instantiate {reply_cls.__name__}: {e}")
        return inst


T = TypeVar("T")