proxy.py/tests/test_assertions.py

27 lines
700 B
Python
Raw Permalink Normal View History

Async `get_events`, `handle_event`, `handle_readables`, `handle_writables` (#769) * Asynchronous `handle_event` and `LocalExecutor` thread * Bail out on first task completion * mypy * Add `helper/benchmark.sh` and fix threaded which must now use asyncio (reduced performance of threaded) * Print open file diff from `benchmark.sh` * Add `--local-executor` flag, disabled by default for now until tests are updated * Async `handle_readables` and `handle_writables` for `HttpProtocolHandlerPlugin` interface (doesnt impact proxy/web plugins for now) * Async `get_events` * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Address tests after async changes * mypy and flake8 * spelldoc * `check.py` and trailing comma * Rename to `_assertions.py` * Add missing `pytest-mock` and `pytest-asyncio` deps * Add `pytest-mock` to `pylint` deps * Correct use of `parameterize` and add `PT007` to flake8 ignores * Fix mypy hints broken for `< Python3.9` * Remove usage of `asynccontextmanager` which is not available for all Python versions that `proxy.py` supports * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Fix for pre-python-3.9 versions * `AsyncTask` apis `set_name` and `get_name` are not available on all supported versions * Install setuptools via `lib-dep` until we recommend editable install * Deprecate support for `Python 3.6` * Use recommendation suggested here https://github.com/abhinavsingh/proxy.py/pull/769\#discussion_r753840929 * Address recommendation here https://github.com/abhinavsingh/proxy.py/pull/769\#discussion_r753841906 * Make `Threadless` agnostic of `multiprocessing.Process` * Acceptors must dispatch to local executor in non-blocking fashion * No daemon for executor processes and fix shutdown logic * Only return fds from `_selected_events` not all events data * Refactor logic * Prefix private methods with `_` * `work_queue` and not `client_queue` * Turn `Threadless` into an abstract executor. Introduce `RemoteExecutor` * Make `LocalExecutor` agnostic of `threading.Thread` * `LocalExecutor` now implements `Threadless` * `get_events` and `get_descriptors` now must return int and not sock. `Threadless` now avoids repeated register/unregister and instead make use of `selectors.modify` * Fix `main` tests * Apply suggestions from code review Co-authored-by: Sviatoslav Sydorenko <wk@sydorenko.org.ua> * Apply code review recommendations manually * [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci * Revert back `Any` and use `addr or None` * Address `flake8` * Update tests to use `fileno` * Fix doc build * Fix doc spell, use tear down and not teardown * Doc updates * Add back support for `Python 3.6` * Acceptors dont need loop initialization * On Python 3.6 `asyncio.new_event_loop()` is necessary * Make doc happy * `--threaded` needs a new event loop for 3.7 too * Always use `asyncio.new_event_loop()` for threaded mode Added e2e integration tests (subprocess & curl) for all modes. * Lint fixes Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com> Co-authored-by: Sviatoslav Sydorenko <wk@sydorenko.org.ua>
2021-11-23 09:32:00 +00:00
# -*- coding: utf-8 -*-
"""
proxy.py
~~~~~~~~
Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
Network monitoring, controls & Application development, testing, debugging.
:copyright: (c) 2013-present by Abhinav Singh and contributors.
:license: BSD, see LICENSE for more details.
"""
from typing import Any
class Assertions:
def assertTrue(self, obj: Any) -> None:
assert obj
def assertFalse(self, obj: Any) -> None:
assert not obj
def assertEqual(self, obj1: Any, obj2: Any) -> None:
assert obj1 == obj2
def assertNotEqual(self, obj1: Any, obj2: Any) -> None:
assert obj1 != obj2