proxy package#



Module contents#

⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on Network monitoring, controls & Application development, testing, debugging.

  1. 2013-present by Abhinav Singh and contributors.


BSD, see LICENSE for more details.

class proxy.Proxy(input_args: Optional[List[str]] = None, **opts: Any)[source]#

Bases: object

Proxy is a context manager to control library core.

By default, AcceptorPool is started with HttpProtocolHandler work class. By definition, it expects HTTP traffic to flow between clients and server.

In --threadless mode and without --local-executor, a ThreadlessPool is also started. Executor pool receives newly accepted work by Acceptor and creates an instance of work class for processing the received work.

In --threadless mode and with --local-executor 0, acceptors will start a companion thread to handle accepted client connections.

Optionally, Proxy class also initializes the EventManager. A multi-process safe pubsub system which can be used to build various patterns for message sharing and/or signaling.

_delete_pid_file() None[source]#
_delete_port_file() None[source]#
static _handle_exit_signal(signum: int, _frame: Any) None[source]#
_handle_siginfo(_signum: int, _frame: Any) None[source]#
_register_signals() None[source]#
_write_pid_file() None[source]#
_write_port_file() None[source]#
property remote_executors_enabled: bool#
setup() None[source]#
shutdown() None[source]#
class proxy.TestCase(methodName='runTest')[source]#


Base TestCase class that automatically setup and tear down

DEFAULT_PROXY_PY_STARTUP_FLAGS = ['--hostname', '', '--port', '0', '--num-workers', '1', '--num-acceptors', '1']#
INPUT_ARGS: Optional[List[str]] = None#
PROXY: Optional[proxy.proxy.Proxy] = None#
run(result: Optional[unittest.result.TestResult] = None) Any[source]#
classmethod setUpClass() None[source]#

Hook method for setting up class fixture before running tests in the class.

classmethod tearDownClass() None[source]#

Hook method for deconstructing the class fixture after running all tests in the class.

vcr() Generator[None, None, None][source]#
static wait_for_server(proxy_port: int, wait_for_seconds: float = 10.0) None[source]#

Wait for server to come up.

proxy.entry_point() None[source]#
proxy.main(**opts: Any) None[source]#
proxy.sleep_loop() None[source]#