from mitmproxy import stateobject
from mitmproxy.coretypes import serializable
from wsproto.frame_protocol import Opcode

WebSocketMessageState = Tuple[int, bool, bytes, float, bool]


class WebSocketMessage(serializable.Serializable):

            

Reported by Pylint.

              
    def __init__(
        self,
        type: Union[int, Opcode],
        from_client: bool,
        content: bytes,
        timestamp: Optional[float] = None,
        killed: bool = False,
    ) -> None:

            

Reported by Pylint.

              """
Mitmproxy used to have its own WebSocketFlow type until mitmproxy 6, but now WebSocket connections now are represented
as HTTP flows as well. They can be distinguished from regular HTTP requests by having the
`mitmproxy.http.HTTPFlow.websocket` attribute set.

This module only defines the classes for individual `WebSocketMessage`s and the `WebSocketData` container.
"""
import time
import warnings

            

Reported by Pylint.

              as HTTP flows as well. They can be distinguished from regular HTTP requests by having the
`mitmproxy.http.HTTPFlow.websocket` attribute set.

This module only defines the classes for individual `WebSocketMessage`s and the `WebSocketData` container.
"""
import time
import warnings
from typing import List, Tuple, Union
from typing import Optional

            

Reported by Pylint.

              
from mitmproxy import stateobject
from mitmproxy.coretypes import serializable
from wsproto.frame_protocol import Opcode

WebSocketMessageState = Tuple[int, bool, bytes, float, bool]


class WebSocketMessage(serializable.Serializable):

            

Reported by Pylint.

              
    The [WebSocket RFC](https://tools.ietf.org/html/rfc6455) specifies both
    text and binary messages. To avoid a whole class of nasty type confusion bugs,
    mitmproxy stores all message contents as `bytes`. If you need a `str`, you can access the `text` property
    on text messages:

    >>> if message.is_text:
    >>>     text = message.text
    """

            

Reported by Pylint.

                  dropped: bool
    """True if the message has not been forwarded by mitmproxy, False otherwise."""

    def __init__(
        self,
        type: Union[int, Opcode],
        from_client: bool,
        content: bytes,
        timestamp: Optional[float] = None,

            

Reported by Pylint.

                      self.type = Opcode(typ)

    def __repr__(self):
        if self.type == Opcode.TEXT:
            return repr(self.content.decode(errors="replace"))
        else:
            return repr(self.content)

    @property

            

Reported by Pylint.

              
    def kill(self):  # pragma: no cover
        """A deprecated alias for `.drop()`."""
        warnings.warn("WebSocketMessage.kill() is deprecated, use .drop() instead.", DeprecationWarning, stacklevel=2)
        self.drop()

    @property
    def text(self) -> str:
        """

            

Reported by Pylint.

                      *See also:* `WebSocketMessage.content`
        """
        if self.type != Opcode.TEXT:
            raise AttributeError(f"{self.type.name.title()} WebSocket frames do not have a 'text' attribute.")

        return self.content.decode()

    @text.setter
    def text(self, value: str) -> None:

            

Reported by Pylint.

              """Handle file paths as command arguments."""
import typing

from mitmproxy import command
from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types


            

Reported by Pylint.

              import typing

from mitmproxy import command
from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types



            

Reported by Pylint.

              
from mitmproxy import command
from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types


class MyAddon:

            

Reported by Pylint.

              from mitmproxy import command
from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types


class MyAddon:
    @command.command("myaddon.histogram")

            

Reported by Pylint.

              from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types


class MyAddon:
    @command.command("myaddon.histogram")
    def histogram(

            

Reported by Pylint.

              """Handle file paths as command arguments."""
import typing

from mitmproxy import command
from mitmproxy import ctx
from mitmproxy import flow
from mitmproxy import http
from mitmproxy import types


            

Reported by Pylint.

              from mitmproxy import types


class MyAddon:
    @command.command("myaddon.histogram")
    def histogram(
        self,
        flows: typing.Sequence[flow.Flow],
        path: types.Path,

            

Reported by Pylint.

              from mitmproxy import types


class MyAddon:
    @command.command("myaddon.histogram")
    def histogram(
        self,
        flows: typing.Sequence[flow.Flow],
        path: types.Path,

            

Reported by Pylint.

              
class MyAddon:
    @command.command("myaddon.histogram")
    def histogram(
        self,
        flows: typing.Sequence[flow.Flow],
        path: types.Path,
    ) -> None:
        totals: typing.Dict[str, int] = {}

            

Reported by Pylint.

              
class MyAddon:
    @command.command("myaddon.histogram")
    def histogram(
        self,
        flows: typing.Sequence[flow.Flow],
        path: types.Path,
    ) -> None:
        totals: typing.Dict[str, int] = {}

            

Reported by Pylint.

              @lru_cache(1)
def parse_json(s: bytes) -> typing.Any:
    try:
        return json.loads(s.decode('utf-8'))
    except ValueError:
        return PARSE_ERROR


def format_json(data: typing.Any) -> typing.Iterator[base.TViewLine]:

            

Reported by Pylint.

              

def format_json(data: typing.Any) -> typing.Iterator[base.TViewLine]:
    encoder = json.JSONEncoder(indent=4, sort_keys=True, ensure_ascii=False)
    current_line: base.TViewLine = []
    for chunk in encoder.iterencode(data):
        if "\n" in chunk:
            rest_of_last_line, chunk = chunk.split("\n", maxsplit=1)
            # rest_of_last_line is a delimiter such as , or [

            

Reported by Pylint.

              import re
import json
from functools import lru_cache

import typing

from mitmproxy.contentviews import base

PARSE_ERROR = object()

            

Reported by Pylint.

                      if data is not PARSE_ERROR:
            return "JSON", format_json(data)

    def render_priority(self, data: bytes, *, content_type: typing.Optional[str] = None, **metadata) -> float:
        if content_type in (
            "application/json",
            "application/json-rpc",
        ):
            return 1

            

Reported by Pylint.

              import re
import json
from functools import lru_cache

import typing

from mitmproxy.contentviews import base

PARSE_ERROR = object()

            

Reported by Pylint.

              

@lru_cache(1)
def parse_json(s: bytes) -> typing.Any:
    try:
        return json.loads(s.decode('utf-8'))
    except ValueError:
        return PARSE_ERROR


            

Reported by Pylint.

              

@lru_cache(1)
def parse_json(s: bytes) -> typing.Any:
    try:
        return json.loads(s.decode('utf-8'))
    except ValueError:
        return PARSE_ERROR


            

Reported by Pylint.

                      return PARSE_ERROR


def format_json(data: typing.Any) -> typing.Iterator[base.TViewLine]:
    encoder = json.JSONEncoder(indent=4, sort_keys=True, ensure_ascii=False)
    current_line: base.TViewLine = []
    for chunk in encoder.iterencode(data):
        if "\n" in chunk:
            rest_of_last_line, chunk = chunk.split("\n", maxsplit=1)

            

Reported by Pylint.

                  yield current_line


class ViewJSON(base.View):
    name = "JSON"

    def __call__(self, data, **metadata):
        data = parse_json(data)
        if data is not PARSE_ERROR:

            

Reported by Pylint.

              class ViewJSON(base.View):
    name = "JSON"

    def __call__(self, data, **metadata):
        data = parse_json(data)
        if data is not PARSE_ERROR:
            return "JSON", format_json(data)

    def render_priority(self, data: bytes, *, content_type: typing.Optional[str] = None, **metadata) -> float:

            

Reported by Pylint.

              import io

import pytest

from mitmproxy.test import taddons
from mitmproxy import ctx


@pytest.mark.asyncio

            

Reported by Pylint.

              import io

import pytest

from mitmproxy.test import taddons
from mitmproxy import ctx


@pytest.mark.asyncio

            

Reported by Pylint.

              

@pytest.mark.asyncio
async def test_recordingmaster():
    with taddons.context() as tctx:
        assert not tctx.master.has_log("nonexistent")
        ctx.log.error("foo")
        assert not tctx.master.has_log("foo", level="debug")
        await tctx.master.await_log("foo", level="error")

            

Reported by Pylint.

              @pytest.mark.asyncio
async def test_recordingmaster():
    with taddons.context() as tctx:
        assert not tctx.master.has_log("nonexistent")
        ctx.log.error("foo")
        assert not tctx.master.has_log("foo", level="debug")
        await tctx.master.await_log("foo", level="error")



            

Reported by Bandit.

                  with taddons.context() as tctx:
        assert not tctx.master.has_log("nonexistent")
        ctx.log.error("foo")
        assert not tctx.master.has_log("foo", level="debug")
        await tctx.master.await_log("foo", level="error")


@pytest.mark.asyncio
async def test_dumplog():

            

Reported by Bandit.

              

@pytest.mark.asyncio
async def test_dumplog():
    with taddons.context() as tctx:
        ctx.log.info("testing")
        await tctx.master.await_log("testing")
        s = io.StringIO()
        tctx.master.dump_log(s)

            

Reported by Pylint.

                  with taddons.context() as tctx:
        ctx.log.info("testing")
        await tctx.master.await_log("testing")
        s = io.StringIO()
        tctx.master.dump_log(s)
        assert s.getvalue()


def test_load_script(tdata):

            

Reported by Pylint.

                      await tctx.master.await_log("testing")
        s = io.StringIO()
        tctx.master.dump_log(s)
        assert s.getvalue()


def test_load_script(tdata):
    with taddons.context() as tctx:
        s = tctx.script(

            

Reported by Bandit.

                      assert s.getvalue()


def test_load_script(tdata):
    with taddons.context() as tctx:
        s = tctx.script(
            tdata.path(
                "mitmproxy/data/addonscripts/recorder/recorder.py"
            )

            

Reported by Pylint.

              
def test_load_script(tdata):
    with taddons.context() as tctx:
        s = tctx.script(
            tdata.path(
                "mitmproxy/data/addonscripts/recorder/recorder.py"
            )
        )
        assert s

            

Reported by Pylint.

              import gc
import threading
from pympler import muppy, refbrowser
from OpenSSL import SSL
# import os
# os.environ["TK_LIBRARY"] = r"C:\Python27\tcl\tcl8.5"
# os.environ["TCL_LIBRARY"] = r"C:\Python27\tcl\tcl8.5"

# Also noteworthy: guppy, objgraph

            

Reported by Pylint.

                  return str(id(obj)) + ": " + str(obj)[:100].replace("\r\n", "\\r\\n").replace("\n", "\\n")


def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")


            

Reported by Pylint.

                  return str(id(obj)) + ": " + str(obj)[:100].replace("\r\n", "\\r\\n").replace("\n", "\\n")


def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")


            

Reported by Pylint.

              

def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")

    step += 1

            

Reported by Pylint.

              import gc
import threading
from pympler import muppy, refbrowser
from OpenSSL import SSL
# import os
# os.environ["TK_LIBRARY"] = r"C:\Python27\tcl\tcl8.5"
# os.environ["TCL_LIBRARY"] = r"C:\Python27\tcl\tcl8.5"

# Also noteworthy: guppy, objgraph

            

Reported by Pylint.

              
# Also noteworthy: guppy, objgraph

step = 0
__memory_locals__ = True


def str_fun(obj):
    if isinstance(obj, dict):

            

Reported by Pylint.

              __memory_locals__ = True


def str_fun(obj):
    if isinstance(obj, dict):
        if "__memory_locals__" in obj:
            return "(-locals-)"
        if "self" in obj and isinstance(obj["self"], refbrowser.InteractiveBrowser):
            return "(-browser-)"

            

Reported by Pylint.

                  return str(id(obj)) + ": " + str(obj)[:100].replace("\r\n", "\\r\\n").replace("\n", "\\n")


def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")


            

Reported by Pylint.

              

def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")

    step += 1

            

Reported by Pylint.

              

def request(ctx, flow):
    global step, ssl
    print("==========")
    print(f"GC: {gc.collect()}")
    print(f"Threads: {threading.active_count()}")

    step += 1

            

Reported by Pylint.

              from mitmproxy.test import tflow

from mitmproxy.addons import anticache
from mitmproxy.test import taddons


class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"

            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"

            

Reported by Pylint.

              

class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"
            f.request.headers["if-none-match"] = "test"

            

Reported by Pylint.

              

class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"
            f.request.headers["if-none-match"] = "test"

            

Reported by Pylint.

              
class TestAntiCache:
    def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"
            f.request.headers["if-none-match"] = "test"


            

Reported by Pylint.

                  def test_simple(self):
        sa = anticache.AntiCache()
        with taddons.context(sa) as tctx:
            f = tflow.tflow(resp=True)
            f.request.headers["if-modified-since"] = "test"
            f.request.headers["if-none-match"] = "test"

            sa.request(f)
            assert "if-modified-since" in f.request.headers

            

Reported by Pylint.

                          f.request.headers["if-none-match"] = "test"

            sa.request(f)
            assert "if-modified-since" in f.request.headers
            assert "if-none-match" in f.request.headers

            tctx.configure(sa, anticache = True)
            sa.request(f)
            assert "if-modified-since" not in f.request.headers

            

Reported by Bandit.

              
            sa.request(f)
            assert "if-modified-since" in f.request.headers
            assert "if-none-match" in f.request.headers

            tctx.configure(sa, anticache = True)
            sa.request(f)
            assert "if-modified-since" not in f.request.headers
            assert "if-none-match" not in f.request.headers

            

Reported by Bandit.

              
            tctx.configure(sa, anticache = True)
            sa.request(f)
            assert "if-modified-since" not in f.request.headers
            assert "if-none-match" not in f.request.headers

            

Reported by Bandit.

              import asyncio
from mitmproxy import ctx


class KeepServing:
    def load(self, loader):
        loader.add_option(
            "keepserving", bool, False,
            """

            

Reported by Pylint.

              from mitmproxy import ctx


class KeepServing:
    def load(self, loader):
        loader.add_option(
            "keepserving", bool, False,
            """
            Continue serving after client playback, server playback or file

            

Reported by Pylint.

              

class KeepServing:
    def load(self, loader):
        loader.add_option(
            "keepserving", bool, False,
            """
            Continue serving after client playback, server playback or file
            read. This option is ignored by interactive tools, which always keep

            

Reported by Pylint.

              

class KeepServing:
    def load(self, loader):
        loader.add_option(
            "keepserving", bool, False,
            """
            Continue serving after client playback, server playback or file
            read. This option is ignored by interactive tools, which always keep

            

Reported by Pylint.

                          """
        )

    def keepgoing(self) -> bool:
        checks = [
            "readfile.reading",
            "replay.client.count",
            "replay.server.count",
        ]

            

Reported by Pylint.

                          """
        )

    def keepgoing(self) -> bool:
        checks = [
            "readfile.reading",
            "replay.client.count",
            "replay.server.count",
        ]

            

Reported by Pylint.

                          "replay.client.count",
            "replay.server.count",
        ]
        return any([ctx.master.commands.call(c) for c in checks])

    def shutdown(self):  # pragma: no cover
        ctx.master.shutdown()

    async def watch(self):

            

Reported by Pylint.

                      ]
        return any([ctx.master.commands.call(c) for c in checks])

    def shutdown(self):  # pragma: no cover
        ctx.master.shutdown()

    async def watch(self):
        while True:
            await asyncio.sleep(0.1)

            

Reported by Pylint.

                      ]
        return any([ctx.master.commands.call(c) for c in checks])

    def shutdown(self):  # pragma: no cover
        ctx.master.shutdown()

    async def watch(self):
        while True:
            await asyncio.sleep(0.1)

            

Reported by Pylint.

                  def shutdown(self):  # pragma: no cover
        ctx.master.shutdown()

    async def watch(self):
        while True:
            await asyncio.sleep(0.1)
            if not self.keepgoing():
                self.shutdown()


            

Reported by Pylint.

                  import mitmproxy.proxy.layer


class Command:
    """
    Base class for all commands
    """

    blocking: Union[bool, "mitmproxy.proxy.layer.Layer"] = False

            

Reported by Pylint.

                  """

    def __repr__(self):
        x = self.__dict__.copy()
        x.pop("blocking", None)
        return f"{type(self).__name__}({repr(x)})"


class ConnectionCommand(Command):

            

Reported by Pylint.

                      return f"{type(self).__name__}({repr(x)})"


class ConnectionCommand(Command):
    """
    Commands involving a specific connection
    """
    connection: Connection


            

Reported by Pylint.

                      self.connection = connection


class SendData(ConnectionCommand):
    """
    Send data to a remote peer
    """
    data: bytes


            

Reported by Pylint.

                      return f"SendData({target}, {self.data})"


class OpenConnection(ConnectionCommand):
    """
    Open a new connection
    """
    connection: Server
    blocking = True

            

Reported by Pylint.

                  blocking = True


class CloseConnection(ConnectionCommand):
    """
    Close a connection. If the client connection is closed,
    all other connections will ultimately be closed during cleanup.
    """
    half_close: bool

            

Reported by Pylint.

                  half_close: bool
    """
    If True, only close our half of the connection by sending a FIN packet.
    This is required from some protocols which close their end to signal completion and then continue reading,
    for example HTTP/1.0 without Content-Length header.
    """

    def __init__(self, connection: Connection, half_close: bool = False):
        super().__init__(connection)

            

Reported by Pylint.

                      return super().__new__(cls, *args, **kwargs)


class GetSocket(ConnectionCommand):
    """
    Get the underlying socket.
    This should really never be used, but is required to implement transparent mode.
    """
    blocking = True

            

Reported by Pylint.

                  blocking = True


class Log(Command):
    message: str
    level: str

    def __init__(self, message: str, level: Literal["error", "warn", "info", "alert", "debug"] = "info"):
        self.message = message

            

Reported by Pylint.

                  blocking = True


class Log(Command):
    message: str
    level: str

    def __init__(self, message: str, level: Literal["error", "warn", "info", "alert", "debug"] = "info"):
        self.message = message

            

Reported by Pylint.

              """
import copy

from .layers import test_tcp, test_tls
from .layers.http import test_http, test_http2


def test_bench_http_roundtrip(tctx, benchmark):
    # benchmark something

            

Reported by Pylint.

              import copy

from .layers import test_tcp, test_tls
from .layers.http import test_http, test_http2


def test_bench_http_roundtrip(tctx, benchmark):
    # benchmark something
    benchmark(test_http.test_http_proxy, tctx)

            

Reported by Pylint.

              from .layers.http import test_http, test_http2


def test_bench_http_roundtrip(tctx, benchmark):
    # benchmark something
    benchmark(test_http.test_http_proxy, tctx)


def test_bench_http2_roundtrip(tctx, benchmark):

            

Reported by Pylint.

                  benchmark(test_http.test_http_proxy, tctx)


def test_bench_http2_roundtrip(tctx, benchmark):
    # benchmark something
    benchmark(test_http2.test_simple, tctx)


def test_bench_tcp_roundtrip(tctx, benchmark):

            

Reported by Pylint.

                  benchmark(test_http2.test_simple, tctx)


def test_bench_tcp_roundtrip(tctx, benchmark):
    # benchmark something
    benchmark(lambda: test_tcp.test_simple(copy.deepcopy(tctx)))


def test_bench_server_tls(tctx, benchmark):

            

Reported by Pylint.

                  benchmark(lambda: test_tcp.test_simple(copy.deepcopy(tctx)))


def test_bench_server_tls(tctx, benchmark):
    t = test_tls.TestServerTLS().test_simple
    benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_client_tls(tctx, benchmark):

            

Reported by Pylint.

              

def test_bench_server_tls(tctx, benchmark):
    t = test_tls.TestServerTLS().test_simple
    benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_client_tls(tctx, benchmark):
    t = test_tls.TestClientTLS().test_client_only

            

Reported by Pylint.

                  benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_client_tls(tctx, benchmark):
    t = test_tls.TestClientTLS().test_client_only
    benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_tls_both(tctx, benchmark):

            

Reported by Pylint.

              

def test_bench_client_tls(tctx, benchmark):
    t = test_tls.TestClientTLS().test_client_only
    benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_tls_both(tctx, benchmark):
    t = test_tls.TestClientTLS().test_server_required

            

Reported by Pylint.

                  benchmark(lambda: t(copy.deepcopy(tctx)))


def test_bench_tls_both(tctx, benchmark):
    t = test_tls.TestClientTLS().test_server_required
    benchmark(lambda: t(copy.deepcopy(tctx)))

            

Reported by Pylint.

              
from mitmproxy.coretypes import multidict
from mitmproxy.net.http import multipart
from . import base


class ViewMultipart(base.View):
    name = "Multipart Form"


            

Reported by Pylint.

                      if v:
            return "Multipart form", self._format(v)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "multipart/form-data")

            

Reported by Pylint.

                      if v:
            return "Multipart form", self._format(v)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "multipart/form-data")

            

Reported by Pylint.

              from typing import Optional

from mitmproxy.coretypes import multidict
from mitmproxy.net.http import multipart
from . import base


class ViewMultipart(base.View):
    name = "Multipart Form"

            

Reported by Pylint.

              from . import base


class ViewMultipart(base.View):
    name = "Multipart Form"

    @staticmethod
    def _format(v):
        yield [("highlight", "Form data:\n")]

            

Reported by Pylint.

                  name = "Multipart Form"

    @staticmethod
    def _format(v):
        yield [("highlight", "Form data:\n")]
        yield from base.format_dict(multidict.MultiDict(v))

    def __call__(self, data: bytes, content_type: Optional[str] = None, **metadata):
        if content_type is None:

            

Reported by Pylint.

                      yield [("highlight", "Form data:\n")]
        yield from base.format_dict(multidict.MultiDict(v))

    def __call__(self, data: bytes, content_type: Optional[str] = None, **metadata):
        if content_type is None:
            return
        v = multipart.decode(content_type, data)
        if v:
            return "Multipart form", self._format(v)

            

Reported by Pylint.

                  def __call__(self, data: bytes, content_type: Optional[str] = None, **metadata):
        if content_type is None:
            return
        v = multipart.decode(content_type, data)
        if v:
            return "Multipart form", self._format(v)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "multipart/form-data")

            

Reported by Pylint.

                      if v:
            return "Multipart form", self._format(v)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "multipart/form-data")

            

Reported by Pylint.

                      if v:
            return "Multipart form", self._format(v)

    def render_priority(self, data: bytes, *, content_type: Optional[str] = None, **metadata) -> float:
        return float(content_type == "multipart/form-data")

            

Reported by Pylint.