import copy
import io
import pytest
import typing
import argparse

from mitmproxy import options
from mitmproxy import optmanager
from mitmproxy import exceptions

            

Reported by Pylint.

                  assert o.one == 1

    with pytest.raises(TypeError):
        TO(nonexistent = "value")
    with pytest.raises(Exception, match="Unknown options"):
        o.nonexistent = "value"
    with pytest.raises(Exception, match="Unknown options"):
        o.update(nonexistent = "value")
    assert o.update_known(nonexistent = "value") == {"nonexistent": "value"}

            

Reported by Pylint.

              def test_required_int():
    o = TO()
    with pytest.raises(exceptions.OptionsError):
        o.parse_setval(o._options["required_int"], None, None)


def test_deepcopy():
    o = TD()
    copy.deepcopy(o)

            

Reported by Pylint.

              
    assert o.one is None
    assert o.two == 2
    o.one = 1
    assert o.one == 1

    with pytest.raises(TypeError):
        TO(nonexistent = "value")
    with pytest.raises(Exception, match="Unknown options"):

            

Reported by Pylint.

                  with pytest.raises(TypeError):
        TO(nonexistent = "value")
    with pytest.raises(Exception, match="Unknown options"):
        o.nonexistent = "value"
    with pytest.raises(Exception, match="Unknown options"):
        o.update(nonexistent = "value")
    assert o.update_known(nonexistent = "value") == {"nonexistent": "value"}

    rec = []

            

Reported by Pylint.

              
    rec = []

    def sub(opts, updated):
        rec.append(copy.copy(opts))

    o.changed.connect(sub)

    o.one = 90

            

Reported by Pylint.

              
    o.changed.connect(sub)

    o.one = 90
    assert len(rec) == 1
    assert rec[-1].one == 90

    o.update(one=3)
    assert len(rec) == 2

            

Reported by Pylint.

                  assert len(o.changed.receivers) == 0

    o.subscribe(r, ["two"])
    o.one = 2
    assert not r.called
    o.two = 3
    assert r.called

    assert len(o.changed.receivers) == 1

            

Reported by Pylint.

                  o.subscribe(r, ["two"])
    o.one = 2
    assert not r.called
    o.two = 3
    assert r.called

    assert len(o.changed.receivers) == 1
    del r
    o.two = 4

            

Reported by Pylint.

              
    assert len(o.changed.receivers) == 1
    del r
    o.two = 4
    assert len(o.changed.receivers) == 0

    class binder:
        def __init__(self):
            self.o = TO()

            

Reported by Pylint.

              from typing import List, Tuple

import h2.settings
import hpack
import hyperframe.frame
import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server

            

Reported by Pylint.

              from typing import List, Tuple

import h2.settings
import hpack
import hyperframe.frame
import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server

            

Reported by Pylint.

              
import h2.settings
import hpack
import hyperframe.frame
import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error

            

Reported by Pylint.

              import h2.settings
import hpack
import hyperframe.frame
import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request

            

Reported by Pylint.

              import hpack
import hyperframe.frame
import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes

            

Reported by Pylint.

              import pytest
from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.context import Context

            

Reported by Pylint.

              from h2.errors import ErrorCodes

from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import ConnectionClosed, DataReceived

            

Reported by Pylint.

              
from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layers import http

            

Reported by Pylint.

              from mitmproxy.connection import ConnectionState, Server
from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layers import http
from mitmproxy.proxy.layers.http import HTTPMode

            

Reported by Pylint.

              from mitmproxy.flow import Error
from mitmproxy.http import HTTPFlow, Headers, Request
from mitmproxy.net.http import status_codes
from mitmproxy.proxy.commands import CloseConnection, OpenConnection, SendData
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import ConnectionClosed, DataReceived
from mitmproxy.proxy.layers import http
from mitmproxy.proxy.layers.http import HTTPMode
from mitmproxy.proxy.layers.http._http2 import Http2Client, split_pseudo_headers

            

Reported by Pylint.

              import pytest
import requests
from examples.complex import xss_scanner as xss
from mitmproxy.test import tflow, tutils


class TestXSSScanner():
    def test_get_XSS_info(self):
        # First type of exploit: <script>PAYLOAD</script>

            

Reported by Pylint.

              import pytest
import requests
from examples.complex import xss_scanner as xss
from mitmproxy.test import tflow, tutils


class TestXSSScanner():
    def test_get_XSS_info(self):
        # First type of exploit: <script>PAYLOAD</script>

            

Reported by Pylint.

              import pytest
import requests
from examples.complex import xss_scanner as xss
from mitmproxy.test import tflow, tutils


class TestXSSScanner():
    def test_get_XSS_info(self):
        # First type of exploit: <script>PAYLOAD</script>

            

Reported by Pylint.

                      assert text == expected_text
        assert xss.paths_to_text("""<html></html>""", "STRING") == []

    def mocked_requests_vuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)


            

Reported by Pylint.

                              self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)

    def mocked_requests_invuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html></html>")


            

Reported by Pylint.

                      assert text == expected_text
        assert xss.paths_to_text("""<html></html>""", "STRING") == []

    def mocked_requests_vuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)


            

Reported by Pylint.

                      assert text == expected_text
        assert xss.paths_to_text("""<html></html>""", "STRING") == []

    def mocked_requests_vuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)


            

Reported by Pylint.

              
    def mocked_requests_vuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)

    def mocked_requests_invuln(*args, headers=None, cookies=None):
        class MockResponse:

            

Reported by Pylint.

              
    def mocked_requests_vuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)

    def mocked_requests_invuln(*args, headers=None, cookies=None):
        class MockResponse:

            

Reported by Pylint.

                              self.text = html
        return MockResponse("<html>%s</html>" % xss.FULL_PAYLOAD)

    def mocked_requests_invuln(*args, headers=None, cookies=None):
        class MockResponse:
            def __init__(self, html, headers=None, cookies=None):
                self.text = html
        return MockResponse("<html></html>")


            

Reported by Pylint.

              import pytest
import os
import typing
import contextlib

import mitmproxy.exceptions
import mitmproxy.types
from mitmproxy.test import taddons
from mitmproxy.test import tflow

            

Reported by Pylint.

              from mitmproxy import command
from mitmproxy import flow

from . import test_command


@contextlib.contextmanager
def chdir(path: str):
    old_dir = os.getcwd()

            

Reported by Pylint.

              
def test_bool():
    with taddons.context() as tctx:
        b = mitmproxy.types._BoolType()
        assert b.completion(tctx.master.commands, bool, "b") == ["false", "true"]
        assert b.parse(tctx.master.commands, bool, "true") is True
        assert b.parse(tctx.master.commands, bool, "false") is False
        assert b.is_valid(tctx.master.commands, bool, True) is True
        assert b.is_valid(tctx.master.commands, bool, "foo") is False

            

Reported by Pylint.

              
def test_str():
    with taddons.context() as tctx:
        b = mitmproxy.types._StrType()
        assert b.is_valid(tctx.master.commands, str, "foo") is True
        assert b.is_valid(tctx.master.commands, str, 1) is False
        assert b.completion(tctx.master.commands, str, "") == []
        assert b.parse(tctx.master.commands, str, "foo") == "foo"
        assert b.parse(tctx.master.commands, str, r"foo\nbar") == "foo\nbar"

            

Reported by Pylint.

              
def test_bytes():
    with taddons.context() as tctx:
        b = mitmproxy.types._BytesType()
        assert b.is_valid(tctx.master.commands, bytes, b"foo") is True
        assert b.is_valid(tctx.master.commands, bytes, 1) is False
        assert b.completion(tctx.master.commands, bytes, "") == []
        assert b.parse(tctx.master.commands, bytes, "foo") == b"foo"
        with pytest.raises(mitmproxy.exceptions.TypeError):

            

Reported by Pylint.

              
def test_unknown():
    with taddons.context() as tctx:
        b = mitmproxy.types._UnknownType()
        assert b.is_valid(tctx.master.commands, mitmproxy.types.Unknown, "foo") is False
        assert b.is_valid(tctx.master.commands, mitmproxy.types.Unknown, 1) is False
        assert b.completion(tctx.master.commands, mitmproxy.types.Unknown, "") == []
        assert b.parse(tctx.master.commands, mitmproxy.types.Unknown, "foo") == "foo"


            

Reported by Pylint.

              
def test_int():
    with taddons.context() as tctx:
        b = mitmproxy.types._IntType()
        assert b.is_valid(tctx.master.commands, int, "foo") is False
        assert b.is_valid(tctx.master.commands, int, 1) is True
        assert b.completion(tctx.master.commands, int, "b") == []
        assert b.parse(tctx.master.commands, int, "1") == 1
        assert b.parse(tctx.master.commands, int, "999") == 999

            

Reported by Pylint.

              
def test_path(tdata, monkeypatch):
    with taddons.context() as tctx:
        b = mitmproxy.types._PathType()
        assert b.parse(tctx.master.commands, mitmproxy.types.Path, "/foo") == "/foo"
        assert b.parse(tctx.master.commands, mitmproxy.types.Path, "/bar") == "/bar"
        monkeypatch.setenv("HOME", "/home/test")
        monkeypatch.setenv("USERPROFILE", "/home/test")
        assert b.parse(tctx.master.commands, mitmproxy.types.Path, "~/mitm") == "/home/test/mitm"

            

Reported by Pylint.

              def test_cmd():
    with taddons.context() as tctx:
        tctx.master.addons.add(test_command.TAddon())
        b = mitmproxy.types._CmdType()
        assert b.is_valid(tctx.master.commands, mitmproxy.types.Cmd, "foo") is False
        assert b.is_valid(tctx.master.commands, mitmproxy.types.Cmd, "cmd1") is True
        assert b.parse(tctx.master.commands, mitmproxy.types.Cmd, "cmd1") == "cmd1"
        with pytest.raises(mitmproxy.exceptions.TypeError):
            assert b.parse(tctx.master.commands, mitmproxy.types.Cmd, "foo")

            

Reported by Pylint.

              
def test_cutspec():
    with taddons.context() as tctx:
        b = mitmproxy.types._CutSpecType()
        b.parse(tctx.master.commands, mitmproxy.types.CutSpec, "foo,bar") == ["foo", "bar"]
        assert b.is_valid(tctx.master.commands, mitmproxy.types.CutSpec, 1) is False
        assert b.is_valid(tctx.master.commands, mitmproxy.types.CutSpec, "foo") is False
        assert b.is_valid(tctx.master.commands, mitmproxy.types.CutSpec, "request.path") is True


            

Reported by Pylint.

              import os
import shlex

import pytest
import pyperclip

from mitmproxy import exceptions
from mitmproxy.addons import export  # heh
from mitmproxy.test import tflow

            

Reported by Pylint.

              import shlex

import pytest
import pyperclip

from mitmproxy import exceptions
from mitmproxy.addons import export  # heh
from mitmproxy.test import tflow
from mitmproxy.test import tutils

            

Reported by Pylint.

              

class TestExportCurlCommand:
    def test_get(self, export_curl, get_request):
        result = """curl -H 'header: qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
        assert export_curl(get_request) == result

    def test_post(self, export_curl, post_request):
        post_request.request.content = b'nobinarysupport'

            

Reported by Pylint.

              

class TestExportCurlCommand:
    def test_get(self, export_curl, get_request):
        result = """curl -H 'header: qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
        assert export_curl(get_request) == result

    def test_post(self, export_curl, post_request):
        post_request.request.content = b'nobinarysupport'

            

Reported by Pylint.

                      result = """curl -H 'header: qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
        assert export_curl(get_request) == result

    def test_post(self, export_curl, post_request):
        post_request.request.content = b'nobinarysupport'
        result = "curl -X POST http://address:22/path -d nobinarysupport"
        assert export_curl(post_request) == result

    def test_fails_with_binary_data(self, export_curl, post_request):

            

Reported by Pylint.

                      result = """curl -H 'header: qvalue' 'http://address:22/path?a=foo&a=bar&b=baz'"""
        assert export_curl(get_request) == result

    def test_post(self, export_curl, post_request):
        post_request.request.content = b'nobinarysupport'
        result = "curl -X POST http://address:22/path -d nobinarysupport"
        assert export_curl(post_request) == result

    def test_fails_with_binary_data(self, export_curl, post_request):

            

Reported by Pylint.

                      result = "curl -X POST http://address:22/path -d nobinarysupport"
        assert export_curl(post_request) == result

    def test_fails_with_binary_data(self, export_curl, post_request):
        # shlex.quote doesn't support a bytes object
        # see https://github.com/python/cpython/pull/10871
        post_request.request.headers["Content-Type"] = "application/json; charset=utf-8"
        with pytest.raises(exceptions.CommandError):
            export_curl(post_request)

            

Reported by Pylint.

                      result = "curl -X POST http://address:22/path -d nobinarysupport"
        assert export_curl(post_request) == result

    def test_fails_with_binary_data(self, export_curl, post_request):
        # shlex.quote doesn't support a bytes object
        # see https://github.com/python/cpython/pull/10871
        post_request.request.headers["Content-Type"] = "application/json; charset=utf-8"
        with pytest.raises(exceptions.CommandError):
            export_curl(post_request)

            

Reported by Pylint.

                      with pytest.raises(exceptions.CommandError):
            export_curl(post_request)

    def test_patch(self, export_curl, patch_request):
        result = """curl -H 'header: qvalue' -X PATCH 'http://address:22/path?query=param' -d content"""
        assert export_curl(patch_request) == result

    def test_tcp(self, export_curl, tcp_flow):
        with pytest.raises(exceptions.CommandError):

            

Reported by Pylint.

                      with pytest.raises(exceptions.CommandError):
            export_curl(post_request)

    def test_patch(self, export_curl, patch_request):
        result = """curl -H 'header: qvalue' -X PATCH 'http://address:22/path?query=param' -d content"""
        assert export_curl(patch_request) == result

    def test_tcp(self, export_curl, tcp_flow):
        with pytest.raises(exceptions.CommandError):

            

Reported by Pylint.

              from cryptography import x509
from cryptography.x509 import NameOID

import pytest

from mitmproxy import certs
from ..conftest import skip_windows



            

Reported by Pylint.

              import pytest

from mitmproxy import certs
from ..conftest import skip_windows


# class TestDNTree:
#     def test_simple(self):
#         d = certs.DNTree()

            

Reported by Pylint.

              
        assert ca.default_ca.serial == ca2.default_ca.serial

    def test_create_no_common_name(self, tstore):
        assert tstore.get_cert(None, []).cert.cn is None

    def test_chain_file(self, tdata, tmp_path):
        cert = Path(tdata.path("mitmproxy/data/confdir/mitmproxy-ca.pem")).read_bytes()
        (tmp_path / "mitmproxy-ca.pem").write_bytes(cert)

            

Reported by Pylint.

                      ca = certs.CertStore.from_store(tmp_path, "mitmproxy", 2048)
        assert ca.default_chain_file == (tmp_path / "mitmproxy-ca.pem")

    def test_sans(self, tstore):
        c1 = tstore.get_cert("foo.com", ["*.bar.com"])
        tstore.get_cert("foo.bar.com", [])
        # assert c1 == c2
        c3 = tstore.get_cert("bar.com", [])
        assert not c1 == c3

            

Reported by Pylint.

                      c3 = tstore.get_cert("bar.com", [])
        assert not c1 == c3

    def test_sans_change(self, tstore):
        tstore.get_cert("foo.com", ["*.bar.com"])
        entry = tstore.get_cert("foo.bar.com", ["*.baz.com"])
        assert "*.baz.com" in entry.cert.altnames

    def test_expire(self, tstore):

            

Reported by Pylint.

                      entry = tstore.get_cert("foo.bar.com", ["*.baz.com"])
        assert "*.baz.com" in entry.cert.altnames

    def test_expire(self, tstore):
        tstore.STORE_CAP = 3
        tstore.get_cert("one.com", [])
        tstore.get_cert("two.com", [])
        tstore.get_cert("three.com", [])


            

Reported by Pylint.

                      filename = str(tmpdir.join("secret"))
        with certs.CertStore.umask_secret(), open(filename, "wb"):
            pass
        # TODO: How do we actually attempt to read that file as another user?
        assert os.stat(filename).st_mode & 0o77 == 0


class TestDummyCert:


            

Reported by Pylint.

              
class TestDummyCert:

    def test_with_ca(self, tstore):
        r = certs.dummy_cert(
            tstore.default_privatekey,
            tstore.default_ca._cert,
            "foo.com",
            ["one.com", "two.com", "*.three.com", "127.0.0.1"],

            

Reported by Pylint.

                  def test_with_ca(self, tstore):
        r = certs.dummy_cert(
            tstore.default_privatekey,
            tstore.default_ca._cert,
            "foo.com",
            ["one.com", "two.com", "*.three.com", "127.0.0.1"],
            "Foo Ltd."
        )
        assert r.cn == "foo.com"

            

Reported by Pylint.

              
        r = certs.dummy_cert(
            tstore.default_privatekey,
            tstore.default_ca._cert,
            None,
            [],
            None
        )
        assert r.cn is None

            

Reported by Pylint.

              import io
import typing

import pytest

import mitmproxy.types
from mitmproxy import command
from mitmproxy import exceptions
from mitmproxy import flow

            

Reported by Pylint.

                      return "ret " + foo

    @command.command("cmd2")
    def cmd2(self, foo: str) -> str:
        return 99

    @command.command("cmd3")
    def cmd3(self, foo: int) -> int:
        return foo

            

Reported by Pylint.

                      return foo

    @command.command("cmd4")
    def cmd4(self, a: int, b: str, c: mitmproxy.types.Path) -> str:
        return "ok"

    @command.command("subcommand")
    def subcommand(self, cmd: mitmproxy.types.Cmd, *args: mitmproxy.types.CmdArgs) -> str:
        return "ok"

            

Reported by Pylint.

                      return foo

    @command.command("cmd4")
    def cmd4(self, a: int, b: str, c: mitmproxy.types.Path) -> str:
        return "ok"

    @command.command("subcommand")
    def subcommand(self, cmd: mitmproxy.types.Cmd, *args: mitmproxy.types.CmdArgs) -> str:
        return "ok"

            

Reported by Pylint.

                      return foo

    @command.command("cmd4")
    def cmd4(self, a: int, b: str, c: mitmproxy.types.Path) -> str:
        return "ok"

    @command.command("subcommand")
    def subcommand(self, cmd: mitmproxy.types.Cmd, *args: mitmproxy.types.CmdArgs) -> str:
        return "ok"

            

Reported by Pylint.

                      return "ok"

    @command.command("subcommand")
    def subcommand(self, cmd: mitmproxy.types.Cmd, *args: mitmproxy.types.CmdArgs) -> str:
        return "ok"

    @command.command("empty")
    def empty(self) -> None:
        pass

            

Reported by Pylint.

                      return "ok"

    @command.command("subcommand")
    def subcommand(self, cmd: mitmproxy.types.Cmd, *args: mitmproxy.types.CmdArgs) -> str:
        return "ok"

    @command.command("empty")
    def empty(self) -> None:
        pass

            

Reported by Pylint.

                      pass

    @command.command("varargs")
    def varargs(self, one: str, *var: str) -> typing.Sequence[str]:
        return list(var)

    def choices(self) -> typing.Sequence[str]:
        return ["one", "two", "three"]


            

Reported by Pylint.

                      return ["one", "two", "three"]

    @command.argument("arg", type=mitmproxy.types.Choice("choices"))
    def choose(self, arg: str) -> typing.Sequence[str]:
        return ["one", "two", "three"]

    @command.command("path")
    def path(self, arg: mitmproxy.types.Path) -> None:
        pass

            

Reported by Pylint.

                  assert command.typename(mitmproxy.types.Cmd) == "cmd"

    with pytest.raises(exceptions.CommandError, match="missing type annotation"):
        command.typename(inspect._empty)
    with pytest.raises(exceptions.CommandError, match="unsupported type"):
        command.typename(None)


class DummyConsole:

            

Reported by Pylint.

              import os
from unittest import mock

import pytest

import tornado.testing  # noqa
from tornado import httpclient  # noqa
from tornado import websocket  # noqa


            

Reported by Pylint.

              
import pytest

import tornado.testing  # noqa
from tornado import httpclient  # noqa
from tornado import websocket  # noqa

from mitmproxy import options  # noqa
from mitmproxy.test import tflow  # noqa

            

Reported by Pylint.

              import pytest

import tornado.testing  # noqa
from tornado import httpclient  # noqa
from tornado import websocket  # noqa

from mitmproxy import options  # noqa
from mitmproxy.test import tflow  # noqa
from mitmproxy.tools.web import app  # noqa

            

Reported by Pylint.

              
import tornado.testing  # noqa
from tornado import httpclient  # noqa
from tornado import websocket  # noqa

from mitmproxy import options  # noqa
from mitmproxy.test import tflow  # noqa
from mitmproxy.tools.web import app  # noqa
from mitmproxy.tools.web import master as webmaster  # noqa

            

Reported by Pylint.

                      m.view.add([f])
        m.view.add([tflow.tflow(err=True)])
        m.log.info("test log")
        self.master = m
        self.view = m.view
        self.events = m.events
        webapp = app.Application(m, None)
        webapp.settings["xsrf_cookies"] = False
        return webapp

            

Reported by Pylint.

                      m.view.add([tflow.tflow(err=True)])
        m.log.info("test log")
        self.master = m
        self.view = m.view
        self.events = m.events
        webapp = app.Application(m, None)
        webapp.settings["xsrf_cookies"] = False
        return webapp


            

Reported by Pylint.

                      m.log.info("test log")
        self.master = m
        self.view = m.view
        self.events = m.events
        webapp = app.Application(m, None)
        webapp.settings["xsrf_cookies"] = False
        return webapp

    def fetch(self, *args, **kwargs) -> httpclient.HTTPResponse:

            

Reported by Pylint.

                      f.backup()
        f.request.method = "PATCH"
        self.fetch("/flows/42/revert", method="POST")
        assert not f._backup

    def test_flow_replay(self):
        with mock.patch("mitmproxy.command.CommandManager.call") as replay_call:
            assert self.fetch("/flows/42/replay", method="POST").code == 200
            assert replay_call.called

            

Reported by Pylint.

              import asyncio
import json as _json
import logging
import os
from unittest import mock

import pytest

import tornado.testing  # noqa

            

Reported by Pylint.

              

@pytest.fixture(scope="module")
def no_tornado_logging():
    logging.getLogger('tornado.access').disabled = True
    logging.getLogger('tornado.application').disabled = True
    logging.getLogger('tornado.general').disabled = True
    yield
    logging.getLogger('tornado.access').disabled = False

            

Reported by Pylint.

              from dataclasses import dataclass
from typing import DefaultDict, Dict, List, Optional, Tuple, Union

import wsproto.handshake
from mitmproxy import flow, http
from mitmproxy.connection import Connection, Server
from mitmproxy.net import server_spec
from mitmproxy.net.http import status_codes, url
from mitmproxy.net.http.http1 import expected_http_body_size

            

Reported by Pylint.

              from mitmproxy.proxy.utils import expect
from mitmproxy.utils import human
from mitmproxy.websocket import WebSocketData
from ._base import HttpCommand, HttpConnection, ReceiveHttp, StreamId
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, RequestTrailers, \
    ResponseData, ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError, ResponseTrailers
from ._hooks import HttpConnectHook, HttpErrorHook, HttpRequestHeadersHook, HttpRequestHook, HttpResponseHeadersHook, \
    HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server

            

Reported by Pylint.

              from mitmproxy.utils import human
from mitmproxy.websocket import WebSocketData
from ._base import HttpCommand, HttpConnection, ReceiveHttp, StreamId
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, RequestTrailers, \
    ResponseData, ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError, ResponseTrailers
from ._hooks import HttpConnectHook, HttpErrorHook, HttpRequestHeadersHook, HttpRequestHook, HttpResponseHeadersHook, \
    HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server
from ._http2 import Http2Client, Http2Server

            

Reported by Pylint.

              from ._base import HttpCommand, HttpConnection, ReceiveHttp, StreamId
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, RequestTrailers, \
    ResponseData, ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError, ResponseTrailers
from ._hooks import HttpConnectHook, HttpErrorHook, HttpRequestHeadersHook, HttpRequestHook, HttpResponseHeadersHook, \
    HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server
from ._http2 import Http2Client, Http2Server
from ...context import Context


            

Reported by Pylint.

                  ResponseData, ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError, ResponseTrailers
from ._hooks import HttpConnectHook, HttpErrorHook, HttpRequestHeadersHook, HttpRequestHook, HttpResponseHeadersHook, \
    HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server
from ._http2 import Http2Client, Http2Server
from ...context import Context


class HTTPMode(enum.Enum):

            

Reported by Pylint.

              from ._hooks import HttpConnectHook, HttpErrorHook, HttpRequestHeadersHook, HttpRequestHook, HttpResponseHeadersHook, \
    HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server
from ._http2 import Http2Client, Http2Server
from ...context import Context


class HTTPMode(enum.Enum):
    regular = 1

            

Reported by Pylint.

                  HttpResponseHook, HttpConnectUpstreamHook  # noqa
from ._http1 import Http1Client, Http1Connection, Http1Server
from ._http2 import Http2Client, Http2Server
from ...context import Context


class HTTPMode(enum.Enum):
    regular = 1
    transparent = 2

            

Reported by Pylint.

                      return f"Invalid request scheme: {request.scheme}"
    if mode is HTTPMode.transparent and request.method == "CONNECT":
        return (
            f"mitmproxy received an HTTP CONNECT request even though it is not running in regular/upstream mode. "
            f"This usually indicates a misconfiguration, please see the mitmproxy mode documentation for details."
        )
    return None



            

Reported by Pylint.

                      self.stream_id = stream_id

    def __repr__(self):
        if self._handle_event == self.passthrough:
            return f"HttpStream(id={self.stream_id}, passthrough)"
        else:
            return (
                f"HttpStream("
                f"id={self.stream_id}, "

            

Reported by Pylint.

                          elif self.context.options.rawtcp:
                self.child_layer = tcp.TCPLayer(self.context)
            else:
                yield commands.Log(f"Sent HTTP 101 response, but no protocol is enabled to upgrade to.", "warn")
                yield commands.CloseConnection(self.context.client)
                self.client_state = self.server_state = self.state_errored
                return
            if self.debug:
                yield commands.Log(f"{self.debug}[http] upgrading to {self.child_layer}", "debug")

            

Reported by Pylint.

              from unittest.mock import patch
from pathlib import Path

import pytest

from mitmproxy.addons import command_history
from mitmproxy.test import taddons



            

Reported by Pylint.

                          ch.clear_history()

    @pytest.mark.asyncio
    async def test_clear_failed(self, monkeypatch):
        ch = command_history.CommandHistory()

        with taddons.context(ch) as tctx:
            tctx.options.confdir = '/non/existent/path/foobar1234/'


            

Reported by Pylint.

              import os
from unittest.mock import patch
from pathlib import Path

import pytest

from mitmproxy.addons import command_history
from mitmproxy.test import taddons


            

Reported by Pylint.

              from mitmproxy.test import taddons


class TestCommandHistory:
    def test_load_and_save(self, tmpdir):
        history_file = tmpdir.join('command_history')
        commands = ["cmd1", "cmd2", "cmd3"]
        with open(history_file, 'w') as f:
            f.write("\n".join(commands))

            

Reported by Pylint.

              

class TestCommandHistory:
    def test_load_and_save(self, tmpdir):
        history_file = tmpdir.join('command_history')
        commands = ["cmd1", "cmd2", "cmd3"]
        with open(history_file, 'w') as f:
            f.write("\n".join(commands))


            

Reported by Pylint.

              

class TestCommandHistory:
    def test_load_and_save(self, tmpdir):
        history_file = tmpdir.join('command_history')
        commands = ["cmd1", "cmd2", "cmd3"]
        with open(history_file, 'w') as f:
            f.write("\n".join(commands))


            

Reported by Pylint.

                  def test_load_and_save(self, tmpdir):
        history_file = tmpdir.join('command_history')
        commands = ["cmd1", "cmd2", "cmd3"]
        with open(history_file, 'w') as f:
            f.write("\n".join(commands))

        ch = command_history.CommandHistory()
        ch.VACUUM_SIZE = 4
        with taddons.context(ch) as tctx:

            

Reported by Pylint.

                      with open(history_file, 'w') as f:
            f.write("\n".join(commands))

        ch = command_history.CommandHistory()
        ch.VACUUM_SIZE = 4
        with taddons.context(ch) as tctx:
            tctx.options.confdir = str(tmpdir)
            assert ch.history == commands
            ch.add_command("cmd4")

            

Reported by Pylint.

                      ch.VACUUM_SIZE = 4
        with taddons.context(ch) as tctx:
            tctx.options.confdir = str(tmpdir)
            assert ch.history == commands
            ch.add_command("cmd4")
            ch.done()

        with open(history_file) as f:
            assert f.read() == "cmd3\ncmd4\n"

            

Reported by Bandit.

                          ch.add_command("cmd4")
            ch.done()

        with open(history_file) as f:
            assert f.read() == "cmd3\ncmd4\n"

    @pytest.mark.asyncio
    async def test_done_writing_failed(self):
        ch = command_history.CommandHistory()

            

Reported by Pylint.