diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index df370c17..8f92df41 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,7 +13,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"] + python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"] steps: - uses: actions/checkout@v6 @@ -26,7 +26,7 @@ jobs: - name: Install dependencies run: | python -m pip install --upgrade pip - pip install -Ue .[test] + pip install -Ue . --group test pip install -Ue . - name: Run tests diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index c7a60d29..a87ae78e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -25,7 +25,7 @@ repos: - id: rst-directive-colons - id: rst-inline-touching-normal - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.21 + rev: v0.16.0 hooks: - id: ruff-check alias: "ruff" @@ -54,8 +54,13 @@ repos: - id: conventional-pre-commit stages: [commit-msg] args: ["--strict"] + - repo: https://github.com/astral-sh/ty-pre-commit + rev: v0.0.63 + hooks: + - id: ty + stages: [manual] - repo: https://github.com/pre-commit/mirrors-mypy - rev: v2.2.0 + rev: v2.3.0 hooks: - id: mypy stages: [manual] diff --git a/Makefile b/Makefile index f51821b5..d8c5eae4 100755 --- a/Makefile +++ b/Makefile @@ -1,13 +1,30 @@ #!/usr/bin/make -f +.PHONY: dist pushrelease pushreleasetest clean lint lint-fix format + dist: - python3 setup.py sdist bdist_wheel + python3 setup.py sdist bdist_wheel pushreleasetest: - python3 -m twine upload --repository testpypi dist/* + python3 -m twine upload --repository testpypi dist/* pushrelease: - python3 -m twine upload dist/* + python3 -m twine upload dist/* clean: - $(RM) -r dist + $(RM) -r dist + +lint: + -prek run -a + +lint-fix: + -prek run -a --hook-stage manual ruff-fix + +format: + -prek run -a --hook-stage manual ruff-format + +ty: + -prek run -a --hook-stage manual ty + +mypy: + -prek run -a --hook-stage manual mypy diff --git a/httoop/__main__.py b/httoop/__main__.py index 64089b95..d8010d8e 100644 --- a/httoop/__main__.py +++ b/httoop/__main__.py @@ -9,7 +9,8 @@ import pathlib import sys -from argparse import ArgumentParser, FileType +from argparse import ArgumentParser +from typing import IO from httoop import Request, Response, __name__ as name, __version__ as version from httoop.client import ClientStateMachine @@ -51,7 +52,7 @@ def add_subparsers(self) -> None: request = parse_message_subparsers.add_parser('request', parents=[self.parent_parser]) request.set_defaults(func=self.parse_request) add = request.add_argument - add('--file', default='-', type=FileType('rb')) + add('--file', default='-') add('--scheme', default='http') add('--host', default='www.example.net') add('--port', default=80, type=int) @@ -59,24 +60,27 @@ def add_subparsers(self) -> None: response = parse_message_subparsers.add_parser('response', parents=[self.parent_parser]) add = response.add_argument response.set_defaults(func=self.parse_response) - add('--file', default='-', type=FileType('rb')) + add('--file', default='-') def parse_arguments(self) -> None: self.arguments = self.parser.parse_args() - - if self.arguments.action == 'parse' and hasattr(self.arguments.file, 'buffer'): - # https://bugs.python.org/issue14156 - self.arguments.file = self.arguments.file.buffer self.arguments.func() - def add_common_arguments(self, add) -> None: + @classmethod + def add_common_arguments(cls, add) -> None: add('--protocol') add('-H', '--header', action='append', default=[]) add('-b', '--body', default='') + @classmethod + def get_file(cls, file: str) -> IO: + if file == '-': + return sys.stdin.buffer + return pathlib.Path(file).open('rb') + def parse_request(self) -> None: server = ServerStateMachine(self.arguments.scheme, self.arguments.host, self.arguments.port) - for _request, response in server.parse(self.arguments.file.read()): + for _request, response in server.parse(self.get_file(self.arguments.file).read()): print(repr(response)) print(repr(response.headers)) print(repr(response.body)) @@ -84,12 +88,13 @@ def parse_request(self) -> None: def parse_response(self) -> None: client = ClientStateMachine() client.request = Request() - for response in client.parse(self.arguments.file.read()): + for response in client.parse(self.get_file(self.arguments.file).read()): print(repr(response)) print(repr(response.headers)) print(repr(response.body)) print(repr(bytes(response.body))) if client.buffer: + assert client.message # noqa: S101 print('WARNING: response not yet complete!:') print(repr(client.message)) print(repr(client.message.headers)) @@ -115,10 +120,11 @@ def compose_response(self) -> None: self.common() def common(self) -> None: + assert self.message is not None # noqa: S101 if self.arguments.protocol: protocol = self.arguments.protocol try: - protocol = [int(x) for x in protocol.split('.', 1)] + protocol = tuple(int(x) for x in protocol.split('.', 1)) except ValueError: pass else: @@ -133,14 +139,15 @@ def common(self) -> None: if body == '-': body = sys.stdin.read() elif body.startswith('@'): - body = pathlib.Path(body[1:]).open('rb') + body = pathlib.Path(body[1:]).open('rb') # noqa: SIM115 self.message.body = body sys.stdout.write(self.decode(bytes(self.message))) sys.stdout.write(self.decode(bytes(self.message.headers))) sys.stdout.write(self.decode(bytes(self.message.body))) - def decode(self, data): + @classmethod + def decode(cls, data): if str is not bytes: data = data.decode('ISO8859-1') return data diff --git a/httoop/authentication/__init__.py b/httoop/authentication/__init__.py index a36ba6a8..8004ff64 100644 --- a/httoop/authentication/__init__.py +++ b/httoop/authentication/__init__.py @@ -1,7 +1,7 @@ from __future__ import annotations import re -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, ClassVar from httoop.authentication.basic import BasicAuthRequestScheme, BasicAuthResponseScheme from httoop.authentication.digest import DigestAuthRequestScheme, DigestAuthResponseScheme, SecureDigestAuthRequestScheme @@ -16,7 +16,7 @@ class AuthElement(HeaderElement): - schemes = {} + schemes: ClassVar = {} RE_SPACE_SPLIT = re.compile(rb'\s+(?=(?:[^"]*"[^"]*")*[^"]*$)') def sanitize(self) -> None: @@ -31,16 +31,16 @@ def parseparams(cls, elementstr: bytes) -> tuple[bytes, dict[bytes, str]] | tupl try: scheme, authinfo = elementstr.split(b' ', 1) except ValueError: - raise InvalidHeader(_('Authorization headers must contain authentication scheme')) + raise InvalidHeader(_('Authorization headers must contain authentication scheme')) from None try: parser = cls.schemes[scheme.decode('ISO8859-1').lower()] except KeyError: - raise InvalidHeader(_('Unsupported authentication scheme: %r'), scheme) + raise InvalidHeader(_('Unsupported authentication scheme: %r'), scheme) from None try: authinfo = parser.parse(authinfo) except KeyError as key: - raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), scheme) + raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), scheme) from None return scheme.title(), authinfo @@ -48,12 +48,12 @@ def compose(self) -> bytes: try: scheme = self.schemes[self.value.lower()] except KeyError: - raise InvalidHeader(_('Unsupported authentication scheme: %r'), self.value) + raise InvalidHeader(_('Unsupported authentication scheme: %r'), self.value) from None try: authinfo = scheme.compose(self.params) except KeyError as key: - raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), self.value) + raise InvalidHeader(_('Missing parameter %r for authentication scheme %r'), str(key), self.value) from None return b'%s %s' % (self.value.encode('ASCII').title(), authinfo) @@ -68,7 +68,7 @@ class AuthRequestElement(AuthElement): encoding = 'ASCII' - schemes = { + schemes: ClassVar = { 'basic': BasicAuthRequestScheme, 'digest': SecureDigestAuthRequestScheme, } @@ -103,7 +103,7 @@ def password(self, password) -> None: class AuthResponseElement(AuthElement): - schemes = { + schemes: ClassVar = { 'basic': BasicAuthResponseScheme, 'digest': DigestAuthResponseScheme, } diff --git a/httoop/authentication/basic.py b/httoop/authentication/basic.py index 5eb81080..59583940 100644 --- a/httoop/authentication/basic.py +++ b/httoop/authentication/basic.py @@ -19,9 +19,9 @@ def parse(authinfo: bytes) -> dict[str, bytes]: if not username: raise ValueError() except Base64Error: - raise InvalidHeader(_('Basic authentication contains invalid base64')) + raise InvalidHeader(_('Basic authentication contains invalid base64')) from None except ValueError: - raise InvalidHeader(_('No username:password provided')) + raise InvalidHeader(_('No username:password provided')) from None return { # 'username': username.decode('ISO8859-1'), @@ -42,11 +42,11 @@ def compose(authinfo: ByteUnicodeDict) -> bytes: class BasicAuthResponseScheme: @staticmethod - def parse(authinfo: bytes) -> dict[bytes, str | bytes]: + def parse(authinfo: bytes) -> dict[bytes, str]: params = HeaderElement.parseparams(b'X;%s' % authinfo)[1] - params.setdefault(b'realm', b'') + params.setdefault(b'realm', '') return params @staticmethod def compose(authinfo: ByteUnicodeDict) -> bytes: - return HeaderElement.formatparam(b'realm', authinfo['realm'], True) + return HeaderElement.formatparam(b'realm', authinfo['realm'], quote=True) diff --git a/httoop/authentication/digest.py b/httoop/authentication/digest.py index f8bf0de1..d1fc00f5 100644 --- a/httoop/authentication/digest.py +++ b/httoop/authentication/digest.py @@ -1,9 +1,10 @@ +# ruff: file-ignore[N806,N802] from __future__ import annotations from hashlib import md5, new, sha256 from hmac import compare_digest from secrets import token_bytes -from typing import Callable +from typing import Callable, ClassVar from httoop.exceptions import InvalidHeader from httoop.header.element import HeaderElement @@ -11,11 +12,11 @@ class DigestAuthScheme: - algorithms = { - 'MD5': lambda: md5(), # noqa: S324 - 'MD5-sess': lambda: md5(), # noqa: S324 - 'SHA-256': lambda: sha256(), - 'SHA-256-sess': lambda: sha256(), + algorithms: ClassVar = { + 'MD5': md5, + 'MD5-sess': md5, + 'SHA-256': sha256, + 'SHA-256-sess': sha256, 'SHA-512-256': lambda: new('sha512_256'), 'SHA-512-256-sess': lambda: new('sha512_256'), } # not case insensitive per RFC diff --git a/httoop/client/__init__.py b/httoop/client/__init__.py index af4e4a2e..6393b695 100644 --- a/httoop/client/__init__.py +++ b/httoop/client/__init__.py @@ -1,15 +1,17 @@ from __future__ import annotations from httoop.exceptions import InvalidLine -from httoop.messages import Response +from httoop.messages import Request, Response from httoop.parser import NOT_RECEIVED_YET, StateMachine class ClientStateMachine(StateMachine): Message = Response + request: Request + message: Response - def __init__(self, *, strict: bool = True, max_status_line_length: int | float = 256) -> None: + def __init__(self, *, strict: bool = True, max_status_line_length: float = 256) -> None: super().__init__(strict=strict) self.max_status_line_length = max_status_line_length diff --git a/httoop/codecs/__init__.py b/httoop/codecs/__init__.py index 1ddd3e35..bb9c0074 100644 --- a/httoop/codecs/__init__.py +++ b/httoop/codecs/__init__.py @@ -18,7 +18,7 @@ __all__ = ['CODECS', 'Codec', 'application', 'audio', 'example', 'image', 'message', 'model', 'multipart', 'text', 'video'] -def lookup(encoding: str, raise_errors: bool = True) -> Any: +def lookup(encoding: str, *, raise_errors: bool = True) -> Any: type_ = '%s/*' % (encoding.split('/', 1)[0],) return CODECS.get(encoding) or CODECS.get(type_) or (raise_errors and CODECS[encoding]) or None diff --git a/httoop/codecs/application/gzip.py b/httoop/codecs/application/gzip.py index 93dc9052..e457ee35 100644 --- a/httoop/codecs/application/gzip.py +++ b/httoop/codecs/application/gzip.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import gzip import io import zlib @@ -20,30 +22,30 @@ def encode(cls, data: bytes, charset: None = None, mimetype: None = None) -> byt fd.write(Codec.encode(data, charset)) return out.getvalue() except zlib.error: # pragma: no cover - raise EncodeError(_('Invalid gzip data.')) + raise EncodeError(_('Invalid gzip data.')) from None @classmethod - def decode(cls, data: bytes, charset: None = None, mimetype: None = None) -> str: + def decode(cls, data: bytes, charset: str | None = None, mimetype: None = None) -> str: try: with gzip.GzipFile(fileobj=io.BytesIO(data)) as fd: data = fd.read() except (zlib.error, OSError, EOFError): - raise DecodeError(_('Invalid gzip data.')) + raise DecodeError(_('Invalid gzip data.')) from None return Codec.decode(data, charset) @classmethod def iterencode(cls, data, charset=None, mimetype=None): + out = io.BytesIO() try: - out = io.BytesIO() with gzip.GzipFile(fileobj=out, mode='w', compresslevel=cls.compression_level) as fd: for part in data: fd.write(Codec.encode(part, charset)) yield out.getvalue() out.seek(0) out.truncate() - yield out.getvalue() except zlib.error: # pragma: no cover - raise EncodeError(_('Invalid gzip data.')) + raise EncodeError(_('Invalid gzip data.')) from None + yield out.getvalue() @classmethod def iterdecode(cls, data, charset=None, mimetype=None): @@ -62,4 +64,4 @@ def iterdecode(cls, data, charset=None, mimetype=None): fd.seek(0) yield Codec.decode(gzfd.read(), charset) except (zlib.error, OSError, EOFError): - raise DecodeError(_('Invalid gzip data.')) + raise DecodeError(_('Invalid gzip data.')) from None diff --git a/httoop/codecs/application/hal_json.py b/httoop/codecs/application/hal_json.py index e89cd0d2..c19fd44a 100644 --- a/httoop/codecs/application/hal_json.py +++ b/httoop/codecs/application/hal_json.py @@ -95,7 +95,7 @@ def get_resource(self, relation: str) -> Resource | None: except StopIteration: pass - def expand(___self, ___href, **templates): + def expand(___self, ___href, **templates): # noqa: N805 return expand(___href, templates) def get_curie(self, relation: str) -> str: @@ -125,10 +125,10 @@ class HAL(JSON): @classmethod def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | None = None) -> Resource: - data = super().decode(data) - if not isinstance(data, dict): + doc = super().decode(data) + if not isinstance(doc, dict): raise DecodeError('HAL documents must be JSON objects.') - return Resource(data) + return Resource(doc) @classmethod def encode(cls, data: dict[str, None] | Resource, charset: str | None = None, mimetype: ContentType | None = None) -> bytes: @@ -138,6 +138,6 @@ def encode(cls, data: dict[str, None] | Resource, charset: str | None = None, mi try: Resource(data.copy()) except DecodeError as exc: - raise EncodeError(str(exc)) + raise EncodeError(str(exc)) from exc return super().encode(data) diff --git a/httoop/codecs/application/json.py b/httoop/codecs/application/json.py index 097ae831..8fcfc7eb 100644 --- a/httoop/codecs/application/json.py +++ b/httoop/codecs/application/json.py @@ -15,10 +15,8 @@ class JSON(Codec): @classmethod def encode(cls, data: dict[str, str], charset: str | None = None, mimetype: ContentType | None = None) -> bytes: - data = json_encode(data) - if not isinstance(data, bytes): # python3 - data = data.encode(charset or 'UTF-8') - return data + doc = json_encode(data) + return doc.encode(charset or 'UTF-8') @classmethod def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | None = None) -> dict[str, Any]: diff --git a/httoop/codecs/application/x_www_form_urlencoded.py b/httoop/codecs/application/x_www_form_urlencoded.py index 70f3af02..8ecaca28 100644 --- a/httoop/codecs/application/x_www_form_urlencoded.py +++ b/httoop/codecs/application/x_www_form_urlencoded.py @@ -39,5 +39,4 @@ def unquote(cls, data: bytes, charset: str | None = None) -> str: @classmethod def quote(cls, data: str | list[int], charset: str | None = None) -> bytes: - data = data.encode(charset or 'ISO8859-1') - return Percent.quote(data, cls.UNQUOTED) + return Percent.quote(data.encode(charset or 'ISO8859-1'), cls.UNQUOTED) diff --git a/httoop/codecs/application/xml.py b/httoop/codecs/application/xml.py index f11983d4..a9a2a204 100644 --- a/httoop/codecs/application/xml.py +++ b/httoop/codecs/application/xml.py @@ -33,7 +33,7 @@ def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | try: return fromstring(data) except ParseError as exc: - raise DecodeError(f'Could not decode as {mimetype}: {exc}') + raise DecodeError(f'Could not decode as {mimetype}: {exc}') from exc @classmethod def encode(cls, root: Element, charset: str | None = None, mimetype: ContentType | None = None) -> bytes: diff --git a/httoop/codecs/application/zlib.py b/httoop/codecs/application/zlib.py index e497815b..68a51a34 100644 --- a/httoop/codecs/application/zlib.py +++ b/httoop/codecs/application/zlib.py @@ -15,12 +15,12 @@ def encode(cls, data: bytes, charset: None = None, mimetype: None = None) -> byt try: return zlib.compress(Codec.encode(data, charset)) except zlib.error: # pragma: no cover - raise EncodeError(_('Invalid zlib/deflate data.')) + raise EncodeError(_('Invalid zlib/deflate data.')) from None @classmethod def decode(cls, data: bytes, charset: str | None = None, mimetype: None = None) -> str: try: data = zlib.decompress(data) except zlib.error: - raise DecodeError(_('Invalid zlib/deflate data.')) + raise DecodeError(_('Invalid zlib/deflate data.')) from None return Codec.decode(data, charset) diff --git a/httoop/codecs/codec.py b/httoop/codecs/codec.py index cfaa38b9..e2cb8b7e 100644 --- a/httoop/codecs/codec.py +++ b/httoop/codecs/codec.py @@ -9,14 +9,16 @@ class Codec: + mimetype: str + @classmethod - def decode(cls, data: bytes, charset: str | None = None, mimetype: None = None) -> str: # pragma: no cover + def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | None = None) -> str: # pragma: no cover if isinstance(data, bytes): data = data.decode(charset or 'ascii') return data @classmethod - def encode(cls, data: bytes, charset: None = None, mimetype: None = None) -> bytes: # pragma: no cover + def encode(cls, data: bytes, charset: None = None, mimetype: ContentType | None = None) -> bytes: # pragma: no cover if isinstance(data, str): data = data.encode(charset or 'ascii') return data @@ -26,5 +28,5 @@ def iterencode(cls, data: Any, charset: str | None = None, mimetype: ContentType yield cls.encode(data, charset, mimetype) @classmethod - def iterdecode(cls, data: Any, charset=None, mimetype=None): # pragma: no cover + def iterdecode(cls, data: Any, charset=None, mimetype: ContentType | None = None): # pragma: no cover yield cls.decode(data, charset, mimetype) diff --git a/httoop/codecs/message/http.py b/httoop/codecs/message/http.py index ec79fce0..4f11ba01 100644 --- a/httoop/codecs/message/http.py +++ b/httoop/codecs/message/http.py @@ -21,7 +21,7 @@ def encode(cls, data: Request | Response, charset: str | None = None, mimetype: @classmethod def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | None = None) -> Request | Response: - from httoop.messages import Request, Response + from httoop.messages import Request, Response # noqa: PLC0415 line, data = data.split(b'\r\n', 1) message = Request() diff --git a/httoop/codecs/multipart/multipart.py b/httoop/codecs/multipart/multipart.py index 59bf42f0..207a30f2 100644 --- a/httoop/codecs/multipart/multipart.py +++ b/httoop/codecs/multipart/multipart.py @@ -37,14 +37,13 @@ def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | if part not in {b'--', b'--\r\n'}: raise DecodeError(_('Invalid multipart end: %r'), part.decode('ISO8859-1')) - from httoop.messages.body import Body + from httoop.messages.body import Body # noqa: PLC0415 multiparts = [] for part in parts: if not part.startswith(b'\r\n'): raise DecodeError(_('Invalid boundary end: %r'), part[:2].decode('ISO8859-1')) - part = part[2:] - headers, separator, content = part.partition(b'\r\n\r\n') + headers, separator, content = part[2:].partition(b'\r\n\r\n') if not separator: raise DecodeError(_('Multipart does not contain CRLF header separator')) if not content.endswith(b'\r\n'): diff --git a/httoop/codecs/text/plain.py b/httoop/codecs/text/plain.py index b9819bf1..bf80972f 100644 --- a/httoop/codecs/text/plain.py +++ b/httoop/codecs/text/plain.py @@ -17,15 +17,15 @@ class PlainText(Codec): @classmethod def decode(cls, data: bytes, charset: str | None = None, mimetype: ContentType | None = None) -> str: try: - assert isinstance(data, bytes) + assert isinstance(data, bytes) # noqa: S101 return data.decode(charset or 'UTF-8') except UnicodeDecodeError: - raise DecodeError('Wrong encoding.') + raise DecodeError('Wrong encoding.') from None @classmethod def encode(cls, data: str, charset: str | None = None, mimetype: ContentType | None = None) -> bytes: try: - assert not isinstance(data, bytes) + assert not isinstance(data, bytes) # noqa: S101 return data.encode(charset or 'UTF-8') except UnicodeEncodeError: - raise EncodeError('Wrong encoding.') + raise EncodeError('Wrong encoding.') from None diff --git a/httoop/date.py b/httoop/date.py index 3157c175..9e27fc2e 100644 --- a/httoop/date.py +++ b/httoop/date.py @@ -9,7 +9,7 @@ import time # import calendar -from datetime import datetime +from datetime import datetime as datetime_type, timezone from email.utils import parsedate_tz from typing import Any @@ -21,7 +21,7 @@ __all__ = ['Date'] -class Date(Semantic): +class Date(Semantic): # noqa: PLW1641 """ A HTTP Date string. @@ -50,7 +50,7 @@ def __init__(self, timeval: Any | None = None) -> None: or a timetuple """ self.__composed = None - self.__timestamp = None + self.__timestamp: float = 0.0 self.__datetime = None self.__time_struct = None @@ -61,7 +61,7 @@ def __init__(self, timeval: Any | None = None) -> None: elif isinstance(timeval, (tuple, time.struct_time)): # self.__timestamp = calendar.timegm(timeval) self.__timestamp = time.mktime(timeval) - time.timezone - elif isinstance(timeval, datetime): + elif isinstance(timeval, datetime_type): self.__datetime = timeval # self.__timestamp = calendar.timegm(self.datetime.utctimetuple()) self.__timestamp = time.mktime(self.datetime.utctimetuple()) - time.timezone @@ -75,9 +75,9 @@ def __init__(self, timeval: Any | None = None) -> None: raise TypeError('Date(): got invalid argument') @property - def datetime(self) -> datetime: + def datetime(self) -> datetime_type: if self.__datetime is None: - self.__datetime = datetime.utcfromtimestamp(int(self)) + self.__datetime = datetime_type.fromtimestamp(int(self), tz=timezone.utc) return self.__datetime @property @@ -104,7 +104,7 @@ def __compose(self) -> bytes: ) @classmethod - def parse(cls, timestr: bytes) -> Date: + def parse(cls, data: bytes) -> Date: """ Parses a HTTP date string and returns a :class:`Date` object. @@ -115,7 +115,7 @@ def parse(cls, timestr: bytes) -> Date: :rtype : :class:`Date` """ - timestr = timestr.decode('ISO8859-1') + timestr = data.decode('ISO8859-1') # parse the most common HTTP Date formats (RFC 2822, RFC 1036, C's asctime) date = parsedate_tz(timestr) @@ -148,17 +148,18 @@ def __lt__(self, other: Date | str | None) -> bool: except NotImplementedError: # pragma: no cover return NotImplemented - def __other(self, other): + @staticmethod + def __other(other) -> Date: if other is None: return Date(0) - raise NotImplementedError() # pragma: no cover + # raise NotImplementedError() if isinstance(other, Date): return other try: return Date(other) except (InvalidDate, TypeError): return Date(0) - raise NotImplementedError() # pragma: no cover + # raise NotImplementedError() def __repr__(self) -> str: return '' % (int(self),) diff --git a/httoop/exceptions.py b/httoop/exceptions.py index 72219024..53eb47ce 100644 --- a/httoop/exceptions.py +++ b/httoop/exceptions.py @@ -3,7 +3,7 @@ from httoop.util import _Translateable -class Invalid(_Translateable, ValueError): +class Invalid(_Translateable, ValueError): # noqa: N818 """base class for invalid values.""" diff --git a/httoop/gateway/wsgi.py b/httoop/gateway/wsgi.py index 2d80365c..3f5ecfa5 100644 --- a/httoop/gateway/wsgi.py +++ b/httoop/gateway/wsgi.py @@ -9,7 +9,7 @@ from typing import Any, Callable, Iterator import httoop.header -from httoop.messages import Body +from httoop.messages import Body, Request, Response __all__ = ('WSGI',) @@ -32,7 +32,10 @@ def read(self, *size): class WSGI: """A mixin class which implements the WSGI interface.""" - def __init__(self, environ: dict[str, str] | None = None, use_path_info: bool = False, *args, **kwargs) -> None: + request: Request + response: Response + + def __init__(self, environ: dict[str, str] | None = None, *args, use_path_info: bool = False, **kwargs) -> None: self.use_path_info = use_path_info super().__init__() self.exc_info = None @@ -64,7 +67,7 @@ def __init__(self, environ: dict[str, str] | None = None, use_path_info: bool = def start_response(self) -> None: pass - def __call__(self, application: Callable) -> Iterator[Any]: + def __call__(self, application: Callable) -> Iterator[Any] | None: def write(data): if not self.headers_set: raise RuntimeError('write() before start_response()') @@ -112,12 +115,13 @@ def start_response(status, response_headers, exc_info=None): def buffered(data): try: yield data - for data in result: + for data in result: # noqa: PLR1704 if data: yield data finally: - if hasattr(result, 'close'): - result.close() + close = getattr(result, 'close', None) + if callable(close): + close() self.response.body = buffered(data) return raw_result diff --git a/httoop/header/__init__.py b/httoop/header/__init__.py index 5487969d..b5c04d1f 100644 --- a/httoop/header/__init__.py +++ b/httoop/header/__init__.py @@ -14,8 +14,8 @@ from httoop.header.messaging import Server, UserAgent -__all__ = ['Headers', 'Server', 'UserAgent', 'auth', 'cache', 'conditional', 'messaging', 'ranges', 'security', 'HeaderElement'] +__all__ = ['HeaderElement', 'Headers', 'Server', 'UserAgent', 'auth', 'cache', 'conditional', 'messaging', 'ranges', 'security'] for member in HEADER.values(): name = member.__name__ - __all__.append(name) + __all__ += name # noqa: PLE0605 globals()[name] = member diff --git a/httoop/header/cache.py b/httoop/header/cache.py index 92550f08..36d248e0 100644 --- a/httoop/header/cache.py +++ b/httoop/header/cache.py @@ -27,6 +27,6 @@ class Vary(HeaderElement): is_response_header = True -class Warning(HeaderElement): # pylint: disable=W0622 +class Warning(HeaderElement): # noqa: A001 is_response_header = True diff --git a/httoop/header/conditional.py b/httoop/header/conditional.py index 8a5ea94f..25a000a0 100644 --- a/httoop/header/conditional.py +++ b/httoop/header/conditional.py @@ -3,9 +3,10 @@ from httoop.header.element import HeaderElement -class _DateComparable: +class _DateComparable: # noqa: PLW1641 Date = Date + value: str def sanitize(self) -> None: super().sanitize() @@ -25,7 +26,8 @@ def __int__(self) -> int: return int(self.value) -class _MatchElement: +class _MatchElement: # noqa: PLW1641 + value: str def __eq__(self, other: object) -> bool: return self.value in {other, '*'} @@ -51,7 +53,7 @@ def matches_etag(self, etag, *, strong: bool = True): return value[1:-1] == etag -class ETag(HeaderElement): +class ETag(HeaderElement): # noqa: PLW1641 is_response_header = True diff --git a/httoop/header/element.py b/httoop/header/element.py index 5aeb1c9f..bd1ee415 100644 --- a/httoop/header/element.py +++ b/httoop/header/element.py @@ -25,7 +25,7 @@ HEADER = CaseInsensitiveDict() -class HeaderElement: +class HeaderElement: # noqa: PLW1641 """An element (with parameters) from an HTTP header's element list.""" name = '' @@ -38,7 +38,7 @@ class HeaderElement: # Regular expression that matches `special' characters in parameters, the # existence of which force quoting of the parameter value. - RE_TSPECIALS = re.compile(b'[ \\(\\)<>@,;:\\\\"/\\[\\]\\?=]') + RE_TSPECIALS = re.compile(rb'[ \(\)<>@,;:\\"/\[\]\?=]') RE_SPLIT = re.compile(rb',(?=(?:[^"]*"[^"]*")*[^"]*$)') RE_PARAMS = re.compile(rb';(?=(?:[^"]*"[^"]*")*[^"]*$)') @@ -100,7 +100,7 @@ def parseparam(cls, atom: bytes) -> tuple[bytes, bytes, bool]: try: val, quoted = cls.unescape_param(val.strip()) except InvalidHeader: - raise InvalidHeader(_('Unquoted parameter %r in %r containing TSPECIALS: %r'), key, cls.name, val) + raise InvalidHeader(_('Unquoted parameter %r in %r containing TSPECIALS: %r'), key, cls.name, val) # noqa: B904 return cls.unescape_key(key), val, quoted @classmethod @@ -111,7 +111,7 @@ def unescape_key(cls, key: bytes) -> bytes: def unescape_param(cls, value: bytes) -> tuple[bytes, bool]: quoted = value.startswith(b'"') and value.endswith(b'"') if quoted: - value = re.sub(b'\\\\(?!\\\\)', b'', value[1:-1]) + value = re.sub(rb'\\(?!\\)', b'', value[1:-1]) elif cls.RE_TSPECIALS.search(value): raise InvalidHeader(_('Unquoted parameter in %r containing TSPECIALS: %r'), cls.name, value) return value, quoted @@ -127,20 +127,20 @@ def _sanitize_encoding(cls, charset: str) -> str: def _rfc2231_and_continuation_params(cls, params: Iterator[Any]) -> Iterator[tuple[bytes, str]]: # TODO: complexity count = set() continuations = {} - for key, value, quoted in params: + for key, val, quoted in params: if key in count: raise InvalidHeader(_('Parameter given twice: %r'), key.decode('ISO8859-1')) count.add(key) if b'*' in key: - if key.endswith(b'*') and not quoted and not value.startswith(b"'") and value.count(b"'") >= 2: - charset, _language, value_ = value.split(b"'", 2) + if key.endswith(b'*') and not quoted and not val.startswith(b"'") and val.count(b"'") >= 2: # noqa: PLR2004 + charset, _language, value_ = val.split(b"'", 2) encoding = cls._sanitize_encoding(charset.decode('ASCII', 'replace')) try: - key, value = key[:-1], Percent.unquote(value_).decode(encoding) + key, value = key[:-1], Percent.unquote(value_).decode(encoding) # noqa: PLW2901 except UnicodeDecodeError as exc: - raise InvalidHeader(_('%s') % (exc,)) + raise InvalidHeader(_('%s') % (exc,)) from None else: - value = value.decode('ISO8859-1') + value = val.decode('ISO8859-1') key_, asterisk, num = key.rpartition(b'*') if asterisk: try: @@ -153,7 +153,7 @@ def _rfc2231_and_continuation_params(cls, params: Iterator[Any]) -> Iterator[tup continuations.setdefault(key_, {})[num] = value continue else: - value = value.decode('ISO8859-1') + value = val.decode('ISO8859-1') yield key, value for key, lines in continuations.items(): @@ -172,9 +172,9 @@ def _rfc2231_and_continuation_params(cls, params: Iterator[Any]) -> Iterator[tup yield b'%s*%d' % (key, k), v @classmethod - def parse(cls, elementstr: bytes) -> HeaderElement: + def parse(cls, element: bytes) -> HeaderElement: """Construct an instance from a string of the form 'token;key=val'.""" - elementstr, encoding = cls.decode_rfc2047_charset(elementstr) + elementstr, encoding = cls.decode_rfc2047_charset(element) ival, params = cls.parseparams(elementstr.encode(encoding)) return cls(ival.decode(encoding), params) @@ -195,25 +195,27 @@ def merge(cls, elements: list, others: list) -> bytes: return cls.join([bytes(x) for x in cls.sorted(elements + others)]) @classmethod - def formatparam(cls, param: bytes, value: bytes | str | None = None, quote: bool = False) -> bytes: + def formatparam(cls, param: bytes, value: bytes | str | None = None, *, quote: bool = False) -> bytes: """ Convenience function to format and return a key=value pair. This will quote the value if needed or if quote is true. """ if value: - if not isinstance(value, bytes): + if isinstance(value, bytes): + val = value + else: try: - value = value.encode('ASCII') + val = value.encode('ASCII') except UnicodeEncodeError: param += b'*' - value = b"utf-8''%s" % (Percent.quote(value.encode('UTF-8')),) + val = b"utf-8''%s" % (Percent.quote(value.encode('UTF-8')),) quote = False - if quote or cls.RE_TSPECIALS.search(value): - value = value.replace(b'\\', b'\\\\').replace(b'"', rb'\"') - return b'%s="%s"' % (param, value) - return b'%s=%s' % (param, value) + if quote or cls.RE_TSPECIALS.search(val): + val = val.replace(b'\\', b'\\\\').replace(b'"', rb'\"') + return b'%s="%s"' % (param, val) + return b'%s=%s' % (param, val) return param @classmethod @@ -227,7 +229,7 @@ def decode_rfc2047_charset(cls, value: bytes) -> tuple[str, str]: try: return ''.join(atom.decode(cls._sanitize_encoding(charset or 'ISO8859-1')) for atom, charset in decode_header(value.decode('ISO8859-1'))), 'UTF-8' except (UnicodeDecodeError, HeaderParseError) as exc: - raise InvalidHeader(str(exc)) + raise InvalidHeader(str(exc)) from exc try: return value.decode('ASCII'), 'ASCII' except UnicodeDecodeError: @@ -257,6 +259,9 @@ class MimeType: .. seealso:: rfc:`3023` """ + value: str + params: dict + @property def mimetype(self) -> str: return f'{self.type}/{self.subtype_wo_vendor}' @@ -265,7 +270,7 @@ def mimetype(self) -> str: def type(self): return self.value.split('/', 1)[0] - @type.setter + @type.setter # noqa: A003 # https://github.com/astral-sh/ruff/issues/23074 def type(self, type_) -> None: self.value = f'{type_}/{self.subtype}' @@ -305,7 +310,7 @@ def version(self, version) -> None: self.params['version'] = str(version) -class _AcceptElement(HeaderElement): +class _AcceptElement(HeaderElement): # noqa: PLW1641 """ An Accept element with quality value. @@ -324,18 +329,18 @@ def quality(self) -> float: val = val.value if val: return float(val) - return None + return 0.0 def sanitize(self) -> None: super().sanitize() try: - self.quality + self.quality # noqa: B018 except ValueError: - raise InvalidHeader(_('Quality value must be float.')) + raise InvalidHeader(_('Quality value must be float.')) from None @classmethod - def parse(cls, elementstr: bytes) -> HeaderElement: - elementstr, encoding = cls.decode_rfc2047_charset(elementstr) + def parse(cls, element: bytes) -> HeaderElement: + elementstr, encoding = cls.decode_rfc2047_charset(element) qvalue = None # The first "q" parameter (if any) separates the initial # media-range parameter(s) (if any) from the accept-params. @@ -348,7 +353,7 @@ def parse(cls, elementstr: bytes) -> HeaderElement: media_type, params = cls.parseparams(media_range) if qvalue is not None: - params['q'] = bytes(qvalue) + params[b'q'] = str(qvalue) return cls(media_type.decode(encoding), params) @@ -356,12 +361,12 @@ def parse(cls, elementstr: bytes) -> HeaderElement: def sorted(cls, elements: list) -> list: return sorted(elements, reverse=True) - def __eq__(self, other: str) -> bool: + def __eq__(self, other: object) -> bool: if not isinstance(other, _AcceptElement): other = _AcceptElement(other) return other.value == self.value and other.quality == self.quality - def __lt__(self, other: str) -> bool: + def __lt__(self, other: object) -> bool: if not isinstance(other, _AcceptElement): other = _AcceptElement(other) if self.quality == other.quality: @@ -380,8 +385,8 @@ def __init__(self, cookie_name: str, cookie_value: str, params: dict[bytes, str] super().__init__(self.value, params) @classmethod - def parse(cls, elementstr: bytes) -> HeaderElement: - elementstr, encoding = cls.decode_rfc2047_charset(elementstr) + def parse(cls, element: bytes) -> HeaderElement: + elementstr, encoding = cls.decode_rfc2047_charset(element) value, params = cls.parseparams(elementstr.encode(encoding)) cookie_name, cookie_value, __ = cls.parseparam(value) return cls(cookie_name.decode(encoding), cookie_value.decode(encoding), params) diff --git a/httoop/header/headers.py b/httoop/header/headers.py index 1de4c94a..0a2372c6 100644 --- a/httoop/header/headers.py +++ b/httoop/header/headers.py @@ -24,22 +24,19 @@ def __init__(self, *args, **kwargs) -> None: @staticmethod def formatvalue(value: Any) -> bytes: - if isinstance(value, str): - val = HeaderElement.encode_rfc2047(value) - else: - val = bytes(value) + val = HeaderElement.encode_rfc2047(value) if isinstance(value, str) else bytes(value) if b'\r' in val or b'\n' in val: raise InvalidHeader(_('Invalid header value: %r'), val.decode('ISO8859-1')) return val def __getitem__(self, key: str) -> str: - Element = HEADER.get(key, HeaderElement) - return Element.decode_rfc2047(super().__getitem__(key)) + element_cls = HEADER.get(key, HeaderElement) + return element_cls.decode_rfc2047(super().__getitem__(key)) def get(self, key: str, default: bytes | str | None = None) -> bytes | str | None: - Element = HEADER.get(key, HeaderElement) + element_cls = HEADER.get(key, HeaderElement) try: - return Element.decode_rfc2047(super().__getitem__(key)) + return element_cls.decode_rfc2047(super().__getitem__(key)) except KeyError: return default @@ -68,14 +65,14 @@ def elements(self, fieldname: bytes | str) -> list[Any]: if not fieldvalue: return [] - Element = HEADER.get(fieldname, HeaderElement) - return Element.sorted([Element.parse(element) for element in Element.split(fieldvalue)]) + element_cls = HEADER.get(fieldname, HeaderElement) + return element_cls.sorted([element_cls.parse(element) for element in element_cls.split(fieldvalue)]) def element(self, fieldname: bytes | str, default: None = None) -> Any: """Treat the field as single element.""" if fieldname in self: - Element = HEADER.get(fieldname, HeaderElement) - return Element.parse(super().__getitem__(fieldname)) + element_cls = HEADER.get(fieldname, HeaderElement) + return element_cls.parse(super().__getitem__(fieldname)) return default def get_element(self, fieldname: str, which: str | None = None, default: object | None = None) -> Any: @@ -90,9 +87,10 @@ def set_element(self, fieldname: str, *args, **kwargs) -> None: def append_element(self, fieldname: str, *args, **kwargs) -> None: self.append(fieldname, bytes(self.create_element(fieldname, *args, **kwargs))) - def create_element(self, fieldname: str, *args, **kwargs) -> HeaderElement: - Element = HEADER.get(fieldname, HeaderElement) - return Element(*args, **kwargs) + @classmethod + def create_element(cls, fieldname: str, *args, **kwargs) -> HeaderElement: + element_cls = HEADER.get(fieldname, HeaderElement) + return element_cls(*args, **kwargs) def values(self, *key) -> list[Any | str]: if not key: @@ -103,21 +101,21 @@ def values(self, *key) -> list[Any | str]: def append(self, _name: str, _value: bytes | str, **params) -> None: _value = self.formatvalue(_value) if params: - Element = HEADER.get(_name, HeaderElement) - parts = [_value or b''] + [Element.formatparam(k.encode(), v) for k, v in params.items()] + element_cls = HEADER.get(_name, HeaderElement) + parts = [_value or b''] + [element_cls.formatparam(k.encode(), v) for k, v in params.items()] _value = b'; '.join(parts) if _name not in self or not self[_name]: self[_name] = _value else: - Element = HEADER.get(_name, HeaderElement) - self[_name] = Element.join([super().__getitem__(_name), _value]) + element_cls = HEADER.get(_name, HeaderElement) + self[_name] = element_cls.join([super().__getitem__(_name), _value]) def merge(self, other: Headers) -> None: other = self.__class__(other) for key in other: - Element = HEADER.get(key, HeaderElement) - self[key] = Element.merge(self.elements(key), other.elements(key)) + element_cls = HEADER.get(key, HeaderElement) + self[key] = element_cls.merge(self.elements(key), other.elements(key)) def set(self, headers: dict[str, str]) -> None: self.clear() @@ -152,11 +150,11 @@ def parse(self, data: bytes) -> None: while lines and lines[0].startswith((b' ', b'\t')): value.append(lines.pop(0)[1:]) value = b''.join(value).rstrip() - Element = HEADER.get(name, HeaderElement) + element_cls = HEADER.get(name, HeaderElement) name = name.decode('ascii') if name in self: - value = Element.join([super().__getitem__(name), value]) + value = element_cls.join([super().__getitem__(name), value]) super().__setitem__(name, value) if len(self) >= self.max_header_count: raise InvalidHeaderSize(_('Maximum allowed header count (%d) reached.'), self.max_header_count) @@ -173,15 +171,13 @@ def __items(self): def __encoded_items(self): for key, values in self.items(): - Element = HEADER.get(key, HeaderElement) - if Element is not HeaderElement: - key = Element.name - key = key.encode('ascii', 'ignore') - if Element.list_element: - for value in Element.split(values): - yield key, value + element_cls = HEADER.get(key, HeaderElement) + name = (element_cls.name if element_cls is not HeaderElement else key).encode('ascii', 'ignore') + if element_cls.list_element: + for value in element_cls.split(values): + yield name, value else: - yield key, values + yield name, values def __repr__(self) -> str: return f'' diff --git a/httoop/header/messaging.py b/httoop/header/messaging.py index a5336b19..7533c88d 100644 --- a/httoop/header/messaging.py +++ b/httoop/header/messaging.py @@ -2,7 +2,7 @@ from __future__ import annotations import re -from typing import Any +from typing import Any, ClassVar from httoop.codecs import lookup from httoop.exceptions import InvalidDate, InvalidHeader @@ -12,6 +12,9 @@ class CodecElement: + value: str + mimetype: str + name: str CODECS = None raise_on_missing_codec = True @@ -28,7 +31,8 @@ def codec(self) -> Any: except AttributeError: mimetype = self.value - for encoding in (self.value, mimetype): + for val in (self.value, mimetype): + encoding = val if self.CODECS is not None: encoding = self.CODECS.get(encoding) if not isinstance(encoding, (bytes, str)): @@ -93,7 +97,7 @@ class ContentDisposition(HeaderElement, name='Content-Disposition'): is_response_header = True - from httoop.date import Date + from httoop.date import Date # noqa: PLC0415 @property def filename(self) -> str | None: @@ -143,7 +147,7 @@ class ContentEncoding(CodecElement, HeaderElement, name='Content-Encoding'): is_response_header = True # IANA assigned HTTP Content-Encoding values - CODECS = { + CODECS: ClassVar = { 'gzip': 'application/gzip', 'deflate': 'application/zlib', # TODO: implement the following @@ -279,7 +283,7 @@ class Host(HeaderElement): @property def is_ip4(self) -> bool: - from socket import AF_INET, inet_pton + from socket import AF_INET, inet_pton # noqa: PLC0415 try: inet_pton(AF_INET, self.host) @@ -289,7 +293,7 @@ def is_ip4(self) -> bool: @property def is_ip6(self) -> bool: - from socket import AF_INET6, inet_pton + from socket import AF_INET6, inet_pton # noqa: PLC0415 try: inet_pton(AF_INET6, self.host) @@ -370,7 +374,7 @@ class SetCookie(_ListElement, _CookieElement, name='Set-Cookie'): is_response_header = True - from httoop.date import Date + from httoop.date import Date # noqa: PLC0415 @classmethod def split(cls, fieldvalue: bytes) -> list[bytes]: @@ -398,21 +402,21 @@ def persistent(self) -> bool: return 'max-age' in self.params or 'expires' in self.params @property - def max_age(self) -> None: + def max_age(self) -> int | None: if self.params.get('max-age'): try: return integer(self.params['max-age']) except ValueError: - raise InvalidHeader(_('Cookie: max-age is not a number: %r'), self.params['max-age']) + raise InvalidHeader(_('Cookie: max-age is not a number: %r'), self.params['max-age']) from None return None @property - def expires(self) -> Date: + def expires(self) -> Date | None: if self.params.get('expires'): try: return self.Date(self.params['expires']) except InvalidDate: - raise InvalidHeader(_('Cookie: expires is not a valid date: %r'), self.params['expires']) + raise InvalidHeader(_('Cookie: expires is not a valid date: %r'), self.params['expires']) from None return None @@ -453,7 +457,7 @@ class TransferEncoding(_HopByHopElement, CodecElement, HeaderElement, name='Tran is_request_header = True # IANA assigned HTTP Transfer-Encoding values - CODECS = { + CODECS: ClassVar = { 'chunked': False, 'compress': NotImplementedError, 'deflate': 'application/zlib', diff --git a/httoop/header/ranges.py b/httoop/header/ranges.py index f5cd4362..322f4f11 100644 --- a/httoop/header/ranges.py +++ b/httoop/header/ranges.py @@ -38,7 +38,7 @@ def compose(self) -> bytes: @classmethod def parse(cls, elementstr: bytes) -> ContentRange: value, start, end, complete_length = None, None, None, None - try: + try: # noqa: PLW0717 value, content_range = elementstr.split(None, 1) if value != b'bytes': raise InvalidHeader(_('Only "bytes" Content-Range supported')) @@ -56,8 +56,8 @@ def parse(cls, elementstr: bytes) -> ContentRange: raise ValueError() if complete_length is None and start is None: raise ValueError() - except ValueError: - raise InvalidHeader(_('Content-Range: %r'), elementstr) + except ValueError as exc: + raise InvalidHeader(_('Content-Range: %r'), elementstr) from exc return cls(value.decode('ISO8859-1'), (start, end), complete_length) @@ -91,8 +91,8 @@ def parse(cls, elementstr: bytes) -> Range: stop = integer(stop) if stop else None if (start and start < 0) or (stop and stop < 0): raise ValueError() - except ValueError: - raise InvalidHeader(_('no range number.')) + except ValueError as exc: + raise InvalidHeader(_('no range number.')) from exc if start is not None and stop is not None and stop <= start: raise InvalidHeader(_('range start must be smaller than end.')) if not start and not stop: @@ -114,10 +114,12 @@ def prevent_denial_of_service(self) -> None: raise InvalidHeader(_('duplicated range.')) byterange.update(range_) - if self.stddev([(x[1] or float('inf')) - (x[0] or 0) for x in self.ranges]) > 2.0: + max_dev = 2.0 + if self.stddev([(x[1] or float('inf')) - (x[0] or 0) for x in self.ranges]) > max_dev: raise InvalidHeader(_('ranges exceeding high standard deviation')) - def stddev(self, xs: list[int | float]) -> float: + @classmethod + def stddev(cls, xs: list[int | float]) -> float: def average(xs): return sum(xs) * 1.0 / len(xs) @@ -140,5 +142,5 @@ def positions(self) -> Iterator[tuple[int, int, int] | tuple[int, int, None]]: def get_range_content(self, fd: BytesIO) -> Iterator[bytes]: for offset, whence, length in self.positions: fd.seek(offset, whence) - length = () if length is None else (length,) - yield fd.read(*length) + args = () if length is None else (length,) + yield fd.read(*args) diff --git a/httoop/header/security.py b/httoop/header/security.py index 12460bff..6357cbc4 100644 --- a/httoop/header/security.py +++ b/httoop/header/security.py @@ -25,7 +25,7 @@ class ContentSecurityPolicy(HeaderElement, name='Content-Security-Policy'): is_response_header = True RE_SPLIT = re.compile(rb';') - RE_PARAMS = re.compile(b'\\s+') + RE_PARAMS = re.compile(rb'\s+') def compose(self) -> bytes: return b'%s %s; ' % (self.value.encode('ISO8859-1'), b' '.join(self.params.keys())) @@ -82,7 +82,7 @@ class FrameOptions(HeaderElement, name='X-Frame-Options'): is_response_header = True - RE_PARAMS = re.compile(b'\\s+') + RE_PARAMS = re.compile(rb'\s+') @property def deny(self) -> bool: @@ -93,9 +93,9 @@ def same_origin(self) -> bool: return self.value.upper() == 'SAMEORIGIN' @property - def allow_from(self) -> list[HTTPS]: + def allow_from(self) -> list[HTTPS] | None: if self.value.upper() == 'ALLOW-FROM': - from httoop.uri import URI + from httoop.uri import URI # noqa: PLC0415 return [URI(uri) for uri in self.params] return None diff --git a/httoop/messages/body.py b/httoop/messages/body.py index ea5aa506..e13712ba 100644 --- a/httoop/messages/body.py +++ b/httoop/messages/body.py @@ -246,10 +246,11 @@ def __content_iter(self): if not data: continue if isinstance(data, str): - data = data.encode(self.encoding) - elif not isinstance(data, bytes): # pragma: no cover + yield data.encode(self.encoding) + elif isinstance(data, bytes): + yield data + else: # pragma: no cover raise TypeError(f'Iterable contained non-bytes: {type(data).__name__!r}') - yield data finally: self.seek(t) diff --git a/httoop/messages/protocol.py b/httoop/messages/protocol.py index 33f9f39f..abb41b8a 100644 --- a/httoop/messages/protocol.py +++ b/httoop/messages/protocol.py @@ -16,7 +16,7 @@ __all__ = ('Protocol',) -class Protocol(Semantic): +class Protocol(Semantic): # noqa: PLW1641 """The HTTP protocol version.""" __slots__ = ('__protocol', 'name') @@ -57,7 +57,7 @@ def parse(self, protocol: bytes) -> None: try: self.__protocol = (integer(major), integer(minor)) except ValueError: - raise InvalidLine(_('Invalid HTTP protocol: %r'), protocol.decode('ISO8859-1')) + raise InvalidLine(_('Invalid HTTP protocol: %r'), protocol.decode('ISO8859-1')) from None def compose(self) -> bytes: return b'%s/%d.%d' % (self.name, self.major, self.minor) diff --git a/httoop/messages/request.py b/httoop/messages/request.py index 0effd8ac..b2d0a1d1 100644 --- a/httoop/messages/request.py +++ b/httoop/messages/request.py @@ -64,7 +64,7 @@ def parse(self, line: bytes) -> None: try: method, uri, version = bits except ValueError: - raise InvalidLine(_('Invalid request line: %r'), line.decode('ISO8859-1')) + raise InvalidLine(_('Invalid request line: %r'), line.decode('ISO8859-1')) from None # protocol version super().parse(version) diff --git a/httoop/messages/response.py b/httoop/messages/response.py index 38b25cbc..0f0a28ca 100644 --- a/httoop/messages/response.py +++ b/httoop/messages/response.py @@ -49,7 +49,7 @@ def parse(self, line: bytes) -> None: try: version, status = bits except ValueError: - raise InvalidLine(_('Invalid response line: %r'), line.decode('ISO8859-1')) + raise InvalidLine(_('Invalid response line: %r'), line.decode('ISO8859-1')) from None # version super().parse(version) diff --git a/httoop/meta.py b/httoop/meta.py index 98d72be5..911c82f7 100644 --- a/httoop/meta.py +++ b/httoop/meta.py @@ -6,7 +6,7 @@ __all__ = ['Semantic'] -class Semantic: +class Semantic: # noqa: PLW1641 """Implements the HTTP Semantic interface.""" __slots__ = () diff --git a/httoop/parser.py b/httoop/parser.py index 9f5f4816..6ae5ad61 100644 --- a/httoop/parser.py +++ b/httoop/parser.py @@ -96,6 +96,7 @@ def on_body_complete(self) -> None: def on_message_complete(self) -> Response | Request: message = self.message + assert self.message # noqa: S101 self.message = None return message @@ -111,7 +112,7 @@ def parse(self, data: bytes) -> tuple[tuple[Request, Response]] | tuple[tuple[Re try: return tuple(x for x in self._parse() if x is not None) except (InvalidHeader, InvalidLine, InvalidURI, InvalidBody) as exc: - raise BAD_REQUEST(str(exc)) + raise BAD_REQUEST(str(exc)) from exc def _parse(self) -> Iterator[Response | tuple[Request, Response] | None]: while self.buffer: @@ -149,10 +150,11 @@ def parse_startline(self) -> bool | None: requestline, self.buffer = self.buffer.split(self.line_end, 1) # parse request line + assert self.message # noqa: S101 try: self.message.parse(bytes(requestline)) except (InvalidLine, InvalidURI) as exc: - raise BAD_REQUEST(str(exc)) + raise BAD_REQUEST(str(exc)) from exc def parse_headers(self) -> bool | None: # empty headers? @@ -177,7 +179,7 @@ def _parse_single_headers(self) -> None: rest += self.buffer[-len(self.line_end):] else: headers, _, rest = self.buffer.rpartition(self.line_end) - if headers and _ and rest[:1] not in (b'', b'\t', b' '): + if headers and _ and bytes(rest[:1]) not in {b'', b'\t', b' '}: self.buffer = rest self._parse_header(headers) @@ -187,11 +189,12 @@ def _parse_header(self, headers: bytearray) -> None: if self.state['_header_section_size'] >= self.max_header_section_size: raise REQUEST_HEADER_FIELDS_TOO_LARGE(_('Maximum allowed header section size (%d) reached.') % (self.max_header_section_size,)) try: + assert self.message # noqa: S101 self.message.headers.parse(bytes(headers)) except InvalidHeaderSize as exc: - raise REQUEST_HEADER_FIELDS_TOO_LARGE(str(exc)) + raise REQUEST_HEADER_FIELDS_TOO_LARGE(str(exc)) from exc except InvalidHeader as exc: - raise BAD_REQUEST(str(exc)) + raise BAD_REQUEST(str(exc)) from exc def parse_body(self) -> bool | None: if self.message_length is None and not self.chunked: @@ -207,6 +210,7 @@ def determine_message_length(self) -> None: # RFC 2616 Section 4.4 # get message length + assert self.message # noqa: S101 message = self.message if 'Transfer-Encoding' in message.headers and 'Content-Length' in message.headers: raise BAD_REQUEST(_('Invalid Content-Length and Transfer-Encoding combination header.')) @@ -223,16 +227,17 @@ def determine_message_length(self) -> None: try: self.message_length = integer(message.headers.get('Content-Length', '0')) except ValueError: - raise BAD_REQUEST(_('Invalid Content-Length header.')) + raise BAD_REQUEST(_('Invalid Content-Length header.')) from None if self.message_length > self.max_body_size: raise PAYLOAD_TOO_LARGE(_('Maximum content size (%d) reached') % (self.max_body_size,)) def parse_body_with_message_length(self) -> bool | None: + assert self.message # noqa: S101 body, self.buffer = self.buffer[:self.message_length], self.buffer[self.message_length:] try: self.message.body.parse(bytes(body)) except InvalidBodySize as exc: - raise PAYLOAD_TOO_LARGE(str(exc)) + raise PAYLOAD_TOO_LARGE(str(exc)) from exc blen = len(body) unfinished = blen < self.message_length @@ -244,6 +249,7 @@ def parse_body_with_message_length(self) -> bool | None: return None def parse_chunked_body(self) -> bool: + assert self.message # noqa: S101 if self.state['trailer']: return self.parse_trailers() if self.line_end not in self.buffer: @@ -260,7 +266,7 @@ def parse_chunked_body(self) -> bool: try: self.message.body.parse(bytes(body_part)) except InvalidBodySize as exc: - raise PAYLOAD_TOO_LARGE(str(exc)) + raise PAYLOAD_TOO_LARGE(str(exc)) from exc self.buffer = rest_chunk if chunk_size == 0: @@ -283,7 +289,7 @@ def __parse_chunk_size(self): raise ValueError() except (ValueError, OverflowError): exc = InvalidHeader(_('Invalid chunk size: %r'), chunk_size_.decode('ISO8859-1')) - raise BAD_REQUEST(str(exc)) + raise BAD_REQUEST(str(exc)) from exc else: return chunk_size, rest_chunk @@ -305,7 +311,7 @@ def parse_trailers(self) -> bool: self.trailers.parse(bytes(trailers)) except InvalidHeader as exc: exc = InvalidHeader(_('Invalid trailers: %r'), str(exc)) - raise BAD_REQUEST(str(exc)) + raise BAD_REQUEST(str(exc)) from exc for forbidden in Trailer.forbidden_headers: if forbidden in self.trailers: @@ -317,6 +323,7 @@ def parse_trailers(self) -> bool: def merge_trailer_into_header(self) -> None: message = self.message + assert self.message # noqa: S101 for name in message.headers.values('Trailer'): value = self.trailers.pop(name, None) if value is not None: @@ -327,18 +334,21 @@ def merge_trailer_into_header(self) -> None: del self.trailers def set_body_content_encoding(self) -> None: + assert self.message # noqa: S101 if 'Content-Encoding' in self.message.headers: try: self.message.body.content_encoding = self.message.headers.element('Content-Encoding') - self.message.body.content_encoding.codec # pylint: disable=W0104 + self.message.body.content_encoding.codec # noqa: B018 except Invalid as exc: - raise NOT_IMPLEMENTED(str(exc)) + raise NOT_IMPLEMENTED(str(exc)) from exc def set_body_content_type(self) -> None: + assert self.message # noqa: S101 if 'Content-Type' in self.message.headers: self.message.body.mimetype = self.message.headers.element('Content-Type') def set_content_length(self) -> None: + assert self.message # noqa: S101 if 'Content-Length' not in self.message.headers: self.message.headers['Content-Length'] = str(len(self.message.body)).encode('ASCII') if self.chunked: diff --git a/httoop/semantic/message.py b/httoop/semantic/message.py index 8a277530..c723ce64 100644 --- a/httoop/semantic/message.py +++ b/httoop/semantic/message.py @@ -1,9 +1,23 @@ +from __future__ import annotations + from contextlib import contextmanager -from typing import Iterator +from typing import TYPE_CHECKING, Iterator + + +if TYPE_CHECKING: + from httoop import Request, Response + + +try: + from typing import override +except ImportError: # < Py 3.11 + from typing_extensions import override class ComposedMessage: + message: Request | Response + # FIXME: use it @property def close(self): # pragma: no cover @@ -44,6 +58,7 @@ def chunked(self, chunked) -> None: if not te: self.message.headers.pop('Transfer-Encoding') + @override @contextmanager def _composing(self) -> Iterator[None]: yield diff --git a/httoop/semantic/request.py b/httoop/semantic/request.py index bc9a1281..3d3b98f2 100644 --- a/httoop/semantic/request.py +++ b/httoop/semantic/request.py @@ -11,6 +11,7 @@ class ComposedRequest(ComposedMessage): USER_AGENT = UserAgentHeader + message: Request def __init__(self, request: Request) -> None: self.message = request diff --git a/httoop/semantic/response.py b/httoop/semantic/response.py index 00b0c4d5..6d1e248f 100644 --- a/httoop/semantic/response.py +++ b/httoop/semantic/response.py @@ -1,3 +1,4 @@ +# ruff: file-ignore[PLR2004] from __future__ import annotations from email.generator import _make_boundary as make_boundary @@ -111,7 +112,8 @@ def prepare_range(self, range_: Range) -> bool: response.headers['Content-Length'] = str(len(response.body)).encode('ASCII') # FIXME: len(response.body) causes the whole body to be generated return True - def multipart_byteranges(self, range_body: Iterator[Any], range_: Range, content_length: str, content_type: str) -> Iterator[Body]: + @staticmethod + def multipart_byteranges(range_body: Iterator[Any], range_: Range, content_length: str, content_type: str) -> Iterator[Body]: for content, byterange in zip(range_body, range_.ranges): body = Body(content) body.headers['Content-Type'] = content_type diff --git a/httoop/server/__init__.py b/httoop/server/__init__.py index 1e0ba124..40799a85 100644 --- a/httoop/server/__init__.py +++ b/httoop/server/__init__.py @@ -18,6 +18,7 @@ class ServerStateMachine(StateMachine): Message = Request + message: Request HTTP2 = None def __init__(self, scheme: str, host: str, port: int, *, strict: bool = True, max_uri_length: float = 8192) -> None: @@ -41,6 +42,8 @@ def on_message_started(self) -> None: def on_message_complete(self) -> tuple[httoop.messages.request.Request, httoop.messages.response.Response]: request = super().on_message_complete() response = self.response + assert request is not None # noqa: S101 + assert response is not None # noqa: S101 self.request = None self.response = None return (request, response) @@ -62,6 +65,7 @@ def on_startline_complete(self) -> None: def on_uri_complete(self) -> None: super().on_uri_complete() + assert self.request # noqa: S101 self._check_uri_max_length(bytes(self.request.uri)) self.sanitize_request_uri_path() self.validate_request_uri_scheme() @@ -91,6 +95,8 @@ def check_request_protocol(self) -> None: def set_response_protocol(self) -> None: # set appropriate response protocol version + assert self.message # noqa: S101 + assert self.response # noqa: S101 self.response.protocol = min(self.message.protocol, ServerProtocol) def _check_uri_max_length(self, uri: bytearray | bytes) -> None: @@ -114,6 +120,7 @@ def validate_request_uri_scheme(self) -> None: self.message.uri.port = self._default_port def set_server_response_header(self) -> None: + assert self.response # noqa: S101 self.response.headers.setdefault('Server', ServerHeader) def check_host_header_exists(self) -> None: @@ -150,6 +157,7 @@ def is_http2_upgrade(): if all(is_http2_upgrade()): if self.HTTP2 is None: return + assert self.response # noqa: S101 self.response.headers['Upgrade'] = 'h2c' self.response.headers['Connection'] = 'Upgrade' self.__class__ = self.HTTP2 diff --git a/httoop/status/__init__.py b/httoop/status/__init__.py index a15a94a1..719917c9 100644 --- a/httoop/status/__init__.py +++ b/httoop/status/__init__.py @@ -17,5 +17,5 @@ __all__ = ['REASONS', 'ClientErrorStatus', 'InformationalStatus', 'RedirectStatus', 'ServerErrorStatus', 'Status', 'StatusException', 'SuccessStatus'] for member in STATUSES.values(): - __all__.append(member.__name__) + __all__ += member.__name__ # noqa: PLE0605 globals()[member.__name__] = member diff --git a/httoop/status/status.py b/httoop/status/status.py index d89a50a1..779bcc43 100644 --- a/httoop/status/status.py +++ b/httoop/status/status.py @@ -4,6 +4,7 @@ .. seealso:: :rfc:`2616#section-6.2` .. seealso:: :rfc:`2616#section-10` """ +# ruff: file-ignore[PLR2004] from __future__ import annotations @@ -18,7 +19,7 @@ STATUSES = {} -class Status(Semantic): +class Status(Semantic): # noqa: PLW1641 """ A HTTP Status. diff --git a/httoop/status/types.py b/httoop/status/types.py index 5169fcb0..0890f63b 100644 --- a/httoop/status/types.py +++ b/httoop/status/types.py @@ -9,7 +9,7 @@ from httoop.status.status import Status -class StatusException(Status, Exception): +class StatusException(Status, Exception): # noqa: N818 """ This class represents a small HTTP Response message for error handling purposes @@ -23,7 +23,7 @@ def headers(self) -> dict[str, str]: @property def body(self): if not hasattr(self, '_body'): - from httoop.messages.body import Body + from httoop.messages.body import Body # noqa: PLC0415 self._body = Body(mimetype='application/json') self._body.data = self.to_dict() diff --git a/httoop/uri/percent_encoding.py b/httoop/uri/percent_encoding.py index 8482ccc6..7c12193f 100644 --- a/httoop/uri/percent_encoding.py +++ b/httoop/uri/percent_encoding.py @@ -1,4 +1,4 @@ -from typing import Iterator +from typing import ClassVar, Iterator class Percent: @@ -11,7 +11,7 @@ class Percent: b"!#$&'()*+,/:;=?@[]" """ - HEX_MAP = {(a + b).encode('ASCII'): bytes((int(a + b, 16),)) for a in '0123456789ABCDEFabcdef' for b in '0123456789ABCDEFabcdef'} + HEX_MAP: ClassVar = {(a + b).encode('ASCII'): bytes((int(a + b, 16),)) for a in '0123456789ABCDEFabcdef' for b in '0123456789ABCDEFabcdef'} # ABNF GEN_DELIMS = b':/?#[]@' @@ -35,18 +35,19 @@ def unquote(cls, data: bytes) -> bytes: @classmethod def _decode_iter(cls, data: bytes) -> Iterator[bytes]: - data = data.split(b'%') - yield data.pop(0) - for item in data: - try: - yield cls.HEX_MAP[item[:2]] + datas = data.split(b'%') + yield datas.pop(0) + for item in datas: + mapped = cls.HEX_MAP.get(item[:2]) + if mapped is not None: + yield mapped yield item[2:] - except KeyError: + else: yield b'%' yield item @classmethod def quote(cls, data: bytes, charset: bytes = UNRESERVED) -> bytes: - charset = {bytes((c,)) for c in iter(charset)} - {b'%'} - data = (bytes((d,)) for d in iter(data)) - return b''.join(b'%%%X' % (ord(d),) if d not in charset else d for d in data) + charset_s = {bytes((c,)) for c in iter(charset)} - {b'%'} + datas = (bytes((d,)) for d in iter(data)) + return b''.join(b'%%%X' % (ord(d),) if d not in charset_s else d for d in datas) diff --git a/httoop/uri/uri.py b/httoop/uri/uri.py index c9572179..201ef416 100644 --- a/httoop/uri/uri.py +++ b/httoop/uri/uri.py @@ -9,7 +9,7 @@ import re from socket import AF_INET, AF_INET6, inet_ntop, inet_pton -from typing import TYPE_CHECKING, Any, Iterator +from typing import TYPE_CHECKING, Any, ClassVar, Iterator from httoop.exceptions import InvalidURI from httoop.meta import Semantic @@ -21,14 +21,16 @@ if TYPE_CHECKING: from httoop.uri.http import HTTP +_MAX_PORT = 65535 -class URI(Semantic): + +class URI(Semantic): # noqa: PLW1641 """Uniform Resource Identifier.""" __slots__ = ('scheme', 'username', 'password', 'host', '_port', 'path', 'query_string', 'fragment') # noqa: RUF023 slots = __slots__ - SCHEMES = {} + SCHEMES: ClassVar = {} PORT = None encoding = 'UTF-8' @@ -65,10 +67,10 @@ def port(self, port) -> None: if port: try: port = integer(port) - if not 0 < port <= 65535: + if not 0 < port <= _MAX_PORT: raise ValueError except ValueError: - raise InvalidURI(_('Invalid port: %r'), port) # TODO: TypeError + raise InvalidURI(_('Invalid port: %r'), port) from None # TODO: TypeError self._port = port def __init__(self, uri: Any | None = None, *args, **kwargs) -> None: @@ -138,7 +140,7 @@ def abspath(self) -> None: >>> u = URI(b'/foo/../bar/.'); u.abspath(); u.path == u'/bar/' True """ - path = re.sub('\\/{2,}', '/', self.path) # remove // + path = re.sub(r'\/{2,}', '/', self.path) # remove // if not path: return unsplit = [] @@ -176,17 +178,17 @@ def dict(self): slots = (key.lstrip('_') for key in self.slots) return {key: getattr(self, key) for key in slots} - @dict.setter + @dict.setter # noqa: A003 def dict(self, uri) -> None: - for key in self.slots: - key = key.lstrip('_') + for name in self.slots: + key = name.lstrip('_') setattr(self, key, uri.get(key, '')) @property def tuple(self): return tuple(getattr(self, key) for key in self.slots) - @tuple.setter + @tuple.setter # noqa: A003 def tuple(self, tuple_) -> None: (self.scheme, self.username, self.password, self.host, self.port, self.path, self.query_string, self.fragment) = tuple_ @@ -240,10 +242,10 @@ def parse(self, uri: bytes) -> None: try: scheme = scheme.decode('ascii').lower() except UnicodeDecodeError: # pragma: no cover - raise InvalidURI(_('Invalid scheme: must be ASCII.')) + raise InvalidURI(_('Invalid scheme: must be ASCII.')) from None if scheme and scheme.strip('abcdefghijklmnopqrstuvwxyz0123456789.-+'): - raise InvalidURI(_('Invalid scheme: must only contain alphanumeric letters or plus, dash, dot.')) + raise InvalidURI(_('Invalid scheme: must only contain alphanumeric letters or plus, dash, dot.')) from None if query_string: query_string = QueryString.encode(QueryString.decode(query_string, self.encoding), self.encoding) @@ -273,14 +275,14 @@ def _unquote_host(self, host: bytes) -> str: try: return '[%s]' % host.decode('ascii') except UnicodeDecodeError: # pragma: no cover - raise InvalidURI(_('Invalid IPvFuture address: must be ASCII.')) - raise InvalidURI(_('Invalid IP address in URI.')) + raise InvalidURI(_('Invalid IPvFuture address: must be ASCII.')) from None + raise InvalidURI(_('Invalid IP address in URI.')) from None # IPv4 if all(x.isdigit() for x in host.split(b'.')): try: return inet_ntop(AF_INET, inet_pton(AF_INET, host.decode('ascii'))) except (OSError, UnicodeDecodeError): - raise InvalidURI(_('Invalid IPv4 address in URI.')) + raise InvalidURI(_('Invalid IPv4 address in URI.')) from None if host.strip(Percent.UNRESERVED + Percent.SUB_DELIMS + b'%'): raise InvalidURI(_('Invalid URI host.')) @@ -290,7 +292,7 @@ def _unquote_host(self, host: bytes) -> str: try: return host.encode('ascii').decode('idna').lower() except UnicodeError: # pragma: no cover - raise InvalidURI(_('Invalid host.')) + raise InvalidURI(_('Invalid host.')) from None def compose(self) -> bytes: return b''.join(self._compose_absolute_iter()) @@ -324,10 +326,10 @@ def _compose_authority_iter(self) -> Iterator[bytes]: def _compose_relative_iter(self) -> Iterator[bytes]: """Composes the relative URI beginning with the path.""" scheme, path, query_string, quote, fragment = self.scheme, self.path, self.query_string, self.quote, self.fragment - PATH = Percent.PATH + path_quote_chars = Percent.PATH if not scheme and not path.startswith('/'): - PATH = b''.join({bytes((c,)) for c in iter(PATH)} - {b':', b'@'}) - yield b'/'.join(quote(x, PATH) for x in path.split('/')) + path_quote_chars = b''.join({bytes((c,)) for c in iter(path_quote_chars)} - {b':', b'@'}) + yield b'/'.join(quote(x, path_quote_chars) for x in path.split('/')) if query_string: yield b'?' yield query_string.encode(self.encoding) diff --git a/httoop/util.py b/httoop/util.py index a0de78f1..1699c90b 100644 --- a/httoop/util.py +++ b/httoop/util.py @@ -5,7 +5,7 @@ import codecs import string from binascii import Error as Base64Error, a2b_base64, b2a_base64 -from typing import Any, Callable +from typing import IO, Any, Callable try: @@ -18,7 +18,7 @@ def _(x: str) -> str: return x -__all__ = ['CaseInsensitiveDict', 'IFile', '_', 'integer', 'sanitize_encoding', 'to_unicode', 'base64', 'base64_decode', 'Base64Error'] +__all__ = ['Base64Error', 'CaseInsensitiveDict', 'IFile', '_', 'base64', 'base64_decode', 'integer', 'sanitize_encoding', 'to_unicode'] KNOWN_ENCODINGS = { 'cp1254', 'cp949', 'cp865', 'cp1257', 'euc_jp', 'cp1250', 'mac-cyrillic', 'mac-latin2', 'cp866', 'cp857', @@ -39,9 +39,9 @@ def base64(data: bytes): def base64_decode(data: bytes): try: - return a2b_base64(data, strict_mode=True) + return a2b_base64(data, strict_mode=True) # type: ignore[unknown-argument] except TypeError: # Py < 3.11 - import base64 + import base64 # noqa: PLC0415 return base64.b64decode(data, validate=True) @@ -68,7 +68,7 @@ def to_unicode(string: bytes | str | None) -> str: def if_has(func: Callable) -> Callable: def _decorated(self, *args, **kwargs): - if hasattr(self.fd, func.__name__): + if hasattr(self.fd, getattr(func, '__name__', '')): return func(self, *args, **kwargs) return False @@ -76,7 +76,7 @@ def _decorated(self, *args, **kwargs): def integer(number: str | bytes, base=10) -> int: - """ + r""" The native Python integer parsing from string allows to many forms which are not allowed by the protocol. >>> integer(bytearray(b'10')) @@ -124,9 +124,8 @@ def integer(number: str | bytes, base=10) -> int: """ if isinstance(number, int): return int(number) - if not number.isdigit(): - if base != 16 or number.strip(string.hexdigits if isinstance(number, str) else string.hexdigits.encode('ASCII')): - raise ValueError() + if not number.isdigit() and (base != 16 or number.strip(string.hexdigits if isinstance(number, str) else string.hexdigits.encode('ASCII'))): # noqa: PLR2004 + raise ValueError() zero = '0' if isinstance(number, str) else b'0' if number != zero and len(number.lstrip(zero)) != len(number): raise ValueError() @@ -137,6 +136,7 @@ class IFile: """The file interface.""" __slots__ = ('fd',) + fd: IO @property def name(self) -> None: @@ -222,8 +222,8 @@ def __contains__(self, key: bytes | str) -> bool: def get(self, key: bytes | str, default: Any | None = None) -> Any: return dict.get(self, self.formatkey(key), default) - def update(self, E: dict[str, str]) -> None: - for key in E: + def update(self, E: dict[str, str]) -> None: # noqa: N803 + for key in E: # noqa: PLC0206 self[self.formatkey(key)] = self.formatvalue(E[key]) # def setdefault(self, key: str, x: Optional[Union[UserAgent, Server, str, bytes]]=None) -> bytes: diff --git a/pyproject.toml b/pyproject.toml index 40f427d8..eca3b788 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -24,7 +24,6 @@ classifiers = [ "Operating System :: POSIX :: Linux", "Operating System :: MacOS :: MacOS X", "Operating System :: Microsoft :: Windows", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", @@ -39,11 +38,17 @@ classifiers = [ "Topic :: Software Development :: Libraries :: Python Modules", ] -requires-python = ">=3.8" +requires-python = ">=3.9" + +dependencies = [ + "typing_extensions>=4.4", +] [project.optional-dependencies] xml = ["defusedxml"] hal = ["uritemplate"] + +[dependency-groups] test = [ "pytest", "pytest-cov", @@ -70,6 +75,21 @@ exclude_lines = [ "pragma: no cover", ] +[tool.ty.terminal] +output-format = "concise" + +[tool.ty.environment] +python-version = "3.8" + +[tool.mypy] +python_version = "3.10" +strict = true +show_error_codes = true +explicit_package_bases = true +namespace_packages = true +mypy_path = ["httoop"] +files = ["httoop/"] + [tool.ruff] line-length = 180 target-version = "py37" @@ -81,6 +101,9 @@ preview = true select = ["E", "W", "F", "I", "D", "TD", "ALL"] ignore = [ # Invalid / don't care: + "RUF105", # noqa-comments + "RUF106", # rule-codes-in-suppression-comments: + "RUF201", # rule-codes-in-selectors "E501", # line-too-long "E203", # whitespace-before-punctuation "COM812", # missing-trailing-comma @@ -119,6 +142,7 @@ ignore = [ # "I", "S101", "Q000", "ANN", "ERA", "UP004", "ARG001", "ARG002", "FURB189", # subclass-builtin "TRY301", # raise-within-try + "RUF067", # non-empty-init-module # too much effort: "ANN001", # missing-type-function-argument @@ -142,52 +166,6 @@ ignore = [ "PLC2801", # unnecessary-dunder-call "TRY003", # raise-vanilla-args "TRY300", # try-consider-else - -# should be fixed! - "A001", # builtin-variable-shadowing - "A003", # builtin-attribute-shadowing - "B015", # useless-comparison - "B018", # useless-expression - "B904", # raise-without-from-inside-except - "D301", # escape-sequence-in-docstring - "DTZ001", # call-datetime-without-tzinfo - "DTZ004", # call-datetime-utcfromtimestamp - "FBT001", # boolean-type-hint-positional-argument - "FBT002", # boolean-default-value-positional-argument - "FBT003", # boolean-positional-value-in-call - "FURB113", # repeated-append - "N802", # invalid-function-name - "N803", # invalid-argument-name - "N805", # invalid-first-argument-name-for-method - "N806", # non-lowercase-variable-in-function - "N818", # error-suffix-on-exception-name - "PERF203", # try-except-in-loop - "PLC0206", # dict-index-missing-items - "PLC0415", # import-outside-top-level - "PLC1901", # compare-to-empty-string - "PLR1704", # redefined-argument-from-local - "PLR2004", # magic-value-comparison - "PLR6201", # literal-membership - "PLR6301", # no-self-use - "PLW0108", # unnecessary-lambda - "PLW0717", # too-many-statements-in-try-clause - "PLW1641", # eq-without-hash - "PLW2901", # redefined-loop-name - "PT011", # pytest-raises-too-broad - "PT012", # pytest-raises-with-multiple-statements - "PYI041", # redundant-numeric-union - "PYI056", # unsupported-method-call-on-all - "RUF005", # collection-literal-concatenation - "RUF012", # mutable-class-default - "RUF022", # unsorted-dunder-all - "RUF039", # unraw-re-pattern - "RUF067", # non-empty-init-module - "RUF069", # float-equality-comparison - "SIM102", # collapsible-if - "SIM108", # if-else-block-instead-of-if-exp - "SIM115", # open-file-with-context-handler - "RUF100", # unused-noqa - "S101", # assert ] @@ -228,6 +206,7 @@ task-tags = ["TODO", "FIXME"] "S404", # suspicious-subprocess-import "S405", # suspicious-xml-etree-import "S603", # subprocess-without-shell-equals-true + "PLR2004", # magic-value-comparison ] "httoop/status/*.py" = [ "N801", diff --git a/tests/api/test_body.py b/tests/api/test_body.py index e4da9428..5bc86673 100644 --- a/tests/api/test_body.py +++ b/tests/api/test_body.py @@ -3,24 +3,24 @@ import pytest -from httoop import Body +from httoop import Body, Request -def test_body_set_unicode(request_): +def test_body_set_unicode(request_: Request): request_.body = 'foobar' assert bytes(request_.body) == b'foobar' assert str(request_.body) == 'foobar' assert request_.body.fileable -def test_body_set_bytes(request_): +def test_body_set_bytes(request_: Request): request_.body = b'foobar' assert bytes(request_.body) == b'foobar' assert str(request_.body) == 'foobar' assert request_.body.fileable -def test_body_set_bytesio(request_): +def test_body_set_bytesio(request_: Request): b = BytesIO(b'ThisIsABytesIOBody') request_.body = b assert bytes(request_.body) == b'ThisIsABytesIOBody' @@ -28,7 +28,7 @@ def test_body_set_bytesio(request_): assert request_.body.fileable -def test_body_set_stringio(request_): +def test_body_set_stringio(request_: Request): s = StringIO('ThisIsAStringIOBody') request_.body = s assert bytes(request_.body) == b'ThisIsAStringIOBody' @@ -36,36 +36,35 @@ def test_body_set_stringio(request_): assert request_.body.fileable -def test_body_set_tempfile(request_): - tempfile = NamedTemporaryFile() - tempfile.write(b'ThisIsANamedTemporaryFile') - tempfile.flush() - request_.body = tempfile - assert len(request_.body) == 25 - assert request_.body == b'ThisIsANamedTemporaryFile' - assert request_.body.fileable - tempfile.close() +def test_body_set_tempfile(request_: Request): + with NamedTemporaryFile() as tempfile: + tempfile.write(b'ThisIsANamedTemporaryFile') + tempfile.flush() + request_.body = tempfile + assert len(request_.body) == 25 + assert request_.body == b'ThisIsANamedTemporaryFile' + assert request_.body.fileable -def test_body_set_bytearray(request_): +def test_body_set_bytearray(request_: Request): a = bytearray(b''.join([b'We', b'', b'are ', b'just', b' ', b'testing\t', b'ByteArrays!'])) request_.body = a assert bytes(request_.body) == b'Weare just testing\tByteArrays!' -def test_body_set_list(request_): +def test_body_set_list(request_: Request): ls = ['This ', 'is', b'\nsome', None, 'list\t', 'content', ''] request_.body = ls assert bytes(request_.body) == b'This is\nsomelist\tcontent' -def test_body_set_tuple(request_): +def test_body_set_tuple(request_: Request): t = ('Testing', ' ', 'a', 'tuple') request_.body = t assert bytes(request_.body) == b'Testing atuple' -def test_body_set_generator(request_): +def test_body_set_generator(request_: Request): def g(): yield 'This ' yield 'is' @@ -83,7 +82,7 @@ def g(): assert len(request_.body) == 19 -def test_body_set_body(request_): +def test_body_set_body(request_: Request): b = Body() b.mimetype = 'application/json' b.set('{}') @@ -93,27 +92,27 @@ def test_body_set_body(request_): assert request_.body.fd is b.fd -def test_body_iterencode(request_): +def test_body_iterencode(request_: Request): request_.body.mimetype = 'application/json' request_.body.iterencode({}) assert bytes(request_.body) == b'{}' -def tets_body_set_none(request_): +def tets_body_set_none(request_: Request): request_.body = 'Foobar' request_.body = None assert not request_.body assert bytes(request_.body) == b'' -def test_closed_body(request_): +def test_closed_body(request_: Request): b = BytesIO() b.close() - with pytest.raises(ValueError): + with pytest.raises(ValueError, match=r'I/O operation on closed file\.'): request_.body = b -def test_body_close_clear(request_): +def test_body_close_clear(request_: Request): request_.body = b'asfd asdf asdf asdf asdf asdf asdf' assert request_.body request_.body.close() @@ -122,7 +121,7 @@ def test_body_close_clear(request_): assert bytes(request_.body) == b'' -def test_empty_body_is_false(request_): +def test_empty_body_is_false(request_: Request): request_.body = 'foobar' assert request_.body assert len(request_.body) == 6 @@ -131,12 +130,12 @@ def test_empty_body_is_false(request_): assert len(request_.body) == 0 -def test_set_invalid_body(request_): +def test_set_invalid_body(request_: Request): with pytest.raises(TypeError): request_.body = 1 -def test_body_iter_list(request_): +def test_body_iter_list(request_: Request): request_.body = ['asdf', 'foo', 'Ba', 'Baz', None, 'blub'] assert next(request_.body) == b'asdf' assert next(request_.body) == b'foo' @@ -152,7 +151,7 @@ def test_body_iter_list(request_): assert next(request_.body) == b'blub' -def test_body_file_interface(request_): +def test_body_file_interface(request_: Request): assert request_.body.name is None request_.body.flush() request_.body = b'foo\nbar\nbaz\nblub' diff --git a/tests/api/test_codecs.py b/tests/api/test_codecs.py index 03a1f90a..2a6729df 100644 --- a/tests/api/test_codecs.py +++ b/tests/api/test_codecs.py @@ -1,5 +1,6 @@ import pytest +from httoop import Body from httoop.codecs import Codec, lookup, register from httoop.exceptions import DecodeError, EncodeError @@ -54,7 +55,7 @@ def test_body_invalid_hal(): hal.decode(b'{"_embedded": {"foo": 1}}').get_resource('foo') -def test_invalid_multipart(body): +def test_invalid_multipart(body: Body): body.mimetype = 'multipart/x; boundary=asdf' codec = body.mimetype.codec '--asdf\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\nfoo\r\n--asdf\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\nbar\r\n--asdf\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\nbaz\r\n--asdf--\r\n' diff --git a/tests/api/test_date.py b/tests/api/test_date.py index f698bb83..d3723234 100644 --- a/tests/api/test_date.py +++ b/tests/api/test_date.py @@ -1,12 +1,24 @@ +from __future__ import annotations + import datetime +from typing import TypedDict import pytest from httoop import Date, InvalidDate -dates = [{ - 'datetime': datetime.datetime(1994, 11, 6, 8, 49, 37), +class DateData(TypedDict): + datetime: datetime.datetime + timestamp: float + lt: Date + gt: Date + gmtime: tuple[int, int, int, int, int, int, int, int, int] + formats: set[str] + + +dates: list[DateData] = [{ + 'datetime': datetime.datetime(1994, 11, 6, 8, 49, 37, tzinfo=datetime.timezone.utc), 'timestamp': 784111777.0, 'lt': Date(784111776.0), 'gt': Date(784111778.0), @@ -35,7 +47,7 @@ def test_date_datetime(date, expected, lt, gt): def test_date_timestamp(date, expected, lt, gt): for d in (Date.parse(date.encode('utf-8')), Date(date)): assert d is not None - assert float(d) == expected + assert float(d) == expected # noqa: RUF069 assert d == expected assert d == Date(date) @@ -56,7 +68,7 @@ def test_date_gmtime(date, expected, lt, gt): def test_date_comparing_none(): - d = Date(datetime.datetime(1994, 11, 6, 8, 49, 37)) + d = Date(datetime.datetime(1994, 11, 6, 8, 49, 37, tzinfo=datetime.timezone.utc)) assert d == Date(d) assert not d == None # noqa: E711, SIM201 assert d > None diff --git a/tests/api/test_header.py b/tests/api/test_header.py index bf967c0e..95db8cb2 100644 --- a/tests/api/test_header.py +++ b/tests/api/test_header.py @@ -74,7 +74,7 @@ def test_pop_header(headers): def test_util_to_unicode(): - assert to_unicode(None) == '' + assert to_unicode(None) == '' # noqa: PLC1901 assert to_unicode(b'\xff') == '\xff' @@ -104,6 +104,6 @@ def test_header_append_params(headers): def test_headers_composing_rejects_invalid_values(headers): - dict.__setitem__(headers, 'Location', b'Foo\r\nSet-Cookie: foo=bar') # noqa: PLC2801 + dict.__setitem__(headers, 'Location', b'Foo\r\nSet-Cookie: foo=bar') with pytest.raises(InvalidHeader): bytes(headers) diff --git a/tests/api/test_method.py b/tests/api/test_method.py index 10716d37..9dd7bdb6 100644 --- a/tests/api/test_method.py +++ b/tests/api/test_method.py @@ -2,8 +2,10 @@ import pytest +from httoop import Request -def test_safe_methods(request_): + +def test_safe_methods(request_: Request): all_methods = ( ('GET', True, True), ('HEAD', True, True), @@ -20,7 +22,7 @@ def test_safe_methods(request_): @pytest.mark.xfail(reason='hash changing + fixed references') -def test_hashable_methods(request_): +def test_hashable_methods(request_: Request): methods = {} request_.method = 'GET' methods[request_.method] = 1 diff --git a/tests/api/test_protocol.py b/tests/api/test_protocol.py index 1543113a..c46539ff 100644 --- a/tests/api/test_protocol.py +++ b/tests/api/test_protocol.py @@ -1,25 +1,26 @@ import pytest +from httoop import Request from httoop.exceptions import InvalidLine -def test_protocol_tuple(request_): +def test_protocol_tuple(request_: Request): request_.protocol.parse(b'HTTP/1.0') assert request_.protocol == (1, 0) -def test_set_protocol_tuple(request_): +def test_set_protocol_tuple(request_: Request): request_.protocol = (1, 0) assert bytes(request_.protocol) == b'HTTP/1.0' -def test_protocol_minor_mayor(request_): +def test_protocol_minor_mayor(request_: Request): request_.protocol = (1, 0) assert request_.protocol.major == 1 assert request_.protocol.minor == 0 -def test_protocol_compare_bytes(request_): +def test_protocol_compare_bytes(request_: Request): request_.protocol = (1, 0) assert request_.protocol == b'HTTP/1.0' @@ -42,7 +43,7 @@ def test_invalid_protocol(request_, invalid): request_.protocol.set(invalid) -def test_protocol_comparision(request_): +def test_protocol_comparision(request_: Request): request_.protocol = (1, 2) assert request_.protocol < (2, 0) assert request_.protocol < 2 diff --git a/tests/api/test_slots.py b/tests/api/test_slots.py index 9e2f6d65..031705ef 100644 --- a/tests/api/test_slots.py +++ b/tests/api/test_slots.py @@ -33,6 +33,6 @@ def test_slots(class_, args): obj = class_(*args) assert obj.__slots__ is not None with pytest.raises(AttributeError): - obj.__dict__ + _ = obj.__dict__ with pytest.raises(AttributeError): obj.foo = True diff --git a/tests/api/test_statemachine.py b/tests/api/test_statemachine.py index b91e1dd1..c0f8a6e3 100644 --- a/tests/api/test_statemachine.py +++ b/tests/api/test_statemachine.py @@ -42,18 +42,18 @@ def test_max_uri_length(statemachine): statemachine.MAX_URI_LENGTH = 100 path = b'A' * (statemachine.MAX_URI_LENGTH - 5) statemachine.parse(b'GET /%s HTTP/1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n' % (path,)) + path = b'A' * statemachine.MAX_URI_LENGTH with pytest.raises(URI_TOO_LONG) as exc: - path = b'A' * statemachine.MAX_URI_LENGTH statemachine.parse(b'GET /%s HTTP/1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n' % (path,)) assert 'The maximum length of the request is 100' in str(exc.value.description) def test_max_request_line_length(statemachine): statemachine.MAX_URI_LENGTH = 100 + path = b'A' * statemachine.MAX_URI_LENGTH with pytest.raises(URI_TOO_LONG) as exc: - path = b'A' * statemachine.MAX_URI_LENGTH statemachine.parse(b'GET /%s HTTP/' % (path,)) - statemachine.parse(b'1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n') + # statemachine.parse(b'1.1\r\nHost: example.com\r\nContent-Length: 0\r\n\r\n') assert 'The maximum length of the request is 100' in str(exc.value.description) diff --git a/tests/api/test_status.py b/tests/api/test_status.py index eef8d274..f8bc9e2a 100644 --- a/tests/api/test_status.py +++ b/tests/api/test_status.py @@ -1,6 +1,7 @@ import pytest from httoop import Status +from httoop.messages import Response from httoop.status import ( CREATED, INTERNAL_SERVER_ERROR, @@ -66,7 +67,7 @@ def test_stats_int(response): assert int(response.status) == 100 -def test_status_tuple(response): +def test_status_tuple(response: Response): response.status = 200 assert response.status.code == 200 assert response.status.reason == 'OK' @@ -122,7 +123,7 @@ def test_status_parse(response): assert not response.status.reason -def test_status_type(response): +def test_status_type(response: Response): response.status = 100 assert response.status.informational assert not response.status.successful @@ -165,13 +166,13 @@ def test_invalid_status_code(code, response): response.status = code with pytest.raises(TypeError): response.status.code = code - from httoop.status import Status + from httoop.status import Status # noqa: PLC0415 with pytest.raises(TypeError): Status(code) def test_invalid_status_subclasses(): - from httoop.status import ServerErrorStatus + from httoop.status import ServerErrorStatus # noqa: PLC0415 with pytest.raises(RuntimeError): diff --git a/tests/api/test_uri.py b/tests/api/test_uri.py index c859b6a2..25740916 100644 --- a/tests/api/test_uri.py +++ b/tests/api/test_uri.py @@ -1,19 +1,20 @@ import pytest +from httoop import Request from httoop.exceptions import InvalidURI -def test_uri_set_string(request_): +def test_uri_set_string(request_: Request): request_.uri = '/foo' assert request_.uri == '/foo' -def test_uri_set_bytes(request_): +def test_uri_set_bytes(request_: Request): request_.uri = b'/foo' assert request_.uri == b'/foo' -def test_uri_set_dict(request_): +def test_uri_set_dict(request_: Request): uri = { 'scheme': 'http', 'username': 'username', @@ -30,7 +31,7 @@ def test_uri_set_dict(request_): assert bytes(request_.uri) == b'http://username:password@host:8090/path?query=string#fragment' -def test_set_invalid_uri_nonascii(request_): +def test_set_invalid_uri_nonascii(request_: Request): with pytest.raises(InvalidURI): request_.uri = '/fooäbar' with pytest.raises(InvalidURI): @@ -39,20 +40,20 @@ def test_set_invalid_uri_nonascii(request_): request_.uri = '/fooäbar'.encode() -def test_set_invalid_uri(request_): +def test_set_invalid_uri(request_: Request): with pytest.raises(TypeError): request_.uri = 1 with pytest.raises(TypeError): request_.uri.path = 1 -def test_set_latin1_bytes_uri_path(request_): # just for code coverage... behvaior is stupid +def test_set_latin1_bytes_uri_path(request_: Request): # just for code coverage... behvaior is stupid request_.uri.path = b'/foo\xffbar' assert bytes(request_.uri) == b'/foo%C3%BFbar' @pytest.mark.xfail -def test_uri_path_segments(request_): +def test_uri_path_segments(request_: Request): request_.uri.parse(b'/fo%2fbar/baz%2Fblub') assert request_.uri.path_segments == ['', 'fo/bar', 'baz/blub'] request_.uri.path_segments = ['', 'my/path', 'segments'] diff --git a/tests/api/test_wsgi.py b/tests/api/test_wsgi.py index e0df8723..305970b0 100644 --- a/tests/api/test_wsgi.py +++ b/tests/api/test_wsgi.py @@ -149,7 +149,7 @@ def application14(environ, start_response): def test_eror_reraising(): client = WSGIClient() - with pytest.raises(ValueError): + with pytest.raises(ValueError, match='test'): client(application14) @@ -165,13 +165,14 @@ def application15(environ, start_response): def test_essential_parameters(): client = WSGIClient({'CONTENT_TYPE': 'text/html', 'CONTENT_LENGTH': '0', 'HTTP_HOST': 'foobar'}) client(application15) + assert client.exc_info is not None assert client.exc_info[1].args[0] is True def test_client_reraising(): client = WSGIClient({'CONTENT_TYPE': 'text/html', 'CONTENT_LENGTH': '0', 'HTTP_HOST': 'foobar'}) client.headers_sent = True - with pytest.raises(ValueError) as exc: + with pytest.raises(ValueError, match='True') as exc: client(application15) assert exc.value.args[0] is True diff --git a/tests/authentication/test_auth.py b/tests/authentication/test_auth.py index f03b5174..eada592d 100644 --- a/tests/authentication/test_auth.py +++ b/tests/authentication/test_auth.py @@ -48,7 +48,8 @@ def test_order(headers): def test_no_realm(headers): headers.parse(b'WWW-Authenticate: basic foo="bar"') - assert headers.get_element('WWW-Authenticate').realm == '' + assert not headers.get_element('WWW-Authenticate').realm + assert isinstance(headers.get_element('WWW-Authenticate').realm, str) def test_multiple_headers(headers): diff --git a/tests/authentication/test_digest.py b/tests/authentication/test_digest.py index 0dabb318..e1f8577d 100644 --- a/tests/authentication/test_digest.py +++ b/tests/authentication/test_digest.py @@ -85,7 +85,7 @@ def test_digest_authorization(headers): assert elem.username == 'Mufasa' assert elem.password is None elem.username = 'test' - elem.password = 'test' # noqa: S105 + elem.password = 'test' assert elem.username == 'test' assert elem.password is None diff --git a/tests/conftest.py b/tests/conftest.py index adb4c795..065422cb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,37 +4,37 @@ @pytest.fixture -def request_(): +def request_() -> Request: return Request() @pytest.fixture -def response(): +def response() -> Response: return Response() @pytest.fixture -def headers(): +def headers() -> Headers: return Headers() @pytest.fixture -def body(): +def body() -> Body: return Body() @pytest.fixture -def statemachine(): +def statemachine() -> ServerStateMachine: return ServerStateMachine('http', 'localhost', 8090) @pytest.fixture -def clientstatemachine(): +def clientstatemachine() -> ClientStateMachine: c = ClientStateMachine() c.request = Request() return c @pytest.fixture -def uri(): +def uri() -> URI: return URI() diff --git a/tests/headers/test_accept_headers.py b/tests/headers/test_accept_headers.py index bf0bfc08..0996b939 100644 --- a/tests/headers/test_accept_headers.py +++ b/tests/headers/test_accept_headers.py @@ -8,8 +8,10 @@ def test_quality_parameter_in_accept_header(headers): def test_comparing_accept(headers): headers.parse(b'Accept: application/json; q=0.2, text/plain, text/html; q=0.5, *; q=0') - headers.elements('Accept')[0] < 'text/html' - headers.elements('Accept')[0] < '*' + assert headers.elements('Accept')[2] < 'text/html' + assert headers.elements('Accept')[2] < '*' + # FIXME: assert headers.elements('Accept')[1] <= 'text/html' + assert headers.elements('Accept')[1] < 'text/html' @pytest.mark.parametrize('header_name', ['Accept', 'Content-Type']) diff --git a/tests/headers/test_cookies.py b/tests/headers/test_cookies.py index c61f10ba..5550ad88 100644 --- a/tests/headers/test_cookies.py +++ b/tests/headers/test_cookies.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import datetime, timezone import pytest @@ -23,7 +23,7 @@ def test_cookie(cookie, values, headers): (('PREF=ID=1111111111111111:FF=0:TM=1442750337:LM=1442750337:V=1:S=OzLcd0aN7JCguSAs; expires=Thu, 31-Dec-2015 16:02:17 GMT; path=/; domain=.google.de, NID=71=qx1aDrIv1ZCfe9nzprBX_6_GMe5jmnD2RniOFz5UINXwR_3TQU0Kon20XczY4aUNlt75z_2r1wHOJw4FKL9RMUCo5QIEbmKGw3W4U7nkpZZolbPBCGbw6RN2N0p7D3q6fhXQbg; expires=Mon, 21-Mar-2016 11:58:57 GMT; path=/; domain=.google.de; HttpOnly', { 'PREF': { 'cookie_value': 'ID=1111111111111111:FF=0:TM=1442750337:LM=1442750337:V=1:S=OzLcd0aN7JCguSAs', - 'expires': datetime(2015, 12, 31, 16, 2, 17), + 'expires': datetime(2015, 12, 31, 16, 2, 17, tzinfo=timezone.utc), 'path': '/', 'domain': '.google.de', 'httponly': False, @@ -33,7 +33,7 @@ def test_cookie(cookie, values, headers): }, 'NID': { 'cookie_value': '71=qx1aDrIv1ZCfe9nzprBX_6_GMe5jmnD2RniOFz5UINXwR_3TQU0Kon20XczY4aUNlt75z_2r1wHOJw4FKL9RMUCo5QIEbmKGw3W4U7nkpZZolbPBCGbw6RN2N0p7D3q6fhXQbg', - 'expires': datetime(2016, 3, 21, 11, 58, 57), + 'expires': datetime(2016, 3, 21, 11, 58, 57, tzinfo=timezone.utc), 'path': '/', 'domain': '.google.de', 'httponly': True, @@ -56,13 +56,13 @@ def test_set_cookie(cookie, values, headers): def test_invalid_expires(headers): headers['Set-Cookie'] = 'foo=bar; expires=xyz' with pytest.raises(InvalidHeader): - headers.get_element('Set-Cookie').expires + _ = headers.get_element('Set-Cookie').expires def test_invalid_max_age(headers): headers['Set-Cookie'] = 'foo=bar; max-age=1.1' with pytest.raises(InvalidHeader): - headers.get_element('Set-Cookie').max_age + _ = headers.get_element('Set-Cookie').max_age def test_persistent(headers): diff --git a/tests/headers/test_host_header.py b/tests/headers/test_host_header.py index 906799b6..0eda230b 100644 --- a/tests/headers/test_host_header.py +++ b/tests/headers/test_host_header.py @@ -55,13 +55,12 @@ def _test_iter(header, host, port, headers): return element -@pytest.mark.parametrize('invalid', list( - list((set('\x7F()<>@,;:/\\[\\]={} \t\\\\^"\'') | set(map(chr, range(0x1F)))) - set(';\x00')) + [ - ', ', - pytest.param(';', marks=pytest.mark.xfail), # FIXME: fails - pytest.param('\x00', marks=pytest.mark.xfail), # FIXME: fails - ] -)) +@pytest.mark.parametrize('invalid', [ + *((set('\x7F()<>@,;:/\\[\\]={} \t\\\\^"\'') | set(map(chr, range(0x1F)))) - {';', '\x00'}), + ', ', + pytest.param(';', marks=pytest.mark.xfail(reason='FIXME: fails')), + pytest.param('\x00', marks=pytest.mark.xfail(reason='FIXME: fails')), +]) def test_invalid_host_header(invalid, headers): dict.__setitem__(headers, 'Host', f'foo{invalid}bar'.encode('ISO8859-1')) with pytest.raises(InvalidHeader): diff --git a/tests/headers/test_invalid_headers.py b/tests/headers/test_invalid_headers.py index 8069d444..fdd26ff3 100644 --- a/tests/headers/test_invalid_headers.py +++ b/tests/headers/test_invalid_headers.py @@ -9,8 +9,8 @@ @pytest.mark.parametrize('invalid', iter(INVALID_HEADER_FIELD_NAMES + LATIN_CHARS)) def test_parse_invalid_characters(invalid, request_): + invalid = b'foo%sbaz: blub' % (bytes((invalid,)),) with pytest.raises(InvalidHeader): - invalid = b'foo%sbaz: blub' % (bytes((invalid,)),) request_.headers.parse(invalid) diff --git a/tests/main.py b/tests/main.py index 4746e82f..b9541ed6 100755 --- a/tests/main.py +++ b/tests/main.py @@ -18,9 +18,8 @@ def main(): cmd = ['py.test', '-r', 'fsxX', '--durations=1', '--ignore=tmp', '--color=yes', '--continue-on-collection-errors'] if importable('pytest_cov'): - cmd.append('--cov=httoop') - # cmd.append("--no-cov-on-fail") - cmd.append('--cov-report=html') + cmd.extend(('--cov=httoop', '--cov-report=html')) + # cmd.append("--no-cov-on-fail") cmd.append(pathlib.Path(pathlib.Path(__file__).resolve()).parent) diff --git a/tests/messaging/test_body.py b/tests/messaging/test_body.py index 6bd65be4..a802227d 100644 --- a/tests/messaging/test_body.py +++ b/tests/messaging/test_body.py @@ -2,13 +2,13 @@ import pytest -from httoop import BAD_REQUEST +from httoop import BAD_REQUEST, Request from httoop.codecs import lookup from httoop.semantic.request import ComposedRequest from httoop.semantic.response import ComposedResponse -def test_chunked_body_without_trailer(request_): +def test_chunked_body_without_trailer(request_: Request): def content(): yield 'Let us test this chunked' yield '\n' @@ -32,7 +32,7 @@ def test_parse_chunked_body_without_trailer(statemachine): assert bytes(request_.body) == request_body -def test_parse_chunked_body_without_trailer_2(request_): +def test_parse_chunked_body_without_trailer_2(request_: Request): request_.body = [ 'This is a chunked body with some lines', 'foo', 'bar', 'Baz', '\n', '', 'blah!', 'blub' @@ -54,7 +54,7 @@ def test_parse_transfer_encoding_deflate(statemachine): assert request.body == 'this is a test' -def test_body_parse_transfer_encoding_deflate(request_): +def test_body_parse_transfer_encoding_deflate(request_: Request): request_.body.transfer_encoding = 'deflate' request_.body.parse(b'x\x9c+\xc9\xc8,V\x00\xa2D\x85\x92\xd4\xe2\x12\x00&3\x05\x16') assert request_.body == 'this is a test' @@ -79,7 +79,7 @@ def test_parse_chunked_body_with_untold_trailer(statemachine): assert 'untold trailers: "Bar"' in str(exc.value) -def test_chunked_body_with_trailer(request_): +def test_chunked_body_with_trailer(request_: Request): def content(): yield 'Let us test this chunked' yield '\n' @@ -155,13 +155,13 @@ def test_parse_chunked_body_with_invalid_terminator(statemachine): assert 'Invalid chunk terminator' in str(exc.value.description) -def test_body_compress(request_): +def test_body_compress(request_: Request): request_.body = 'this is a test' request_.body.content_encoding = 'deflate' request_.body.compress() assert lookup('application/zlib').decode(bytes(request_.body)) == 'this is a test' -def test_body_encode_none(request_): +def test_body_encode_none(request_: Request): request_.body.mimetype = 'application/json' request_.body.encode() diff --git a/tests/messaging/test_request_codecs.py b/tests/messaging/test_request_codecs.py index 5d76d488..e0767bce 100644 --- a/tests/messaging/test_request_codecs.py +++ b/tests/messaging/test_request_codecs.py @@ -16,7 +16,7 @@ def test_json(body): def test_invalid_json(body): body.mimetype = 'application/json' - with pytest.raises(ValueError): + with pytest.raises(ValueError, match=r"Expecting ':' delimiter: line 1 column 8 \(char 7\)"): body.decode('{"foo" "bar"}') @@ -38,7 +38,7 @@ def test_percent_encoding(body): ) -def test_parse_multipart_form_data(body): +def test_parse_multipart_form_data(body: Body): body.mimetype = 'multipart/form-data' content_type = body.mimetype content_type.boundary = '------------------------409e1d7f8fe3763a' @@ -58,7 +58,7 @@ def test_parse_multipart_form_data(body): assert bar.headers.element('Content-Disposition').params['filename'] == 'test.txt' -def test_compose_multipart_form_data(body): +def test_compose_multipart_form_data(body: Body): body.mimetype = 'multipart/form-data' content_type = body.mimetype content_type.boundary = '------------------------409e1d7f8fe3763a' @@ -159,7 +159,7 @@ def check_encoding_dict(body, data): def check_raises(body, chars, type_, exception): for chr_ in chars: - with pytest.raises(exception): + with pytest.raises(exception): # noqa: PT012 body.set(chr_) type_(body) diff --git a/tests/messaging/test_semantic.py b/tests/messaging/test_semantic.py index a8737933..33936712 100644 --- a/tests/messaging/test_semantic.py +++ b/tests/messaging/test_semantic.py @@ -1,21 +1,22 @@ import pytest +from httoop import Request, Response from httoop.semantic.request import ComposedRequest from httoop.semantic.response import ComposedResponse -def test_composing(request_, response): +def test_composing(request_: Request, response: Response): c = ComposedResponse(response, request_) c.prepare() -def test_request_close(request_): +def test_request_close(request_: Request): c = ComposedRequest(request_) c.close = True assert request_.headers['Connection'] == 'close' -def test_request_trace_header_removal(request_): +def test_request_trace_header_removal(request_: Request): request_.headers['Cookie'] = 'foo=bar' request_.headers['WWW-Authenticate'] = 'foo bar' request_.method = 'TRACE' @@ -25,14 +26,14 @@ def test_request_trace_header_removal(request_): assert 'WWW-Authenticate' not in request_.headers -def test_request_host_header_set(request_): +def test_request_host_header_set(request_: Request): request_.uri = 'http://www.example.com/foo' c = ComposedRequest(request_) c.prepare() assert request_.headers['Host'] == 'www.example.com' -def test_request_defaults(request_): +def test_request_defaults(request_: Request): request_.body = 'foo' c = ComposedRequest(request_) c.chunked = True @@ -53,7 +54,7 @@ def test_request_defaults_body(request_, method): assert request_.headers.get('Content-Length') == '3' -def test_chunked(request_): +def test_chunked(request_: Request): c = ComposedRequest(request_) c.transfer_encoding = None c.transfer_encoding = b'chunked' @@ -72,7 +73,7 @@ def test_response_no_body(request_, response, status): assert not response.body -def test_response_no_body_head(request_, response): +def test_response_no_body_head(request_: Request, response: Response): response.body = 'foo' request_.method = 'HEAD' c = ComposedResponse(response, request_) @@ -80,14 +81,14 @@ def test_response_no_body_head(request_, response): assert not response.body -def test_response_allow_default_header(request_, response): +def test_response_allow_default_header(request_: Request, response: Response): response.status = 405 c = ComposedResponse(response, request_) c.prepare() assert response.headers['Allow'] == 'GET, HEAD' -def test_byte_ranges(request_, response): +def test_byte_ranges(request_: Request, response: Response): request_.headers['Range'] = 'bytes=3-5' response.headers['ETag'] = 'foo' response.body = 'foobarbaz' @@ -98,7 +99,7 @@ def test_byte_ranges(request_, response): assert response.headers['Content-Range'] == 'bytes 3-5/9' -def test_invalid_byte_ranges(request_, response): +def test_invalid_byte_ranges(request_: Request, response: Response): request_.headers['Range'] = 'bytes=5-3' response.headers['ETag'] = 'foo' response.body = 'foobarbaz' @@ -110,7 +111,7 @@ def test_invalid_byte_ranges(request_, response): assert response.headers['Content-Range'] == 'bytes */9' -def test_multipart_byte_ranges(request_, response): +def test_multipart_byte_ranges(request_: Request, response: Response): request_.headers['Range'] = 'bytes=0-2, 6-' response.headers['ETag'] = 'foo' response.body = 'foobarbaz' @@ -121,7 +122,7 @@ def test_multipart_byte_ranges(request_, response): assert bytes(response.body) == b'--%(boundary)s\r\nContent-Range: bytes 0-2/9\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\nfoo\r\n--%(boundary)s\r\nContent-Range: bytes 6-9/9\r\nContent-Type: text/plain; charset=UTF-8\r\n\r\nbaz\r\n--%(boundary)s--\r\n' % {b'boundary': response.headers.get_element('Content-Type').boundary.encode('ASCII')} -def test_content_range_is_set(request_, response): +def test_content_range_is_set(request_: Request, response: Response): response.status = 416 response.body = 'foobarbaz' c = ComposedResponse(response, request_) @@ -129,7 +130,7 @@ def test_content_range_is_set(request_, response): assert response.headers['Content-Range'] == 'bytes */9' -def test_response_trace_header_removal(request_, response): +def test_response_trace_header_removal(request_: Request, response: Response): response.headers['Set-Cookie'] = 'foo=bar' request_.method = 'TRACE' c = ComposedResponse(response, request_) @@ -137,7 +138,7 @@ def test_response_trace_header_removal(request_, response): assert 'Set-Cookie' not in response.headers -def test_response_close(request_, response): +def test_response_close(request_: Request, response: Response): c = ComposedResponse(response, request_) c.close = True assert response.headers['Connection'] == 'close' diff --git a/tests/tc2231/test_tc2231.py b/tests/tc2231/test_tc2231.py index 1b402e7e..9c79f77c 100644 --- a/tests/tc2231/test_tc2231.py +++ b/tests/tc2231/test_tc2231.py @@ -355,13 +355,13 @@ def test_attabspathwin(content_disposition): def test_attcdate(content_disposition): h = content_disposition(b'Content-Disposition: attachment; creation-date="Wed, 12 Feb 1997 16:29:51 -0500"') - assert h.creation_date == datetime.datetime(1997, 2, 12, 16, 29, 51) + assert h.creation_date == datetime.datetime(1997, 2, 12, 16, 29, 51, tzinfo=datetime.timezone.utc) # assert h.creation_date == 855761391.0 # FIXME: assert == 855761391.0 def test_attmdate(content_disposition): h = content_disposition(b'Content-Disposition: attachment; modification-date="Wed, 12 Feb 1997 16:29:51 -0500"') - assert h.modification_date == datetime.datetime(1997, 2, 12, 16, 29, 51) + assert h.modification_date == datetime.datetime(1997, 2, 12, 16, 29, 51, tzinfo=datetime.timezone.utc) # assert h.modification_date == 855761391.0 # FIXME: assert == 855761391.0 diff --git a/tests/uri/test_uri_query.py b/tests/uri/test_uri_query.py index cdb6264f..8514313e 100644 --- a/tests/uri/test_uri_query.py +++ b/tests/uri/test_uri_query.py @@ -7,8 +7,8 @@ (b'', ()), (b'&', ()), (b'&&', ()), - pytest.param(b'=', (('', ''),), marks=pytest.mark.skipif(True, reason='Dunno')), - pytest.param(b'=a', (('', 'a'),), marks=pytest.mark.skipif(True, reason='Dunno')), + pytest.param(b'=', (('', ''),), marks=pytest.mark.skip(reason='Dunno')), + pytest.param(b'=a', (('', 'a'),), marks=pytest.mark.skip(reason='Dunno')), (b'a', (('a', ''),)), (b'a=', (('a', ''),)), (b'&a=b', (('a', 'b'),)), @@ -26,10 +26,10 @@ def test_query_string_parse(query_string, query): @pytest.mark.parametrize('query_string,query', [ (b'', ()), - pytest.param(b'=', (('', ''),), marks=pytest.mark.skipif(True, reason='Dunno')), - pytest.param(b'=a', (('', 'a'),), marks=pytest.mark.skipif(True, reason='Dunno')), + pytest.param(b'=', (('', ''),), marks=pytest.mark.skip(reason='Dunno')), + pytest.param(b'=a', (('', 'a'),), marks=pytest.mark.skip(reason='Dunno')), (b'a', (('a', ''),)), - pytest.param(b'a=', (('a', ''),), marks=pytest.mark.skipif(True, reason='Dunno')), + pytest.param(b'a=', (('a', ''),), marks=pytest.mark.skip(reason='Dunno')), (b'a=b', (('a', 'b'),)), (b'a=b&b=c&d=f', (('a', 'b'), ('b', 'c'), ('d', 'f'))), (b'a=a+b&b=b+c', (('a', 'a b'), ('b', 'b c'))), @@ -62,7 +62,7 @@ def test_parse_encodings(query_string, encoding, query): def test_urlencode_sequences(): u = URI() u.query = {'a': [1, 2], 'b': (3, 4, 5)} - assert set(u.query_string.split(b'&')) == {b'a=1', b'a=2', b'b=3', b'b=4', b'b=5'} + assert set(u.query_string.split('&')) == {b'a=1', b'a=2', b'b=3', b'b=4', b'b=5'} @pytest.mark.xfail(reason='API not yet implemented.') diff --git a/tests/uri/test_uri_schemes.py b/tests/uri/test_uri_schemes.py index 9ae85ab9..e167ddb7 100644 --- a/tests/uri/test_uri_schemes.py +++ b/tests/uri/test_uri_schemes.py @@ -6,7 +6,7 @@ @pytest.mark.parametrize('url,expected', [ ('ftp://ftp.is.co.za/rfc/rfc1808.txt', ('ftp', '', '', 'ftp.is.co.za', 21, '/rfc/rfc1808.txt', '', '')), ('http://www.ietf.org/rfc/rfc2396.txt', ('http', '', '', 'www.ietf.org', 80, '/rfc/rfc2396.txt', '', '')), - pytest.param('ldap://[2001:db8::7]/c=GB?objectClass?one', ('ldap', '', '', '[2001:db8::7]', 389, '/c=GB', 'objectClass?one', ''), marks=pytest.mark.skipif(True, reason='Parse query in ldap URI?')), + pytest.param('ldap://[2001:db8::7]/c=GB?objectClass?one', ('ldap', '', '', '[2001:db8::7]', 389, '/c=GB', 'objectClass?one', ''), marks=pytest.mark.skip(reason='Parse query in ldap URI?')), ('mailto:John.Doe@example.com', ('mailto', '', '', '', None, 'John.Doe@example.com', '', '')), ('news:comp.infosystems.www.servers.unix', ('news', '', '', '', None, 'comp.infosystems.www.servers.unix', '', '')), ('tel:+1-816-555-1212', ('tel', '', '', '', None, '+1-816-555-1212', '', '')),