import pytest

from mitmproxy.proxy.utils import expect


def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):

            

Reported by Pylint.

              import pytest

from mitmproxy.proxy.utils import expect


def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):

            

Reported by Pylint.

              from mitmproxy.proxy.utils import expect


def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))


            

Reported by Pylint.

              

def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))

        @expect(str)

            

Reported by Pylint.

              def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

            

Reported by Pylint.

              def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

            

Reported by Pylint.

              def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

            

Reported by Pylint.

              def test_expect():
    class Foo:
        @expect(str, int)
        def foo(self, x):
            return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

            

Reported by Pylint.

                          return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

    f = Foo()

    assert f.foo("foo") == "oof"

            

Reported by Pylint.

                          return "".join(reversed(x))

        @expect(str)
        def bar(self, x):
            yield "".join(reversed(x))

    f = Foo()

    assert f.foo("foo") == "oof"

            

Reported by Pylint.

              """
from typing import Optional

from mitmproxy import contentviews, flow
from mitmproxy import http


class ViewSwapCase(contentviews.View):
    name = "swapcase"

            

Reported by Pylint.

              from typing import Optional

from mitmproxy import contentviews, flow
from mitmproxy import http


class ViewSwapCase(contentviews.View):
    name = "swapcase"


            

Reported by Pylint.

                      return "case-swapped text", contentviews.format_text(data.swapcase())

    def render_priority(
        self,
        data: bytes,
        *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,
        http_message: Optional[http.Message] = None,

            

Reported by Pylint.

              
    def render_priority(
        self,
        data: bytes,
        *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,
        http_message: Optional[http.Message] = None,
        **unknown_metadata,

            

Reported by Pylint.

                      data: bytes,
        *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,
        http_message: Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        if content_type == "text/plain":
            return 1

            

Reported by Pylint.

                      data: bytes,
        *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,
        http_message: Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        if content_type == "text/plain":
            return 1

            

Reported by Pylint.

                      *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,
        http_message: Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        if content_type == "text/plain":
            return 1
        else:

            

Reported by Pylint.

              view = ViewSwapCase()


def load(l):
    contentviews.add(view)


def done():
    contentviews.remove(view)

            

Reported by Pylint.

              from mitmproxy import http


class ViewSwapCase(contentviews.View):
    name = "swapcase"

    def __call__(self, data, **metadata) -> contentviews.TViewResult:
        return "case-swapped text", contentviews.format_text(data.swapcase())


            

Reported by Pylint.

                  def __call__(self, data, **metadata) -> contentviews.TViewResult:
        return "case-swapped text", contentviews.format_text(data.swapcase())

    def render_priority(
        self,
        data: bytes,
        *,
        content_type: Optional[str] = None,
        flow: Optional[flow.Flow] = None,

            

Reported by Pylint.

              import pytest

from mitmproxy.http import Headers
from mitmproxy.net.http import multipart


def test_decode():
    boundary = 'somefancyboundary'
    content = (

            

Reported by Pylint.

              import pytest

from mitmproxy.http import Headers
from mitmproxy.net.http import multipart


def test_decode():
    boundary = 'somefancyboundary'
    content = (

            

Reported by Pylint.

              from mitmproxy.net.http import multipart


def test_decode():
    boundary = 'somefancyboundary'
    content = (
        "--{0}\n"
        "Content-Disposition: form-data; name=\"field1\"\n\n"
        "value1\n"

            

Reported by Pylint.

                  )
    form = multipart.decode(f'multipart/form-data; boundary={boundary}', content)

    assert len(form) == 2
    assert form[0] == (b"field1", b"value1")
    assert form[1] == (b"field2", b"value2")

    boundary = 'boundary茅莽'
    result = multipart.decode(f'multipart/form-data; boundary={boundary}', content)

            

Reported by Bandit.

                  form = multipart.decode(f'multipart/form-data; boundary={boundary}', content)

    assert len(form) == 2
    assert form[0] == (b"field1", b"value1")
    assert form[1] == (b"field2", b"value2")

    boundary = 'boundary茅莽'
    result = multipart.decode(f'multipart/form-data; boundary={boundary}', content)
    assert result == []

            

Reported by Bandit.

              
    assert len(form) == 2
    assert form[0] == (b"field1", b"value1")
    assert form[1] == (b"field2", b"value2")

    boundary = 'boundary茅莽'
    result = multipart.decode(f'multipart/form-data; boundary={boundary}', content)
    assert result == []


            

Reported by Bandit.

              
    boundary = 'boundary茅莽'
    result = multipart.decode(f'multipart/form-data; boundary={boundary}', content)
    assert result == []

    assert multipart.decode("", content) == []


def test_encode():

            

Reported by Bandit.

                  result = multipart.decode(f'multipart/form-data; boundary={boundary}', content)
    assert result == []

    assert multipart.decode("", content) == []


def test_encode():
    data = [(b"file", b"shell.jpg"),
            (b"file_size", b"1000")]

            

Reported by Bandit.

                  assert multipart.decode("", content) == []


def test_encode():
    data = [(b"file", b"shell.jpg"),
            (b"file_size", b"1000")]
    headers = Headers(
        content_type='multipart/form-data; boundary=127824672498'
    )

            

Reported by Pylint.

                  )
    content = multipart.encode(headers, data)

    assert b'Content-Disposition: form-data; name="file"' in content
    assert b'Content-Type: text/plain; charset=utf-8\r\n\r\nshell.jpg\r\n\r\n--127824672498\r\n' in content
    assert b'1000\r\n\r\n--127824672498--\r\n'
    assert len(content) == 252

    with pytest.raises(ValueError, match=r"boundary found in encoded string"):

            

Reported by Bandit.

              
KEY_MAX = 30

TTextType = typing.Union[str, bytes]  # FIXME: This should be either bytes or str ultimately.
TViewLine = typing.List[typing.Tuple[str, TTextType]]
TViewResult = typing.Tuple[str, typing.Iterator[TViewLine]]


class View(ABC):

            

Reported by Pylint.

                      data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,
    ) -> TViewResult:
        """
        Transform raw data into human-readable output.

            

Reported by Pylint.

                      raise NotImplementedError()  # pragma: no cover

    def render_priority(
        self,
        data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,

            

Reported by Pylint.

              
    def render_priority(
        self,
        data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,

            

Reported by Pylint.

                      self,
        data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        """

            

Reported by Pylint.

                      data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        """
        Return the priority of this view for rendering `data`.

            

Reported by Pylint.

                      data: bytes,
        *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        """
        Return the priority of this view for rendering `data`.

            

Reported by Pylint.

                      *,
        content_type: typing.Optional[str] = None,
        flow: typing.Optional[flow.Flow] = None,
        http_message: typing.Optional[http.Message] = None,
        **unknown_metadata,
    ) -> float:
        """
        Return the priority of this view for rendering `data`.
        If no particular view is chosen by the user, the view with the highest priority is selected.

            

Reported by Pylint.

              # Default view cutoff *in lines*
import typing
from abc import ABC, abstractmethod

from mitmproxy import flow
from mitmproxy import http

KEY_MAX = 30


            

Reported by Pylint.

              TViewResult = typing.Tuple[str, typing.Iterator[TViewLine]]


class View(ABC):
    name: typing.ClassVar[str]

    @abstractmethod
    def __call__(
        self,

            

Reported by Pylint.

              import asyncio
import pytest

from mitmproxy.addons import keepserving
from mitmproxy.test import taddons
from mitmproxy import command


class Dummy:

            

Reported by Pylint.

                  _is_shutdown = False

    def shutdown(self):
        self.is_shutdown = True


@pytest.mark.asyncio
async def test_keepserving():
    ks = TKS()

            

Reported by Pylint.

              import asyncio
import pytest

from mitmproxy.addons import keepserving
from mitmproxy.test import taddons
from mitmproxy import command


class Dummy:

            

Reported by Pylint.

              from mitmproxy import command


class Dummy:
    def __init__(self, val: bool):
        self.val = val

    def load(self, loader):
        loader.add_option("client_replay", bool, self.val, "test")

            

Reported by Pylint.

                  def __init__(self, val: bool):
        self.val = val

    def load(self, loader):
        loader.add_option("client_replay", bool, self.val, "test")
        loader.add_option("server_replay", bool, self.val, "test")
        loader.add_option("rfile", bool, self.val, "test")

    @command.command("readfile.reading")

            

Reported by Pylint.

                      loader.add_option("rfile", bool, self.val, "test")

    @command.command("readfile.reading")
    def readfile(self) -> bool:
        return self.val

    @command.command("replay.client.count")
    def creplay(self) -> int:
        return 1 if self.val else 0

            

Reported by Pylint.

                      return self.val

    @command.command("replay.client.count")
    def creplay(self) -> int:
        return 1 if self.val else 0

    @command.command("replay.server.count")
    def sreplay(self) -> int:
        return 1 if self.val else 0

            

Reported by Pylint.

                      return 1 if self.val else 0

    @command.command("replay.server.count")
    def sreplay(self) -> int:
        return 1 if self.val else 0


class TKS(keepserving.KeepServing):
    _is_shutdown = False

            

Reported by Pylint.

                      return 1 if self.val else 0


class TKS(keepserving.KeepServing):
    _is_shutdown = False

    def shutdown(self):
        self.is_shutdown = True


            

Reported by Pylint.

              

@pytest.mark.asyncio
async def test_keepserving():
    ks = TKS()
    d = Dummy(True)
    with taddons.context(ks) as tctx:
        tctx.master.addons.add(d)
        ks.running()

            

Reported by Pylint.

              import argparse

import pytest

from mitmproxy import options
from mitmproxy.tools import cmdline
from mitmproxy.tools import main



            

Reported by Pylint.

                      opts = options.Options()
        cmdline.common_options(parser, opts)
        args = parser.parse_args(args=args)
        pconf = main.process_options(parser, opts, args)
        return parser, pconf

    def assert_noerr(self, *args):
        m, p = self.p(*args)
        assert p

            

Reported by Pylint.

                      return parser, pconf

    def assert_noerr(self, *args):
        m, p = self.p(*args)
        assert p
        return p

    def test_simple(self):
        assert self.p()

            

Reported by Pylint.

              import argparse

import pytest

from mitmproxy import options
from mitmproxy.tools import cmdline
from mitmproxy.tools import main



            

Reported by Pylint.

                      raise Exception(message)


class TestProcessProxyOptions:

    def p(self, *args):
        parser = MockParser()
        opts = options.Options()
        cmdline.common_options(parser, opts)

            

Reported by Pylint.

              
class TestProcessProxyOptions:

    def p(self, *args):
        parser = MockParser()
        opts = options.Options()
        cmdline.common_options(parser, opts)
        args = parser.parse_args(args=args)
        pconf = main.process_options(parser, opts, args)

            

Reported by Pylint.

              
class TestProcessProxyOptions:

    def p(self, *args):
        parser = MockParser()
        opts = options.Options()
        cmdline.common_options(parser, opts)
        args = parser.parse_args(args=args)
        pconf = main.process_options(parser, opts, args)

            

Reported by Pylint.

              
class TestProcessProxyOptions:

    def p(self, *args):
        parser = MockParser()
        opts = options.Options()
        cmdline.common_options(parser, opts)
        args = parser.parse_args(args=args)
        pconf = main.process_options(parser, opts, args)

            

Reported by Pylint.

                      pconf = main.process_options(parser, opts, args)
        return parser, pconf

    def assert_noerr(self, *args):
        m, p = self.p(*args)
        assert p
        return p

    def test_simple(self):

            

Reported by Pylint.

                      return parser, pconf

    def assert_noerr(self, *args):
        m, p = self.p(*args)
        assert p
        return p

    def test_simple(self):
        assert self.p()

            

Reported by Pylint.

              from dataclasses import dataclass

import pytest

from mitmproxy import hooks


def test_hook():
    with pytest.raises(TypeError, match="may not be instantiated directly"):

            

Reported by Pylint.

              
    with pytest.warns(RuntimeWarning, match="Two conflicting event classes"):
        @dataclass
        class FooHook2(hooks.Hook):
            name = "foo"

    @dataclass
    class AnotherABC(hooks.Hook):
        name = ""

            

Reported by Pylint.

              from dataclasses import dataclass

import pytest

from mitmproxy import hooks


def test_hook():
    with pytest.raises(TypeError, match="may not be instantiated directly"):

            

Reported by Pylint.

              from mitmproxy import hooks


def test_hook():
    with pytest.raises(TypeError, match="may not be instantiated directly"):
        hooks.Hook()

    class NoDataClass(hooks.Hook):
        pass

            

Reported by Pylint.

                  with pytest.raises(TypeError, match="may not be instantiated directly"):
        hooks.Hook()

    class NoDataClass(hooks.Hook):
        pass

    with pytest.raises(TypeError, match="not a dataclass"):
        NoDataClass()


            

Reported by Pylint.

                  with pytest.raises(TypeError, match="may not be instantiated directly"):
        hooks.Hook()

    class NoDataClass(hooks.Hook):
        pass

    with pytest.raises(TypeError, match="not a dataclass"):
        NoDataClass()


            

Reported by Pylint.

                      NoDataClass()

    @dataclass
    class FooHook(hooks.Hook):
        data: bytes

    e = FooHook(b"foo")
    assert repr(e)
    assert e.args() == [b"foo"]

            

Reported by Pylint.

                  class FooHook(hooks.Hook):
        data: bytes

    e = FooHook(b"foo")
    assert repr(e)
    assert e.args() == [b"foo"]
    assert FooHook in hooks.all_hooks.values()

    with pytest.warns(RuntimeWarning, match="Two conflicting event classes"):

            

Reported by Pylint.

                      data: bytes

    e = FooHook(b"foo")
    assert repr(e)
    assert e.args() == [b"foo"]
    assert FooHook in hooks.all_hooks.values()

    with pytest.warns(RuntimeWarning, match="Two conflicting event classes"):
        @dataclass

            

Reported by Bandit.

              
    e = FooHook(b"foo")
    assert repr(e)
    assert e.args() == [b"foo"]
    assert FooHook in hooks.all_hooks.values()

    with pytest.warns(RuntimeWarning, match="Two conflicting event classes"):
        @dataclass
        class FooHook2(hooks.Hook):

            

Reported by Bandit.

              from mitmproxy import ctx
event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log


            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

                      ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")


def configure(updated):
    event_log.append("scriptconfigure")

            

Reported by Pylint.

                      event_log.append("addonconfigure")


def configure(updated):
    event_log.append("scriptconfigure")


def load(l):
    event_log.append("scriptload")

            

Reported by Pylint.

                  event_log.append("scriptconfigure")


def load(l):
    event_log.append("scriptload")


addons = [Addon()]

            

Reported by Pylint.

              from mitmproxy import ctx
event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log


            

Reported by Pylint.

              event_log = []


class Addon:
    @property
    def event_log(self):
        return event_log

    def load(self, opts):

            

Reported by Pylint.

              
class Addon:
    @property
    def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

                  def event_log(self):
        return event_log

    def load(self, opts):
        ctx.log.info("addon running")
        event_log.append("addonload")

    def configure(self, updated):
        event_log.append("addonconfigure")

            

Reported by Pylint.

              from dataclasses import dataclass

import pytest

from mitmproxy import connection
from mitmproxy.hooks import all_hooks
from mitmproxy.proxy import commands



            

Reported by Pylint.

                  return connection.Server(None)


def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))

            

Reported by Pylint.

              from dataclasses import dataclass

import pytest

from mitmproxy import connection
from mitmproxy.hooks import all_hooks
from mitmproxy.proxy import commands



            

Reported by Pylint.

              

@pytest.fixture
def tconn() -> connection.Server:
    return connection.Server(None)


def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))

            

Reported by Pylint.

                  return connection.Server(None)


def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))

            

Reported by Pylint.

              

def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))


            

Reported by Bandit.

              
def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))



            

Reported by Bandit.

              def test_dataclasses(tconn):
    assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))


def test_start_hook():

            

Reported by Bandit.

                  assert repr(commands.SendData(tconn, b"foo"))
    assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))


def test_start_hook():
    with pytest.raises(TypeError):

            

Reported by Bandit.

                  assert repr(commands.OpenConnection(tconn))
    assert repr(commands.CloseConnection(tconn))
    assert repr(commands.GetSocket(tconn))
    assert repr(commands.Log("hello", "info"))


def test_start_hook():
    with pytest.raises(TypeError):
        commands.StartHook()

            

Reported by Bandit.

              #!/usr/bin/env python3
import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):

            

Reported by Pylint.

              import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()

            

Reported by Pylint.

              
from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:

            

Reported by Pylint.

              #!/usr/bin/env python3
import click

from mitmproxy.addons import dumper
from mitmproxy.test import tflow
from mitmproxy.test import taddons


def show(flow_detail, flows):

            

Reported by Pylint.

              from mitmproxy.test import taddons


def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)

            

Reported by Pylint.

              

def show(flow_detail, flows):
    d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)


            

Reported by Pylint.

                  d = dumper.Dumper()
    with taddons.context() as ctx:
        ctx.configure(d, flow_detail=flow_detail)
        for f in flows:
            ctx.cycle(d, f)


@click.group()
def cli():

            

Reported by Pylint.

              

@click.group()
def cli():
    pass


@cli.command()
@click.option('--level', default=1, help='Detail level')

            

Reported by Pylint.

              
@cli.command()
@click.option('--level', default=1, help='Detail level')
def tcp(level):
    f1 = tflow.ttcpflow(client_conn=True, server_conn=True)
    show(level, [f1])


@cli.command()

            

Reported by Pylint.

              @cli.command()
@click.option('--level', default=1, help='Detail level')
def tcp(level):
    f1 = tflow.ttcpflow(client_conn=True, server_conn=True)
    show(level, [f1])


@cli.command()
@click.option('--level', default=1, help='Detail level')

            

Reported by Pylint.