import typing

from mitmproxy import flow, flowfilter
from mitmproxy import exceptions
from mitmproxy import ctx


class Intercept:
    filt: typing.Optional[flowfilter.TFilter] = None

            

Reported by Pylint.

              from mitmproxy import ctx


class Intercept:
    filt: typing.Optional[flowfilter.TFilter] = None

    def load(self, loader):
        loader.add_option(
            "intercept_active", bool, False,

            

Reported by Pylint.

              class Intercept:
    filt: typing.Optional[flowfilter.TFilter] = None

    def load(self, loader):
        loader.add_option(
            "intercept_active", bool, False,
            "Intercept toggle"
        )
        loader.add_option(

            

Reported by Pylint.

              class Intercept:
    filt: typing.Optional[flowfilter.TFilter] = None

    def load(self, loader):
        loader.add_option(
            "intercept_active", bool, False,
            "Intercept toggle"
        )
        loader.add_option(

            

Reported by Pylint.

                          "Intercept filter expression."
        )

    def configure(self, updated):
        if "intercept" in updated:
            if ctx.options.intercept:
                self.filt = flowfilter.parse(ctx.options.intercept)
                if not self.filt:
                    raise exceptions.OptionsError(f"Invalid interception filter: {ctx.options.intercept}")

            

Reported by Pylint.

                          if ctx.options.intercept:
                self.filt = flowfilter.parse(ctx.options.intercept)
                if not self.filt:
                    raise exceptions.OptionsError(f"Invalid interception filter: {ctx.options.intercept}")
                ctx.options.intercept_active = True
            else:
                self.filt = None
                ctx.options.intercept_active = False


            

Reported by Pylint.

                              self.filt = None
                ctx.options.intercept_active = False

    def should_intercept(self, f: flow.Flow) -> bool:
        return bool(
                ctx.options.intercept_active
                and self.filt
                and self.filt(f)
                and not f.is_replay

            

Reported by Pylint.

                              self.filt = None
                ctx.options.intercept_active = False

    def should_intercept(self, f: flow.Flow) -> bool:
        return bool(
                ctx.options.intercept_active
                and self.filt
                and self.filt(f)
                and not f.is_replay

            

Reported by Pylint.

                              and not f.is_replay
        )

    def process_flow(self, f: flow.Flow) -> None:
        if self.should_intercept(f):
            assert f.reply
            if f.reply.state != "start":
                return ctx.log.debug("Cannot intercept request that is already taken by another addon.")
            f.intercept()

            

Reported by Pylint.

                              and not f.is_replay
        )

    def process_flow(self, f: flow.Flow) -> None:
        if self.should_intercept(f):
            assert f.reply
            if f.reply.state != "start":
                return ctx.log.debug("Cannot intercept request that is already taken by another addon.")
            f.intercept()

            

Reported by Pylint.

              from mitmproxy.contentviews import wbxml
from . import full_eval

datadir = "mitmproxy/contentviews/test_wbxml_data/"


def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())


            

Reported by Pylint.

              
    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples
    with open(path, 'rb') as f:
        input = f.read()
    with open("-formatted.".join(path.rsplit(".", 1))) as f:
        expected = f.read()

    p = wbxml.ASCommandResponse.ASCommandResponse(input)
    assert p.xmlString == expected

            

Reported by Pylint.

              from mitmproxy.contentviews import wbxml
from . import full_eval

datadir = "mitmproxy/contentviews/test_wbxml_data/"


def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())


            

Reported by Pylint.

              from mitmproxy.contentviews import wbxml
from . import full_eval

datadir = "mitmproxy/contentviews/test_wbxml_data/"


def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())


            

Reported by Pylint.

              datadir = "mitmproxy/contentviews/test_wbxml_data/"


def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())

    assert v(b'\x03\x01\x6A\x00') == ('WBXML', [[('text', '<?xml version="1.0" ?>')]])
    assert v(b'foo') is None


            

Reported by Pylint.

              

def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())

    assert v(b'\x03\x01\x6A\x00') == ('WBXML', [[('text', '<?xml version="1.0" ?>')]])
    assert v(b'foo') is None

    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples

            

Reported by Pylint.

              def test_wbxml(tdata):
    v = full_eval(wbxml.ViewWBXML())

    assert v(b'\x03\x01\x6A\x00') == ('WBXML', [[('text', '<?xml version="1.0" ?>')]])
    assert v(b'foo') is None

    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples
    with open(path, 'rb') as f:
        input = f.read()

            

Reported by Bandit.

                  v = full_eval(wbxml.ViewWBXML())

    assert v(b'\x03\x01\x6A\x00') == ('WBXML', [[('text', '<?xml version="1.0" ?>')]])
    assert v(b'foo') is None

    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples
    with open(path, 'rb') as f:
        input = f.read()
    with open("-formatted.".join(path.rsplit(".", 1))) as f:

            

Reported by Bandit.

                  assert v(b'\x03\x01\x6A\x00') == ('WBXML', [[('text', '<?xml version="1.0" ?>')]])
    assert v(b'foo') is None

    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples
    with open(path, 'rb') as f:
        input = f.read()
    with open("-formatted.".join(path.rsplit(".", 1))) as f:
        expected = f.read()


            

Reported by Pylint.

                  assert v(b'foo') is None

    path = tdata.path(datadir + "data.wbxml")  # File taken from https://github.com/davidpshaw/PyWBXMLDecoder/tree/master/wbxml_samples
    with open(path, 'rb') as f:
        input = f.read()
    with open("-formatted.".join(path.rsplit(".", 1))) as f:
        expected = f.read()

    p = wbxml.ASCommandResponse.ASCommandResponse(input)

            

Reported by Pylint.

              
def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


def pretty(value, htchar="    ", lfchar="\n", indent=0):

            

Reported by Pylint.

              def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


def pretty(value, htchar="    ", lfchar="\n", indent=0):
    nlch = lfchar + htchar * (indent + 1)

            

Reported by Pylint.

              def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


def pretty(value, htchar="    ", lfchar="\n", indent=0):
    nlch = lfchar + htchar * (indent + 1)

            

Reported by Pylint.

              def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


def pretty(value, htchar="    ", lfchar="\n", indent=0):
    nlch = lfchar + htchar * (indent + 1)

            

Reported by Pylint.

              import typing

import msgpack


from mitmproxy.contentviews import base

PARSE_ERROR = object()


            

Reported by Pylint.

                      if data is not PARSE_ERROR:
            return "MsgPack", format_msgpack(data)

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

            

Reported by Pylint.

              import typing

import msgpack


from mitmproxy.contentviews import base

PARSE_ERROR = object()


            

Reported by Pylint.

              import msgpack


from mitmproxy.contentviews import base

PARSE_ERROR = object()


def parse_msgpack(s: bytes) -> typing.Any:

            

Reported by Pylint.

              PARSE_ERROR = object()


def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


            

Reported by Pylint.

              PARSE_ERROR = object()


def parse_msgpack(s: bytes) -> typing.Any:
    try:
        return msgpack.unpackb(s, raw=False)
    except (ValueError, msgpack.ExtraData, msgpack.FormatError, msgpack.StackError):
        return PARSE_ERROR


            

Reported by Pylint.

              import pytest

from mitmproxy import options
from mitmproxy.tools.console import statusbar, master


def test_statusbar(monkeypatch):
    o = options.Options()
    m = master.ConsoleMaster(o)

            

Reported by Pylint.

                  monkeypatch.setattr(statusbar.StatusBar, "refresh", lambda x: None)

    bar = statusbar.StatusBar(m)  # this already causes a redraw
    assert bar.ib._w


@pytest.mark.parametrize("message,ready_message", [
    ("", [(None, ""), ("warn", "")]),
    (("info", "Line fits into statusbar"), [("info", "Line fits into statusbar"),

            

Reported by Pylint.

              import pytest

from mitmproxy import options
from mitmproxy.tools.console import statusbar, master


def test_statusbar(monkeypatch):
    o = options.Options()
    m = master.ConsoleMaster(o)

            

Reported by Pylint.

              from mitmproxy.tools.console import statusbar, master


def test_statusbar(monkeypatch):
    o = options.Options()
    m = master.ConsoleMaster(o)
    m.options.update(
        modify_headers=[":~q:foo:bar"],
        modify_body=[":~q:foo:bar"],

            

Reported by Pylint.

              

def test_statusbar(monkeypatch):
    o = options.Options()
    m = master.ConsoleMaster(o)
    m.options.update(
        modify_headers=[":~q:foo:bar"],
        modify_body=[":~q:foo:bar"],
        ignore_hosts=["example.com", "example.org"],

            

Reported by Pylint.

              
def test_statusbar(monkeypatch):
    o = options.Options()
    m = master.ConsoleMaster(o)
    m.options.update(
        modify_headers=[":~q:foo:bar"],
        modify_body=[":~q:foo:bar"],
        ignore_hosts=["example.com", "example.org"],
        tcp_hosts=["example.tcp"],

            

Reported by Pylint.

                  monkeypatch.setattr(m.addons.get("serverplayback"), "count", lambda: 42)
    monkeypatch.setattr(statusbar.StatusBar, "refresh", lambda x: None)

    bar = statusbar.StatusBar(m)  # this already causes a redraw
    assert bar.ib._w


@pytest.mark.parametrize("message,ready_message", [
    ("", [(None, ""), ("warn", "")]),

            

Reported by Pylint.

                  monkeypatch.setattr(statusbar.StatusBar, "refresh", lambda x: None)

    bar = statusbar.StatusBar(m)  # this already causes a redraw
    assert bar.ib._w


@pytest.mark.parametrize("message,ready_message", [
    ("", [(None, ""), ("warn", "")]),
    (("info", "Line fits into statusbar"), [("info", "Line fits into statusbar"),

            

Reported by Bandit.

                                                         ("warn", "(more in eventlog)")]),
    ("Two long lines\nFirst doesn't fit", [(None, "Two long li\u2026"),
                                           ("warn", "(more in eventlog)")])
])
def test_shorten_message(message, ready_message):
    o = options.Options()
    m = master.ConsoleMaster(o)
    ab = statusbar.ActionBar(m)
    assert ab.shorten_message(message, max_width=30) == ready_message

            

Reported by Pylint.

                                                         ("warn", "(more in eventlog)")])
])
def test_shorten_message(message, ready_message):
    o = options.Options()
    m = master.ConsoleMaster(o)
    ab = statusbar.ActionBar(m)
    assert ab.shorten_message(message, max_width=30) == ready_message



            

Reported by Pylint.

                  for i in range(0, len(s), 16):
        offset = f"{i:0=10x}"
        part = s[i:i + 16]
        x = " ".join(f"{i:0=2x}" for i in part)
        x = x.ljust(47)  # 16*2 + 15
        part_repr = always_str(escape_control_characters(
            part.decode("ascii", "replace").replace("\ufffd", "."),
            False
        ))

            

Reported by Pylint.

                  buf = io.StringIO()
    parts = split_special_areas(data, area_delimiter)
    rex = re.compile(fr"[{control_characters}]")
    for i, x in enumerate(parts):
        if i % 2:
            x = rex.sub(_move_to_private_code_plane, x)
        buf.write(x)
    return buf.getvalue()


            

Reported by Pylint.

              import codecs
import io
import re
from typing import Iterable, Union, overload


# https://mypy.readthedocs.io/en/stable/more_types.html#function-overloading

@overload

            

Reported by Pylint.

                  ...


def always_bytes(str_or_bytes: Union[None, str, bytes], *encode_args) -> Union[None, bytes]:
    if str_or_bytes is None or isinstance(str_or_bytes, bytes):
        return str_or_bytes
    elif isinstance(str_or_bytes, str):
        return str_or_bytes.encode(*encode_args)
    else:

            

Reported by Pylint.

              

def always_bytes(str_or_bytes: Union[None, str, bytes], *encode_args) -> Union[None, bytes]:
    if str_or_bytes is None or isinstance(str_or_bytes, bytes):
        return str_or_bytes
    elif isinstance(str_or_bytes, str):
        return str_or_bytes.encode(*encode_args)
    else:
        raise TypeError("Expected str or bytes, but got {}.".format(type(str_or_bytes).__name__))

            

Reported by Pylint.

                  Returns,
        str_or_bytes unmodified, if
    """
    if str_or_bytes is None or isinstance(str_or_bytes, str):
        return str_or_bytes
    elif isinstance(str_or_bytes, bytes):
        return str_or_bytes.decode(*decode_args)
    else:
        raise TypeError("Expected str or bytes, but got {}.".format(type(str_or_bytes).__name__))

            

Reported by Pylint.

                  return text.translate(trans)


def bytes_to_escaped_str(data: bytes, keep_spacing: bool = False, escape_single_quotes: bool = False) -> str:
    """
    Take bytes and return a safe string that can be displayed to the user.

    Single quotes are always escaped, double quotes are never escaped:
        "'" + bytes_to_escaped_str(...) + "'"

            

Reported by Pylint.

                  return codecs.escape_decode(data)[0]  # type: ignore


def is_mostly_bin(s: bytes) -> bool:
    if not s or len(s) == 0:
        return False

    return sum(
        i < 9 or 13 < i < 32 or 126 < i

            

Reported by Pylint.

                  return codecs.escape_decode(data)[0]  # type: ignore


def is_mostly_bin(s: bytes) -> bool:
    if not s or len(s) == 0:
        return False

    return sum(
        i < 9 or 13 < i < 32 or 126 < i

            

Reported by Pylint.

                      return False

    return sum(
        i < 9 or 13 < i < 32 or 126 < i
        for i in s[:100]
    ) / len(s[:100]) > 0.3


def is_xml(s: bytes) -> bool:

            

Reported by Pylint.

                  def get_state(self):
        return copy.copy(self.i)

    def set_state(self, i):
        self.i = i

    @classmethod
    def from_state(cls, state):
        return cls(state)

            

Reported by Pylint.

              import copy

from mitmproxy.coretypes import serializable


class SerializableDummy(serializable.Serializable):
    def __init__(self, i):
        self.i = i


            

Reported by Pylint.

              from mitmproxy.coretypes import serializable


class SerializableDummy(serializable.Serializable):
    def __init__(self, i):
        self.i = i

    def get_state(self):
        return copy.copy(self.i)

            

Reported by Pylint.

                      return cls(state)


class TestSerializable:
    def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42

            

Reported by Pylint.

              

class TestSerializable:
    def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42


            

Reported by Pylint.

              

class TestSerializable:
    def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42


            

Reported by Pylint.

              
class TestSerializable:
    def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42

        a.set_state(1)

            

Reported by Pylint.

              class TestSerializable:
    def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42

        a.set_state(1)
        assert a.i == 1

            

Reported by Bandit.

                  def test_copy(self):
        a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42

        a.set_state(1)
        assert a.i == 1
        assert b.i == 42

            

Reported by Pylint.

                      a = SerializableDummy(42)
        assert a.i == 42
        b = a.copy()
        assert b.i == 42

        a.set_state(1)
        assert a.i == 1
        assert b.i == 42


            

Reported by Bandit.

              import urwid

from mitmproxy import flowfilter
from mitmproxy.tools.console import common
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import tabs


class CListBox(urwid.ListBox):

            

Reported by Pylint.

              import urwid

from mitmproxy import flowfilter
from mitmproxy.tools.console import common
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import tabs


class CListBox(urwid.ListBox):

            

Reported by Pylint.

              from mitmproxy.tools.console import tabs


class CListBox(urwid.ListBox):
    def __init__(self, contents):
        self.length = len(contents)
        contents = contents[:] + [urwid.Text(["\n"])] * 5
        super().__init__(contents)


            

Reported by Pylint.

              from mitmproxy.tools.console import tabs


class CListBox(urwid.ListBox):
    def __init__(self, contents):
        self.length = len(contents)
        contents = contents[:] + [urwid.Text(["\n"])] * 5
        super().__init__(contents)


            

Reported by Pylint.

                      contents = contents[:] + [urwid.Text(["\n"])] * 5
        super().__init__(contents)

    def keypress(self, size, key):
        if key == "m_end":
            self.set_focus(self.length - 1)
        elif key == "m_start":
            self.set_focus(0)
        else:

            

Reported by Pylint.

                      contents = contents[:] + [urwid.Text(["\n"])] * 5
        super().__init__(contents)

    def keypress(self, size, key):
        if key == "m_end":
            self.set_focus(self.length - 1)
        elif key == "m_start":
            self.set_focus(0)
        else:

            

Reported by Pylint.

                          return super().keypress(size, key)


class HelpView(tabs.Tabs, layoutwidget.LayoutWidget):
    title = "Help"
    keyctx = "help"

    def __init__(self, master):
        self.master = master

            

Reported by Pylint.

                          ]
        )

    def keybindings_title(self):
        return "Key Bindings"

    def format_keys(self, binds):
        kvs = []
        for b in binds:

            

Reported by Pylint.

                          ]
        )

    def keybindings_title(self):
        return "Key Bindings"

    def format_keys(self, binds):
        kvs = []
        for b in binds:

            

Reported by Pylint.

                  def keybindings_title(self):
        return "Key Bindings"

    def format_keys(self, binds):
        kvs = []
        for b in binds:
            k = b.key
            if b.key == " ":
                k = "space"

            

Reported by Pylint.

                      try:
            return int(data)
        except ValueError:
            raise ValueError(f"not a tnetstring: invalid integer literal: {data!r}")
    if data_type == ord(b'^'):
        try:
            return float(data)
        except ValueError:
            raise ValueError(f"not a tnetstring: invalid float literal: {data!r}")

            

Reported by Pylint.

                      try:
            return float(data)
        except ValueError:
            raise ValueError(f"not a tnetstring: invalid float literal: {data!r}")
    if data_type == ord(b'!'):
        if data == b'true':
            return True
        elif data == b'false':
            return False

            

Reported by Pylint.

                      blength, data = data.split(b':', 1)
        length = int(blength)
    except ValueError:
        raise ValueError(f"not a tnetstring: missing or invalid length prefix: {data!r}")
    try:
        data, data_type, remain = data[:length], data[length], data[length + 1:]
    except IndexError:
        #  This fires if len(data) < dlen, meaning we don't need
        #  to further validate that data is the right length.

            

Reported by Pylint.

                  except IndexError:
        #  This fires if len(data) < dlen, meaning we don't need
        #  to further validate that data is the right length.
        raise ValueError(f"not a tnetstring: invalid length prefix: {length}")
    # Parse the data based on the type tag.
    return parse(data_type, data), remain


__all__ = ["dump", "dumps", "load", "loads", "pop"]

            

Reported by Pylint.

                  #  This uses a deque to collect output fragments in reverse order,
    #  then joins them together at the end.  It's measurably faster
    #  than creating all the intermediate strings.
    q: collections.deque = collections.deque()
    _rdumpq(q, 0, value)
    return b''.join(q)


def dump(value: TSerializable, file_handle: typing.IO[bytes]) -> None:

            

Reported by Pylint.

                  file_handle.write(dumps(value))


def _rdumpq(q: collections.deque, size: int, value: TSerializable) -> int:
    """
    Dump value as a tnetstring, to a deque instance, last chunks first.

    This function generates the tnetstring representation of the given value,
    pushing chunks of the output onto the given deque instance.  It pushes

            

Reported by Pylint.

                  file_handle.write(dumps(value))


def _rdumpq(q: collections.deque, size: int, value: TSerializable) -> int:
    """
    Dump value as a tnetstring, to a deque instance, last chunks first.

    This function generates the tnetstring representation of the given value,
    pushing chunks of the output onto the given deque instance.  It pushes

            

Reported by Pylint.

                  file_handle.write(dumps(value))


def _rdumpq(q: collections.deque, size: int, value: TSerializable) -> int:
    """
    Dump value as a tnetstring, to a deque instance, last chunks first.

    This function generates the tnetstring representation of the given value,
    pushing chunks of the output onto the given deque instance.  It pushes

            

Reported by Pylint.

                  strings, especially on deeply nested structures.
    """
    write = q.appendleft
    if value is None:
        write(b'0:~')
        return size + 3
    elif value is True:
        write(b'4:true!')
        return size + 7

            

Reported by Pylint.

                  elif isinstance(value, dict):
        write(b'}')
        init_size = size = size + 1
        for (k, v) in value.items():
            size = _rdumpq(q, size, v)
            size = _rdumpq(q, size, k)
        span = str(size - init_size).encode()
        write(b':')
        write(span)

            

Reported by Pylint.

              from mitmproxy.contentviews import image
from .. import full_eval


def test_view_image(tdata):
    v = full_eval(image.ViewImage())
    for img in [
        "mitmproxy/data/image.png",
        "mitmproxy/data/image.gif",

            

Reported by Pylint.

                      "mitmproxy/data/image.ico",
    ]:
        with open(tdata.path(img), "rb") as f:
            viewname, lines = v(f.read())
            assert img.split(".")[-1].upper() in viewname

    assert v(b"flibble") == ('Unknown Image', [[('header', 'Image Format: '), ('text', 'unknown')]])



            

Reported by Pylint.

              from mitmproxy.contentviews import image
from .. import full_eval


def test_view_image(tdata):
    v = full_eval(image.ViewImage())
    for img in [
        "mitmproxy/data/image.png",
        "mitmproxy/data/image.gif",

            

Reported by Pylint.

              from .. import full_eval


def test_view_image(tdata):
    v = full_eval(image.ViewImage())
    for img in [
        "mitmproxy/data/image.png",
        "mitmproxy/data/image.gif",
        "mitmproxy/data/all.jpeg",

            

Reported by Pylint.

              

def test_view_image(tdata):
    v = full_eval(image.ViewImage())
    for img in [
        "mitmproxy/data/image.png",
        "mitmproxy/data/image.gif",
        "mitmproxy/data/all.jpeg",
        "mitmproxy/data/image.ico",

            

Reported by Pylint.

                      "mitmproxy/data/all.jpeg",
        "mitmproxy/data/image.ico",
    ]:
        with open(tdata.path(img), "rb") as f:
            viewname, lines = v(f.read())
            assert img.split(".")[-1].upper() in viewname

    assert v(b"flibble") == ('Unknown Image', [[('header', 'Image Format: '), ('text', 'unknown')]])


            

Reported by Pylint.

                  ]:
        with open(tdata.path(img), "rb") as f:
            viewname, lines = v(f.read())
            assert img.split(".")[-1].upper() in viewname

    assert v(b"flibble") == ('Unknown Image', [[('header', 'Image Format: '), ('text', 'unknown')]])


def test_render_priority():

            

Reported by Bandit.

                          viewname, lines = v(f.read())
            assert img.split(".")[-1].upper() in viewname

    assert v(b"flibble") == ('Unknown Image', [[('header', 'Image Format: '), ('text', 'unknown')]])


def test_render_priority():
    v = image.ViewImage()
    assert v.render_priority(b"", content_type="image/png")

            

Reported by Bandit.

                  assert v(b"flibble") == ('Unknown Image', [[('header', 'Image Format: '), ('text', 'unknown')]])


def test_render_priority():
    v = image.ViewImage()
    assert v.render_priority(b"", content_type="image/png")
    assert v.render_priority(b"", content_type="image/jpeg")
    assert v.render_priority(b"", content_type="image/gif")
    assert v.render_priority(b"", content_type="image/vnd.microsoft.icon")

            

Reported by Pylint.

              

def test_render_priority():
    v = image.ViewImage()
    assert v.render_priority(b"", content_type="image/png")
    assert v.render_priority(b"", content_type="image/jpeg")
    assert v.render_priority(b"", content_type="image/gif")
    assert v.render_priority(b"", content_type="image/vnd.microsoft.icon")
    assert v.render_priority(b"", content_type="image/x-icon")

            

Reported by Pylint.

              import pytest

from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layers import tcp
from mitmproxy.proxy.layers.tcp import TcpMessageInjected
from mitmproxy.tcp import TCPFlow, TCPMessage
from ..tutils import Placeholder, Playbook, reply


            

Reported by Pylint.

              from mitmproxy.proxy.layers import tcp
from mitmproxy.proxy.layers.tcp import TcpMessageInjected
from mitmproxy.tcp import TCPFlow, TCPMessage
from ..tutils import Placeholder, Playbook, reply


def test_open_connection(tctx):
    """
    If there is no server connection yet, establish one,

            

Reported by Pylint.

              import pytest

from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layers import tcp
from mitmproxy.proxy.layers.tcp import TcpMessageInjected
from mitmproxy.tcp import TCPFlow, TCPMessage
from ..tutils import Placeholder, Playbook, reply


            

Reported by Pylint.

                  If there is no server connection yet, establish one,
    because the server may send data first.
    """
    assert (
            Playbook(tcp.TCPLayer(tctx, True))
            << OpenConnection(tctx.server)
    )

    tctx.server.timestamp_start = 1624544785

            

Reported by Bandit.

                  )

    tctx.server.timestamp_start = 1624544785
    assert (
            Playbook(tcp.TCPLayer(tctx, True))
            << None
    )



            

Reported by Bandit.

                  )


def test_open_connection_err(tctx):
    f = Placeholder(TCPFlow)
    assert (
            Playbook(tcp.TCPLayer(tctx))
            << tcp.TcpStartHook(f)
            >> reply()

            

Reported by Pylint.

              

def test_open_connection_err(tctx):
    f = Placeholder(TCPFlow)
    assert (
            Playbook(tcp.TCPLayer(tctx))
            << tcp.TcpStartHook(f)
            >> reply()
            << OpenConnection(tctx.server)

            

Reported by Pylint.

              
def test_open_connection_err(tctx):
    f = Placeholder(TCPFlow)
    assert (
            Playbook(tcp.TCPLayer(tctx))
            << tcp.TcpStartHook(f)
            >> reply()
            << OpenConnection(tctx.server)
            >> reply("Connect call failed")

            

Reported by Bandit.

              
def test_simple(tctx):
    """open connection, receive data, send it to peer"""
    f = Placeholder(TCPFlow)

    assert (
            Playbook(tcp.TCPLayer(tctx))
            << tcp.TcpStartHook(f)
            >> reply()

            

Reported by Pylint.

                  """open connection, receive data, send it to peer"""
    f = Placeholder(TCPFlow)

    assert (
            Playbook(tcp.TCPLayer(tctx))
            << tcp.TcpStartHook(f)
            >> reply()
            << OpenConnection(tctx.server)
            >> reply(None)

            

Reported by Bandit.