diff --git a/MitmLibrary/__init__.py b/MitmLibrary/__init__.py
index 43bdc56..ec88897 100644
--- a/MitmLibrary/__init__.py
+++ b/MitmLibrary/__init__.py
@@ -12,7 +12,8 @@
applications in a more realistic and controlled environment.
"""
-from typing import Any, Dict, List, Optional, Sequence, Union
+from collections.abc import Sequence
+from typing import Any
from mitmproxy.tools import dump
from robot.api import logger
@@ -202,15 +203,15 @@ def __init__(self) -> None:
"""
self.controller: ProxyController = ProxyController()
self.registry: RuleRegistry = RuleRegistry()
- self.interceptor: Optional[Interceptor] = None
- self.recorder: Optional[FlowRecorder] = None
+ self.interceptor: Interceptor | None = None
+ self.recorder: FlowRecorder | None = None
self.log_to_console: bool = True
# Robot Framework calls close() on this when the suite that imported the library
# ends, which releases the port even if the suite never stopped the proxy itself.
self.ROBOT_LIBRARY_LISTENER: LibraryListener = LibraryListener(self.controller)
@property
- def proxy_master(self) -> Optional[dump.DumpMaster]:
+ def proxy_master(self) -> dump.DumpMaster | None:
"""The running mitmproxy master, or None when no proxy is running."""
return self.controller.master
@@ -239,14 +240,14 @@ def start_mitm_proxy(
self,
listen_host: str = "127.0.0.1",
listen_port: int = 8080,
- certificates_directory: Optional[str] = None,
+ certificates_directory: str | None = None,
ssl_insecure: bool = False,
log_to_console: bool = True,
record: bool = False,
record_limit: int = DEFAULT_LIMIT,
record_body_limit: int = DEFAULT_BODY_LIMIT,
- mode: Optional[Union[str, List[str]]] = None,
- proxy_auth: Optional[str] = None,
+ mode: str | list[str] | None = None,
+ proxy_auth: str | None = None,
) -> None:
"""
Starts a proxy at the given host and port.
@@ -309,7 +310,7 @@ def _build_addons(self, master: dump.DumpMaster) -> Sequence[Any]:
there afterwards. Only the addon reading them is rebuilt.
"""
self.interceptor = Interceptor(self.registry, self.log_to_console)
- addons: List[Any] = [self.interceptor]
+ addons: list[Any] = [self.interceptor]
if self.recorder is not None:
addons.append(self.recorder)
return addons
@@ -362,7 +363,7 @@ def block_requests(
url: str,
mode: BlockMode = BlockMode.RESPOND,
status_code: int = 403,
- body: Optional[str] = None,
+ body: str | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
times: int = 0,
@@ -397,8 +398,8 @@ def set_response(
alias: str,
url: str,
status_code: int = 200,
- headers: Optional[Dict[str, str]] = None,
- body: Optional[str] = None,
+ headers: dict[str, str] | None = None,
+ body: str | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
times: int = 0,
@@ -510,8 +511,8 @@ def set_response_headers(
self,
alias: str,
url: str,
- headers: Optional[Dict[str, str]] = None,
- remove: Optional[List[str]] = None,
+ headers: dict[str, str] | None = None,
+ remove: list[str] | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
times: int = 0,
@@ -574,8 +575,8 @@ def set_request_headers(
self,
alias: str,
url: str,
- headers: Optional[Dict[str, str]] = None,
- remove: Optional[List[str]] = None,
+ headers: dict[str, str] | None = None,
+ remove: list[str] | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
times: int = 0,
@@ -671,8 +672,8 @@ def redirect_requests_to_host(
alias: str,
url: str,
host: str,
- port: Optional[int] = None,
- scheme: Optional[str] = None,
+ port: int | None = None,
+ scheme: str | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
times: int = 0,
@@ -743,7 +744,7 @@ def simulate_truncated_response(
self,
alias: str,
url: str,
- keep_bytes: Optional[int] = None,
+ keep_bytes: int | None = None,
keep_fraction: float = 0.5,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
@@ -806,7 +807,7 @@ def clear_all_rules(self) -> None:
self._require_registry().clear()
@keyword
- def get_proxy_rules(self) -> List[DotDict]:
+ def get_proxy_rules(self) -> list[DotDict]:
"""Returns the loaded rules, in the order they are applied.
Each rule is a dictionary with at least `alias`, `url`, `match`, `method`,
@@ -879,10 +880,10 @@ def clear_recorded_requests(self) -> None:
@keyword
def get_recorded_requests(
self,
- url: Optional[str] = None,
+ url: str | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
- ) -> List[DotDict]:
+ ) -> list[DotDict]:
"""Returns the recorded requests, oldest first.
Each request is a dictionary with `method`, `url`, `host`, `path`, `query`,
@@ -904,7 +905,7 @@ def get_recorded_requests(
@keyword
def get_request_count(
self,
- url: Optional[str] = None,
+ url: str | None = None,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
) -> int:
@@ -928,8 +929,8 @@ def request_should_have_been_made(
url: str,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
- times: Optional[int] = None,
- msg: Optional[str] = None,
+ times: int | None = None,
+ msg: str | None = None,
) -> None:
"""Fails unless a matching request was recorded.
@@ -972,7 +973,7 @@ def request_should_not_have_been_made(
url: str,
method: str = ANY_METHOD,
match: MatchMode = MatchMode.SUBSTRING,
- msg: Optional[str] = None,
+ msg: str | None = None,
) -> None:
"""Fails when a matching request was recorded.
@@ -1008,7 +1009,7 @@ def wait_until_request_is_made(
match: MatchMode = MatchMode.SUBSTRING,
timeout: str = "10s",
count: int = 1,
- ) -> List[DotDict]:
+ ) -> list[DotDict]:
"""Waits until matching requests have been recorded, and returns them.
For traffic a test does not trigger directly, such as a request a page makes
@@ -1041,7 +1042,7 @@ def _require_recorder(self) -> FlowRecorder:
@not_keyword
def _recording_matcher(
- self, url: Optional[str], method: str, match: MatchMode
+ self, url: str | None, method: str, match: MatchMode
) -> UrlMatcher:
"""Builds the matcher the recording keywords filter with.
diff --git a/MitmLibrary/async_loop_thread.py b/MitmLibrary/async_loop_thread.py
index 71addc5..39d15c5 100644
--- a/MitmLibrary/async_loop_thread.py
+++ b/MitmLibrary/async_loop_thread.py
@@ -33,7 +33,7 @@ def run(self) -> None:
asyncio.set_event_loop(self.loop)
try:
self.loop.run_forever()
- except Exception as e: # pylint: disable=broad-exception-caught
+ except Exception as e: # noqa: BLE001 - best-effort loop shutdown, must not raise
print(f"Async loop thread error: {e}") # Log the error message
def stop(self, timeout: float = 5) -> None:
@@ -51,5 +51,5 @@ def stop(self, timeout: float = 5) -> None:
self.join(timeout=timeout)
try:
self.loop.close()
- except Exception as error: # pylint: disable=broad-exception-caught
+ except Exception as error: # noqa: BLE001 - best-effort shutdown, must not raise
print(f"Async loop thread could not be closed: {error}")
diff --git a/MitmLibrary/failures.py b/MitmLibrary/failures.py
index f350c13..c14cd7f 100644
--- a/MitmLibrary/failures.py
+++ b/MitmLibrary/failures.py
@@ -14,7 +14,7 @@
import asyncio
from dataclasses import dataclass
-from typing import Any, Dict, Optional
+from typing import Any
from mitmproxy import http
from robot.api import logger
@@ -46,7 +46,7 @@ async def apply_async(self, flow: http.HTTPFlow) -> bool:
kill_flow(flow)
return True
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {"type": "timeout", "hold": self.hold, "hold_seconds": self.hold_seconds}
@@ -61,7 +61,7 @@ class TruncateAction(Action):
with the response at all.
"""
- keep_bytes: Optional[int] = None
+ keep_bytes: int | None = None
keep_fraction: float = 0.5
phase = Phase.RESPONSE
@@ -92,7 +92,7 @@ def _keep(self, length: int) -> int:
return max(0, self.keep_bytes)
return max(0, int(length * self.keep_fraction))
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {
"type": "truncate",
"keep_bytes": self.keep_bytes,
diff --git a/MitmLibrary/interceptor.py b/MitmLibrary/interceptor.py
index 30f4a99..feac4ba 100644
--- a/MitmLibrary/interceptor.py
+++ b/MitmLibrary/interceptor.py
@@ -9,7 +9,6 @@
only decides *when* they run, not *what* they do.
"""
-from typing import Optional
from mitmproxy import http
from robot.api import logger
@@ -69,7 +68,7 @@ def _log(self, rule: Rule, flow: http.HTTPFlow) -> None:
)
-def _method(flow: http.HTTPFlow) -> Optional[str]:
+def _method(flow: http.HTTPFlow) -> str | None:
"""The request method, or None when the flow does not report one."""
method = getattr(flow.request, "method", None)
return method if isinstance(method, str) else None
diff --git a/MitmLibrary/matching.py b/MitmLibrary/matching.py
index 5b90d20..f9f9737 100644
--- a/MitmLibrary/matching.py
+++ b/MitmLibrary/matching.py
@@ -11,7 +11,7 @@
import re
from dataclasses import dataclass, field
from enum import Enum
-from typing import Optional, Pattern
+from re import Pattern
ANY_METHOD = "ANY"
@@ -47,7 +47,7 @@ class UrlMatcher:
pattern: str
mode: MatchMode = MatchMode.SUBSTRING
method: str = ANY_METHOD
- _regex: Optional[Pattern[str]] = field(default=None, init=False, repr=False)
+ _regex: Pattern[str] | None = field(default=None, init=False, repr=False)
def __post_init__(self) -> None:
# The dataclass is frozen so that a rule cannot change what it matches while the
@@ -56,7 +56,7 @@ def __post_init__(self) -> None:
object.__setattr__(self, "method", (self.method or ANY_METHOD).strip().upper())
object.__setattr__(self, "_regex", self._compile())
- def _compile(self) -> Optional[Pattern[str]]:
+ def _compile(self) -> Pattern[str] | None:
"""Builds the expression for the mode, or None when the mode needs no expression."""
if self.mode is MatchMode.SUBSTRING:
return None
@@ -77,13 +77,13 @@ def matches_url(self, url: str) -> bool:
return self._regex.match(url) is not None
return self._regex.search(url) is not None
- def matches_method(self, method: Optional[str]) -> bool:
+ def matches_method(self, method: str | None) -> bool:
"""Whether the method matches. `ANY` matches everything, including no method."""
if self.method == ANY_METHOD:
return True
return (method or "").upper() == self.method
- def matches(self, url: str, method: Optional[str] = None) -> bool:
+ def matches(self, url: str, method: str | None = None) -> bool:
"""Whether both the url and the method match."""
return self.matches_method(method) and self.matches_url(url)
diff --git a/MitmLibrary/proxy_controller.py b/MitmLibrary/proxy_controller.py
index 596c18f..856c2c4 100644
--- a/MitmLibrary/proxy_controller.py
+++ b/MitmLibrary/proxy_controller.py
@@ -12,8 +12,9 @@
import asyncio
import logging
import time
+from collections.abc import Callable, Sequence
from concurrent.futures import Future, TimeoutError as FutureTimeoutError
-from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple, Union
+from typing import Any
from mitmproxy import options
from mitmproxy.proxy import mode_specs
@@ -48,8 +49,8 @@ class StartupErrorCollector(logging.Handler):
def __init__(self) -> None:
super().__init__(level=logging.ERROR)
- self.messages: List[str] = []
- self.bind_failures: List[str] = []
+ self.messages: list[str] = []
+ self.bind_failures: list[str] = []
def emit(self, record: logging.LogRecord) -> None:
if not record.name.startswith("mitmproxy"):
@@ -68,8 +69,8 @@ class ProxyController:
"""
def __init__(self) -> None:
- self.master: Optional[dump.DumpMaster] = None
- self.future: Optional[Future] = None
+ self.master: dump.DumpMaster | None = None
+ self.future: Future | None = None
self.loop_handler: AsyncLoopThread = AsyncLoopThread()
self.loop_handler.start()
@@ -82,11 +83,11 @@ def start(
self,
listen_host: str,
listen_port: int,
- certificates_directory: Optional[str],
+ certificates_directory: str | None,
ssl_insecure: bool,
addon_factory: AddonFactory,
- mode: Optional[Union[str, Sequence[str]]] = None,
- proxy_auth: Optional[str] = None,
+ mode: str | Sequence[str] | None = None,
+ proxy_auth: str | None = None,
) -> None:
"""Starts the proxy and waits until it is actually listening.
@@ -96,7 +97,7 @@ def start(
Raises RuntimeError if the proxy cannot be started, for example when the port is
already in use.
"""
- option_kwargs: Dict[str, Any] = {
+ option_kwargs: dict[str, Any] = {
"listen_host": listen_host,
"listen_port": listen_port,
"ssl_insecure": ssl_insecure,
@@ -137,7 +138,7 @@ def start(
logging.getLogger().removeHandler(collector)
@staticmethod
- def _parse_modes(mode: Union[str, Sequence[str]]) -> List[str]:
+ def _parse_modes(mode: str | Sequence[str]) -> list[str]:
"""Checks the mode specifications and returns them as mitmproxy wants them.
Parsing here means an unusable specification fails the keyword that gave it,
@@ -235,12 +236,12 @@ def remove_addon(self, addon: Any) -> None:
return
try:
self.master.addons.remove(addon)
- except Exception as error: # pylint: disable=broad-exception-caught
+ except Exception as error: # noqa: BLE001 - best-effort shutdown, must not raise
logger.info(f"The addon was already gone: {error}")
def listen_addresses(
- self, proxy_master: Optional[dump.DumpMaster] = None
- ) -> List[Tuple[Any, ...]]:
+ self, proxy_master: dump.DumpMaster | None = None
+ ) -> list[tuple[Any, ...]]:
"""Returns the addresses the proxy server addon is currently bound to.
Reads them from mitmproxy rather than echoing back the requested host and port,
@@ -275,7 +276,7 @@ def discard(self, wait: bool = True) -> None:
f"The proxy did not shut down within {SHUTDOWN_TIMEOUT} seconds; "
f"its port may still be in use."
)
- except Exception as error: # pylint: disable=broad-exception-caught
+ except Exception as error: # noqa: BLE001 - best-effort shutdown, must not raise
logger.info(f"The proxy stopped with an error: {error}")
self._uninstall_log_handler()
self.master = None
@@ -297,7 +298,7 @@ def _uninstall_log_handler(self) -> None:
return
try:
handler.uninstall()
- except Exception as error: # pylint: disable=broad-exception-caught
+ except Exception as error: # noqa: BLE001 - best-effort shutdown, must not raise
logger.info(f"Could not remove the mitmproxy log handler: {error}")
def _close_servers(self) -> None:
@@ -315,7 +316,7 @@ def _close_servers(self) -> None:
asyncio.run_coroutine_threadsafe(
proxyserver.servers.update([]), self.loop_handler.loop
).result(timeout=SHUTDOWN_TIMEOUT)
- except Exception as error: # pylint: disable=broad-exception-caught
+ except Exception as error: # noqa: BLE001 - best-effort shutdown, must not raise
logger.warn(f"Could not close the proxy servers cleanly: {error}")
def shutdown(self) -> None:
diff --git a/MitmLibrary/recorder.py b/MitmLibrary/recorder.py
index 42d2d01..5557a97 100644
--- a/MitmLibrary/recorder.py
+++ b/MitmLibrary/recorder.py
@@ -18,7 +18,7 @@
import threading
import time
from collections import deque
-from typing import Any, Deque, Dict, List, Optional
+from typing import Any
from mitmproxy import http
from robot.utils import DotDict
@@ -29,7 +29,7 @@
DEFAULT_BODY_LIMIT = 65536
-def _decode(content: Optional[bytes], limit: int) -> Any:
+def _decode(content: bytes | None, limit: int) -> Any:
"""Returns the body as text, shortened to the limit, and whether it was shortened."""
if content is None:
return None, False
@@ -53,7 +53,7 @@ def __init__(
self.body_limit = body_limit
self._lock = threading.Lock()
self._new_entry = threading.Condition(self._lock)
- self._entries: Deque[DotDict] = deque(maxlen=limit)
+ self._entries: deque[DotDict] = deque(maxlen=limit)
self._dropped = 0
@property
@@ -122,7 +122,7 @@ def _describe(self, flow: http.HTTPFlow) -> DotDict:
}
)
- def entries(self, matcher: Optional[UrlMatcher] = None) -> List[DotDict]:
+ def entries(self, matcher: UrlMatcher | None = None) -> list[DotDict]:
"""The recorded requests, oldest first, optionally only the matching ones."""
with self._lock:
recorded = list(self._entries)
@@ -130,7 +130,7 @@ def entries(self, matcher: Optional[UrlMatcher] = None) -> List[DotDict]:
return recorded
return [entry for entry in recorded if matcher.matches(entry.url, entry.method)]
- def count(self, matcher: Optional[UrlMatcher] = None) -> int:
+ def count(self, matcher: UrlMatcher | None = None) -> int:
"""How many recorded requests match."""
return len(self.entries(matcher))
@@ -142,7 +142,7 @@ def clear(self) -> None:
def wait_for(
self, matcher: UrlMatcher, timeout: float, count: int = 1
- ) -> List[DotDict]:
+ ) -> list[DotDict]:
"""Waits until at least `count` recorded requests match, and returns them.
Raises AssertionError when the timeout passes first, so Robot Framework reports it
@@ -187,7 +187,7 @@ def _summary_locked(self) -> str:
)
return described
- def stats(self) -> Dict[str, int]:
+ def stats(self) -> dict[str, int]:
"""How much was recorded and how much was dropped."""
with self._lock:
return {
diff --git a/MitmLibrary/rules.py b/MitmLibrary/rules.py
index 369ead2..1fded6c 100644
--- a/MitmLibrary/rules.py
+++ b/MitmLibrary/rules.py
@@ -14,9 +14,10 @@
import asyncio
import threading
+from collections.abc import Sequence
from dataclasses import dataclass, field
from enum import Enum, IntEnum
-from typing import Any, Dict, List, Optional, Sequence, Union
+from typing import Any
from mitmproxy import http
from robot.api import logger
@@ -89,7 +90,7 @@ async def apply_async(self, flow: http.HTTPFlow) -> bool:
"""Applies the action from an async hook. Overridden only by actions that wait."""
return self.apply(flow)
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
"""The action's settings, for `Get Proxy Rules` and the log."""
raise NotImplementedError
@@ -100,7 +101,7 @@ class BlockAction(Action):
mode: BlockMode = BlockMode.RESPOND
status_code: int = 403
- body: Optional[str] = None
+ body: str | None = None
phase = Phase.REQUEST
priority = Priority.TERMINAL
@@ -114,8 +115,8 @@ def apply(self, flow: http.HTTPFlow) -> bool:
)
return True
- def describe(self) -> Dict[str, Any]:
- described: Dict[str, Any] = {"type": "block", "mode": self.mode.value}
+ def describe(self) -> dict[str, Any]:
+ described: dict[str, Any] = {"type": "block", "mode": self.mode.value}
if self.mode is BlockMode.RESPOND:
described["status_code"] = self.status_code
described["body"] = self.body
@@ -127,8 +128,8 @@ class ResponseAction(Action):
"""Replaces the whole response."""
status_code: int = 200
- headers: Optional[Dict[str, str]] = None
- body: Optional[str] = None
+ headers: dict[str, str] | None = None
+ body: str | None = None
phase = Phase.RESPONSE
priority = Priority.REPLACE
@@ -156,7 +157,7 @@ def _headers(self, flow: http.HTTPFlow) -> http.Headers:
return flow.response.headers
return http.Headers()
- def _content(self, flow: http.HTTPFlow) -> Union[str, bytes]:
+ def _content(self, flow: http.HTTPFlow) -> str | bytes:
"""The body to use, keeping the original one when none was given."""
if self.body is not None:
return safe_str(self.body)
@@ -164,7 +165,7 @@ def _content(self, flow: http.HTTPFlow) -> Union[str, bytes]:
return flow.response.content
return b""
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {
"type": "response",
"status_code": self.status_code,
@@ -187,7 +188,7 @@ def apply(self, flow: http.HTTPFlow) -> bool:
flow.response.status_code = self.status_code
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {"type": "status", "status_code": self.status_code}
@@ -208,7 +209,7 @@ async def apply_async(self, flow: http.HTTPFlow) -> bool:
await asyncio.sleep(self.seconds)
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {"type": "delay", "delay": self.delay, "seconds": self.seconds}
@@ -220,13 +221,13 @@ class HeadersAction(Action):
adding one header should not mean restating every other header the response had.
"""
- set_headers: Optional[Dict[str, str]] = None
- remove_headers: Optional[Sequence[str]] = None
+ set_headers: dict[str, str] | None = None
+ remove_headers: Sequence[str] | None = None
phase = Phase.RESPONSE
priority = Priority.MUTATE
- def _target(self, flow: http.HTTPFlow) -> Optional[http.Message]:
+ def _target(self, flow: http.HTTPFlow) -> http.Message | None:
"""The message this action edits."""
return flow.response if self.phase is Phase.RESPONSE else flow.request
@@ -242,7 +243,7 @@ def apply(self, flow: http.HTTPFlow) -> bool:
message.headers[name] = value
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {
"type": f"{self.phase.value}_headers",
"headers": self.set_headers,
@@ -267,7 +268,7 @@ class BodyAction(Action):
phase = Phase.RESPONSE
priority = Priority.MUTATE
- def _target(self, flow: http.HTTPFlow) -> Optional[http.Message]:
+ def _target(self, flow: http.HTTPFlow) -> http.Message | None:
return flow.response if self.phase is Phase.RESPONSE else flow.request
def apply(self, flow: http.HTTPFlow) -> bool:
@@ -279,7 +280,7 @@ def apply(self, flow: http.HTTPFlow) -> bool:
message.set_content(safe_str(self.body).encode("utf-8"))
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {"type": f"{self.phase.value}_body", "body": self.body}
@@ -306,7 +307,7 @@ def apply(self, flow: http.HTTPFlow) -> bool:
flow.request.url = self.target
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {"type": "rewrite", "target": self.target}
@@ -315,8 +316,8 @@ class RedirectAction(Action):
"""Sends the request to a different host, keeping its path and query."""
host: str = ""
- port: Optional[int] = None
- scheme: Optional[str] = None
+ port: int | None = None
+ scheme: str | None = None
phase = Phase.REQUEST
priority = Priority.MUTATE
@@ -337,7 +338,7 @@ def apply(self, flow: http.HTTPFlow) -> bool:
)
return False
- def describe(self) -> Dict[str, Any]:
+ def describe(self) -> dict[str, Any]:
return {
"type": "redirect",
"host": self.host,
@@ -412,7 +413,7 @@ class RuleRegistry:
def __init__(self) -> None:
self._lock = threading.RLock()
- self._rules: Dict[str, Rule] = {}
+ self._rules: dict[str, Rule] = {}
self._counter = 0
def add(self, rule: Rule) -> bool:
@@ -443,11 +444,11 @@ def clear(self) -> None:
with self._lock:
self._rules.clear()
- def get(self, alias: str) -> Optional[Rule]:
+ def get(self, alias: str) -> Rule | None:
with self._lock:
return self._rules.get(alias)
- def snapshot(self, phase: Optional[Phase] = None) -> List[Rule]:
+ def snapshot(self, phase: Phase | None = None) -> list[Rule]:
"""The rules for a phase, in the order they should be applied.
A copy, so the proxy can work through it while a keyword adds or removes rules.
@@ -481,6 +482,6 @@ def consume(self, rule: Rule) -> bool:
rule.used += 1
return True
- def describe(self) -> List[DotDict]:
+ def describe(self) -> list[DotDict]:
"""Every rule, in application order, for `Get Proxy Rules`."""
return [rule.describe() for rule in self.snapshot()]
diff --git a/poetry.lock b/poetry.lock
index 0355a21..e91df61 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -2185,13 +2185,13 @@ robotframework-pythonlibcore = ">=3.0.0"
[[package]]
name = "robotframework-browser"
-version = "20.3.0"
+version = "20.4.0"
description = "Robot Framework Browser library powered by Playwright. Aiming for speed, reliability and visibility."
optional = false
python-versions = ">=3.10"
groups = ["dev"]
files = [
- {file = "robotframework_browser-20.3.0-py3-none-any.whl", hash = "sha256:36efd3f4698a962413c8a10108e9a0e6b05123379b2603dc767336b10d9dd6ff"},
+ {file = "robotframework_browser-20.4.0-py3-none-any.whl", hash = "sha256:db2e2f743c123603e87c43d2ddc9b52e0bb2c3764ad91b943140ad300794339f"},
]
[package.dependencies]
@@ -2202,9 +2202,9 @@ overrides = ">=7.7.0"
protobuf = "7.35.1"
psutil = ">=7.2.2"
PyYAML = ">=6.0.3"
-robotframework = ">=6.1.1,<9.0.0"
-robotframework-assertion-engine = ">=5.0.1,<6.0.0"
-robotframework-pythonlibcore = ">=4.4.1,<5.0.0"
+robotframework = ">=7.1.1,<9.0.0"
+robotframework-assertion-engine = "5.0.1"
+robotframework-pythonlibcore = "4.6.0"
seedir = ">=0.5.1"
wrapt = ">=2.2.2"
@@ -2285,31 +2285,30 @@ oldlibyaml = ["ruamel.yaml.clib ; platform_python_implementation == \"CPython\""
[[package]]
name = "ruff"
-version = "0.12.12"
+version = "0.16.4"
description = "An extremely fast Python linter and code formatter, written in Rust."
optional = false
python-versions = ">=3.7"
groups = ["dev"]
files = [
- {file = "ruff-0.12.12-py3-none-linux_armv6l.whl", hash = "sha256:de1c4b916d98ab289818e55ce481e2cacfaad7710b01d1f990c497edf217dafc"},
- {file = "ruff-0.12.12-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:7acd6045e87fac75a0b0cdedacf9ab3e1ad9d929d149785903cff9bb69ad9727"},
- {file = "ruff-0.12.12-py3-none-macosx_11_0_arm64.whl", hash = "sha256:abf4073688d7d6da16611f2f126be86523a8ec4343d15d276c614bda8ec44edb"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:968e77094b1d7a576992ac078557d1439df678a34c6fe02fd979f973af167577"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:42a67d16e5b1ffc6d21c5f67851e0e769517fb57a8ebad1d0781b30888aa704e"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:b216ec0a0674e4b1214dcc998a5088e54eaf39417327b19ffefba1c4a1e4971e"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:59f909c0fdd8f1dcdbfed0b9569b8bf428cf144bec87d9de298dcd4723f5bee8"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:9ac93d87047e765336f0c18eacad51dad0c1c33c9df7484c40f98e1d773876f5"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:01543c137fd3650d322922e8b14cc133b8ea734617c4891c5a9fccf4bfc9aa92"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2afc2fa864197634e549d87fb1e7b6feb01df0a80fd510d6489e1ce8c0b1cc45"},
- {file = "ruff-0.12.12-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:0c0945246f5ad776cb8925e36af2438e66188d2b57d9cf2eed2c382c58b371e5"},
- {file = "ruff-0.12.12-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:a0fbafe8c58e37aae28b84a80ba1817f2ea552e9450156018a478bf1fa80f4e4"},
- {file = "ruff-0.12.12-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:b9c456fb2fc8e1282affa932c9e40f5ec31ec9cbb66751a316bd131273b57c23"},
- {file = "ruff-0.12.12-py3-none-musllinux_1_2_i686.whl", hash = "sha256:5f12856123b0ad0147d90b3961f5c90e7427f9acd4b40050705499c98983f489"},
- {file = "ruff-0.12.12-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:26a1b5a2bf7dd2c47e3b46d077cd9c0fc3b93e6c6cc9ed750bd312ae9dc302ee"},
- {file = "ruff-0.12.12-py3-none-win32.whl", hash = "sha256:173be2bfc142af07a01e3a759aba6f7791aa47acf3604f610b1c36db888df7b1"},
- {file = "ruff-0.12.12-py3-none-win_amd64.whl", hash = "sha256:e99620bf01884e5f38611934c09dd194eb665b0109104acae3ba6102b600fd0d"},
- {file = "ruff-0.12.12-py3-none-win_arm64.whl", hash = "sha256:2a8199cab4ce4d72d158319b63370abf60991495fb733db96cd923a34c52d093"},
- {file = "ruff-0.12.12.tar.gz", hash = "sha256:b86cd3415dbe31b3b46a71c598f4c4b2f550346d1ccf6326b347cc0c8fd063d6"},
+ {file = "ruff-0.16.4-py3-none-linux_armv6l.whl", hash = "sha256:df4075f71ddac40b9934af60c3ec8a53047dd5a5fdc43224e6e4e8e9a27cb6f7"},
+ {file = "ruff-0.16.4-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:0c95538517af68004306b0fb3214ff2f2af67a65092aee77cd9eb86db6656604"},
+ {file = "ruff-0.16.4-py3-none-macosx_11_0_arm64.whl", hash = "sha256:963f83df8e69e575b64d67dd447ebbc917db41a14bf38d4593a4183e7aaa8255"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:32a5057c7ff3f6e6480a48fccfb3a412a690f48a3d03ac5cf08177d6c2da3ade"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:b3dce8d9b0c57c265b91885a66a567d8ea1372e8eb4e250fa8e5e3f579e99cff"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:7dc651db49283c69f8e72c834eec4fe5573e4c646856aebece0ce385dceb2a80"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3817b87dbcabc92f13b05019257c5b89b5b4d51b5fb20f56fb5235ceb723cd07"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e9fce1499134b2c8c68e5166f95705a5812062bb93aacc5f9873bb1a27084bc7"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f2d812e482f5a7e02eee26cd73d2a37ebbdf47d795ea63ba1b89110ae93e9fb3"},
+ {file = "ruff-0.16.4-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:6baaf984aa7976edf93d3b627fe2d1d22ee94bbca05fa6f90fc76d73924e3454"},
+ {file = "ruff-0.16.4-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:bdfcf0b28662eb890372d50f92c283bb94e67e7635ed93c7fd533970acff7b2b"},
+ {file = "ruff-0.16.4-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:b66b02cb9b04f537643cadf5768e5f98dc461890d530cb67113d71c8c76e605d"},
+ {file = "ruff-0.16.4-py3-none-musllinux_1_2_i686.whl", hash = "sha256:8528bf9a4b291a60bf02ea453511e8ce6215bd2b982ee80405b66b008b6c30a0"},
+ {file = "ruff-0.16.4-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:fbd85d2875fdd67e833213a651f613bbf25303abf6aa822a5121f4531195678d"},
+ {file = "ruff-0.16.4-py3-none-win32.whl", hash = "sha256:312769988007aaeb8e189b443ccdd03c0e6374489e053467be6d96518ebff76e"},
+ {file = "ruff-0.16.4-py3-none-win_amd64.whl", hash = "sha256:05d9d27a18c4bcbefada602480ec9e01e0bc949d432e0ced5df77edac195919c"},
+ {file = "ruff-0.16.4-py3-none-win_arm64.whl", hash = "sha256:a3a61621c9b6f6a89573e938a080e648f1695baa3f58570a3a707bc51ff65a21"},
+ {file = "ruff-0.16.4.tar.gz", hash = "sha256:13171aa9d9af2240ee3504e639de73122c67e74036de5ba2e1d01422cd17e3dc"},
]
[[package]]
@@ -2791,4 +2790,4 @@ cffi = ["cffi (>=1.17,<2.0) ; platform_python_implementation != \"PyPy\" and pyt
[metadata]
lock-version = "2.1"
python-versions = ">=3.12,<3.15"
-content-hash = "9845474715573879ea6f5c6e93193e44f84960eed418af4e1b0d00a421581fcf"
+content-hash = "6e58881a16e7ab0aa9811f0d5a8ebbd4a74ea8dc09b0488a83ad4d7e01829162"
diff --git a/pyproject.toml b/pyproject.toml
index fc2743f..ba18615 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -45,7 +45,7 @@ coverage = "^7.6.10"
robotframework-browser = ">=19.1.2,<21.0.0"
robotframework-requests = "^0.9.7"
setuptools = ">=83,<85"
-ruff = "^0.12.4"
+ruff = ">=0.12.4,<0.17.0"
mypy = "^1.18.2"
robotframework-robocop = "^6.0.0"
@@ -56,6 +56,11 @@ line-length = 120
[tool.ruff.lint]
extend-select = ["I"] # imports
+[tool.ruff.lint.per-file-ignores]
+# Robot Framework requires the library class name to match the package/module
+# name, which forces this PascalCase package directory.
+"MitmLibrary/__init__.py" = ["N999"]
+
[tool.ruff.lint.isort]
combine-as-imports = true
order-by-type = false
diff --git a/tests/test_docs_index.py b/tests/test_docs_index.py
index 05971a5..e6bbb2d 100644
--- a/tests/test_docs_index.py
+++ b/tests/test_docs_index.py
@@ -13,7 +13,7 @@
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "tools"))
-import build_docs_index # noqa: E402 - needs the path above
+import build_docs_index
class TestUpdateVersions(unittest.TestCase):
diff --git a/tests/test_proxy_integration.py b/tests/test_proxy_integration.py
index 9f720df..f250ced 100644
--- a/tests/test_proxy_integration.py
+++ b/tests/test_proxy_integration.py
@@ -96,9 +96,8 @@ def test_port_zero_reports_the_port_the_system_picked(self):
# ourselves proves. Opening a connection would prove it too, but a client that
# connects and never sends a request makes the proxy log an error, and that
# error then belongs to no test in particular.
- with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
- with self.assertRaises(OSError):
- sock.bind(("127.0.0.1", address.port))
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock, self.assertRaises(OSError):
+ sock.bind(("127.0.0.1", address.port))
def test_an_unrelated_error_during_startup_does_not_fail_the_keyword(self):
"""A proxy stopped moments ago still logs from its own teardown, and the rest of
@@ -200,7 +199,7 @@ def _request_through_proxy(proxy_port, url):
)
try:
opener.open(url, timeout=5).close()
- except Exception: # noqa: BLE001 - the answer does not matter, only the record
+ except Exception: # noqa: BLE001, S110 - the answer does not matter, only the record
pass
def test_stopping_removes_the_mitmproxy_log_handler(self):
"""mitmproxy leaves a root logger handler behind that outlives its own loop.
diff --git a/tests/test_proxy_modes.py b/tests/test_proxy_modes.py
index f9336f0..f2ce554 100644
--- a/tests/test_proxy_modes.py
+++ b/tests/test_proxy_modes.py
@@ -48,7 +48,7 @@ async def _runs_until_stopped(stop):
class _Handler(http.server.BaseHTTPRequestHandler):
"""Answers everything with a fixed body, so a test can tell it apart."""
- def do_GET(self): # noqa: N802 - the name is fixed by http.server
+ def do_GET(self):
body = b"hello from the origin"
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
diff --git a/tools/build_docs_index.py b/tools/build_docs_index.py
index bfde6e2..cc699b9 100644
--- a/tools/build_docs_index.py
+++ b/tools/build_docs_index.py
@@ -13,7 +13,7 @@
import json
import sys
from pathlib import Path
-from typing import Any, Dict, List
+from typing import Any
DOCUMENT = "MitmLibraryKeywords.html"
@@ -71,7 +71,7 @@
"""
-def _entry(version: Dict[str, Any]) -> str:
+def _entry(version: dict[str, Any]) -> str:
"""Renders one line of the list."""
name = html.escape(str(version["version"]))
path = html.escape(str(version["path"]))
@@ -81,7 +81,7 @@ def _entry(version: Dict[str, Any]) -> str:
return f' {name}{tags}'
-def render(versions: List[Dict[str, Any]]) -> str:
+def render(versions: list[dict[str, Any]]) -> str:
"""Renders the landing page for the given versions, newest first."""
if not versions:
entries = " No documentation has been published yet."
@@ -94,8 +94,8 @@ def render(versions: List[Dict[str, Any]]) -> str:
def update_versions(
- versions: List[Dict[str, Any]], path: str, is_release: bool
-) -> List[Dict[str, Any]]:
+ versions: list[dict[str, Any]], path: str, is_release: bool
+) -> list[dict[str, Any]]:
"""Records a published version, and works out which release is the newest.
`path` is the directory it was published under: a version number for a release, or
@@ -128,7 +128,7 @@ def _release_order(path: str) -> Any:
return ()
-def _sort_key(version: Dict[str, Any]) -> Any:
+def _sort_key(version: dict[str, Any]) -> Any:
"""Orders releases newest first, with anything unreleased above them."""
raw = str(version["version"])
parts = raw.split(".")
@@ -139,7 +139,7 @@ def _sort_key(version: Dict[str, Any]) -> Any:
return (-1, ())
-def main(argv: List[str]) -> int:
+def main(argv: list[str]) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("versions", type=Path, help="the versions.json to read and write")
parser.add_argument("output", type=Path, help="where to write index.html")
@@ -153,7 +153,7 @@ def main(argv: List[str]) -> int:
)
arguments = parser.parse_args(argv)
- versions: List[Dict[str, Any]] = []
+ versions: list[dict[str, Any]] = []
if arguments.versions.exists():
versions = json.loads(arguments.versions.read_text(encoding="utf-8"))
if arguments.add is not None: