import json

from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.net.http import cookies


class TestHARDump:

            

Reported by Pylint.

              import json

from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.net.http import cookies


class TestHARDump:

            

Reported by Pylint.

              
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.net.http import cookies


class TestHARDump:
    def flow(self, resp_content=b'message'):

            

Reported by Pylint.

              from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.net.http import cookies


class TestHARDump:
    def flow(self, resp_content=b'message'):
        times = dict(

            

Reported by Pylint.

              import json

from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.net.http import cookies


class TestHARDump:

            

Reported by Pylint.

              from mitmproxy.net.http import cookies


class TestHARDump:
    def flow(self, resp_content=b'message'):
        times = dict(
            timestamp_start=746203272,
            timestamp_end=746203272,
        )

            

Reported by Pylint.

              

class TestHARDump:
    def flow(self, resp_content=b'message'):
        times = dict(
            timestamp_start=746203272,
            timestamp_end=746203272,
        )


            

Reported by Pylint.

              

class TestHARDump:
    def flow(self, resp_content=b'message'):
        times = dict(
            timestamp_start=746203272,
            timestamp_end=746203272,
        )


            

Reported by Pylint.

                          resp=tutils.tresp(content=resp_content, **times)
        )

    def test_simple(self, tmpdir, tdata):
        # context is needed to provide ctx.log function that
        # is invoked if there are exceptions
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/contrib/har_dump.py"))
            # check script is read without errors

            

Reported by Pylint.

                      # context is needed to provide ctx.log function that
        # is invoked if there are exceptions
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/contrib/har_dump.py"))
            # check script is read without errors
            assert tctx.master.logs == []
            assert a.name_value   # last function in har_dump.py

            path = str(tmpdir.join("somefile"))

            

Reported by Pylint.

              from mitmproxy.proxy.layers.tcp import TcpMessageInjected
from mitmproxy.proxy.layers.websocket import WebSocketMessageInjected
from mitmproxy.utils import asyncio_utils, human
from wsproto.frame_protocol import Opcode


class AsyncReply(controller.Reply):
    """
    controller.Reply.q.get() is blocking, which we definitely want to avoid in a coroutine.

            

Reported by Pylint.

                  def commit(self):
        super().commit()
        try:
            self.loop.call_soon_threadsafe(lambda: self.done.set())
        except RuntimeError:  # pragma: no cover
            pass  # event loop may already be closed.

    def kill(self, force=False):  # pragma: no cover
        warnings.warn("reply.kill() is deprecated, set the error attribute instead.", DeprecationWarning, stacklevel=2)

            

Reported by Pylint.

              class ProxyConnectionHandler(server.StreamConnectionHandler):
    master: master.Master

    def __init__(self, master, r, w, options):
        self.master = master
        super().__init__(r, w, options)
        self.log_prefix = f"{human.format_address(self.client.peername)}: "

    async def handle_hook(self, hook: commands.StartHook) -> None:

            

Reported by Pylint.

              class ProxyConnectionHandler(server.StreamConnectionHandler):
    master: master.Master

    def __init__(self, master, r, w, options):
        self.master = master
        super().__init__(r, w, options)
        self.log_prefix = f"{human.format_address(self.client.peername)}: "

    async def handle_hook(self, hook: commands.StartHook) -> None:

            

Reported by Pylint.

                          try:
                human.parse_size(ctx.options.stream_large_bodies)
            except ValueError:
                raise exceptions.OptionsError(f"Invalid stream_large_bodies specification: "
                                              f"{ctx.options.stream_large_bodies}")
        if "body_size_limit" in updated:
            try:
                human.parse_size(ctx.options.body_size_limit)
            except ValueError:

            

Reported by Pylint.

                          try:
                human.parse_size(ctx.options.body_size_limit)
            except ValueError:
                raise exceptions.OptionsError(f"Invalid body_size_limit specification: "
                                              f"{ctx.options.body_size_limit}")
        if not self.is_running:
            return
        if "mode" in updated and ctx.options.mode == "transparent":  # pragma: no cover
            platform.init_transparent_mode()

            

Reported by Pylint.

                      peername = w.get_extra_info('peername')
        asyncio_utils.set_task_debug_info(
            asyncio.current_task(),
            name=f"Proxyserver.handle_connection",
            client=peername,
        )
        handler = ProxyConnectionHandler(
            self.master,
            r,

            

Reported by Pylint.

                      self._connections[event.flow.client_conn.peername].server_event(event)

    @command.command("inject.websocket")
    def inject_websocket(self, flow: Flow, to_client: bool, message: bytes, is_text: bool = True):
        if not isinstance(flow, http.HTTPFlow) or not flow.websocket:
            ctx.log.warn("Cannot inject WebSocket messages into non-WebSocket flows.")

        msg = websocket.WebSocketMessage(
            Opcode.TEXT if is_text else Opcode.BINARY,

            

Reported by Pylint.

                          ctx.log.warn(str(e))

    @command.command("inject.tcp")
    def inject_tcp(self, flow: Flow, to_client: bool, message: bytes):
        if not isinstance(flow, tcp.TCPFlow):
            ctx.log.warn("Cannot inject TCP messages into non-TCP flows.")

        event = TcpMessageInjected(flow, tcp.TCPMessage(not to_client, message))
        try:

            

Reported by Pylint.

                      except ValueError as e:
            ctx.log.warn(str(e))

    def server_connect(self, ctx: server_hooks.ServerConnectionHookData):
        assert ctx.server.address
        self_connect = (
            ctx.server.address[1] == self.options.listen_port
            and
            ctx.server.address[0] in ("localhost", "127.0.0.1", "::1", self.options.listen_host)

            

Reported by Pylint.

              import pytest

from mitmproxy import eventsequence
from mitmproxy.proxy import layers
from mitmproxy.test import tflow


@pytest.mark.parametrize("resp, err", [
    (False, False),

            

Reported by Pylint.

              import pytest

from mitmproxy import eventsequence
from mitmproxy.proxy import layers
from mitmproxy.test import tflow


@pytest.mark.parametrize("resp, err", [
    (False, False),

            

Reported by Pylint.

                  (True, False),
    (False, True),
    (True, True),
])
def test_http_flow(resp, err):
    f = tflow.tflow(resp=resp, err=err)
    i = eventsequence.iterate(f)
    assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)

            

Reported by Pylint.

                  (True, True),
])
def test_http_flow(resp, err):
    f = tflow.tflow(resp=resp, err=err)
    i = eventsequence.iterate(f)
    assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)
    if resp:
        assert isinstance(next(i), layers.http.HttpResponseHeadersHook)

            

Reported by Pylint.

              def test_http_flow(resp, err):
    f = tflow.tflow(resp=resp, err=err)
    i = eventsequence.iterate(f)
    assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)
    if resp:
        assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
        assert isinstance(next(i), layers.http.HttpResponseHook)
    if err:

            

Reported by Bandit.

                  f = tflow.tflow(resp=resp, err=err)
    i = eventsequence.iterate(f)
    assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)
    if resp:
        assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
        assert isinstance(next(i), layers.http.HttpResponseHook)
    if err:
        assert isinstance(next(i), layers.http.HttpErrorHook)

            

Reported by Bandit.

                  assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)
    if resp:
        assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
        assert isinstance(next(i), layers.http.HttpResponseHook)
    if err:
        assert isinstance(next(i), layers.http.HttpErrorHook)



            

Reported by Bandit.

                  assert isinstance(next(i), layers.http.HttpRequestHook)
    if resp:
        assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
        assert isinstance(next(i), layers.http.HttpResponseHook)
    if err:
        assert isinstance(next(i), layers.http.HttpErrorHook)


def test_websocket_flow():

            

Reported by Bandit.

                      assert isinstance(next(i), layers.http.HttpResponseHeadersHook)
        assert isinstance(next(i), layers.http.HttpResponseHook)
    if err:
        assert isinstance(next(i), layers.http.HttpErrorHook)


def test_websocket_flow():
    f = tflow.twebsocketflow()
    i = eventsequence.iterate(f)

            

Reported by Bandit.

                      assert isinstance(next(i), layers.http.HttpErrorHook)


def test_websocket_flow():
    f = tflow.twebsocketflow()
    i = eventsequence.iterate(f)

    assert isinstance(next(i), layers.http.HttpRequestHeadersHook)
    assert isinstance(next(i), layers.http.HttpRequestHook)

            

Reported by Pylint.

              import json
from unittest import mock
import pytest

from mitmproxy.test import taddons
from mitmproxy.test import tflow

from mitmproxy import flowfilter
from mitmproxy.tools.web.app import flow_to_json

            

Reported by Pylint.

              

@mock.patch('mitmproxy.ctx.log')
def test_save_flows_content(ctx, tmpdir):
    flows = [tflow.tflow(req=True, resp=None), tflow.tflow(req=True, resp=True)]
    with mock.patch('time.time', mock.Mock(side_effect=[1, 2, 2] * 4)):
        static_viewer.save_flows_content(tmpdir, flows)
    flows_path = tmpdir.join('flows')
    assert len(flows_path.listdir()) == len(flows)

            

Reported by Pylint.

              import json
from unittest import mock
import pytest

from mitmproxy.test import taddons
from mitmproxy.test import tflow

from mitmproxy import flowfilter
from mitmproxy.tools.web.app import flow_to_json

            

Reported by Pylint.

              from mitmproxy.addons import save, readfile


def test_save_static(tmpdir):
    tmpdir.mkdir('static')
    static_viewer.save_static(tmpdir)
    assert len(tmpdir.listdir()) == 2
    assert tmpdir.join('index.html').check(file=1)
    assert tmpdir.join('static/static.js').read() == 'MITMWEB_STATIC = true;'

            

Reported by Pylint.

              def test_save_static(tmpdir):
    tmpdir.mkdir('static')
    static_viewer.save_static(tmpdir)
    assert len(tmpdir.listdir()) == 2
    assert tmpdir.join('index.html').check(file=1)
    assert tmpdir.join('static/static.js').read() == 'MITMWEB_STATIC = true;'


def test_save_filter_help(tmpdir):

            

Reported by Bandit.

                  tmpdir.mkdir('static')
    static_viewer.save_static(tmpdir)
    assert len(tmpdir.listdir()) == 2
    assert tmpdir.join('index.html').check(file=1)
    assert tmpdir.join('static/static.js').read() == 'MITMWEB_STATIC = true;'


def test_save_filter_help(tmpdir):
    static_viewer.save_filter_help(tmpdir)

            

Reported by Bandit.

                  static_viewer.save_static(tmpdir)
    assert len(tmpdir.listdir()) == 2
    assert tmpdir.join('index.html').check(file=1)
    assert tmpdir.join('static/static.js').read() == 'MITMWEB_STATIC = true;'


def test_save_filter_help(tmpdir):
    static_viewer.save_filter_help(tmpdir)
    f = tmpdir.join('/filter-help.json')

            

Reported by Bandit.

                  assert tmpdir.join('static/static.js').read() == 'MITMWEB_STATIC = true;'


def test_save_filter_help(tmpdir):
    static_viewer.save_filter_help(tmpdir)
    f = tmpdir.join('/filter-help.json')
    assert f.check(file=1)
    assert f.read() == json.dumps(dict(commands=flowfilter.help))


            

Reported by Pylint.

              
def test_save_filter_help(tmpdir):
    static_viewer.save_filter_help(tmpdir)
    f = tmpdir.join('/filter-help.json')
    assert f.check(file=1)
    assert f.read() == json.dumps(dict(commands=flowfilter.help))


def test_save_settings(tmpdir):

            

Reported by Pylint.

              def test_save_filter_help(tmpdir):
    static_viewer.save_filter_help(tmpdir)
    f = tmpdir.join('/filter-help.json')
    assert f.check(file=1)
    assert f.read() == json.dumps(dict(commands=flowfilter.help))


def test_save_settings(tmpdir):
    static_viewer.save_settings(tmpdir)

            

Reported by Bandit.

              import urwid
import blinker
import textwrap

from mitmproxy import command
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import signals

HELP_HEIGHT = 5

            

Reported by Pylint.

                  def selectable(self):
        return True

    def keypress(self, size, key):
        return key


class CommandListWalker(urwid.ListWalker):
    def __init__(self, master):

            

Reported by Pylint.

                  def set_focus(self, index):
        cmd = self.cmds[index]
        self.index = index
        self.focus_obj = self._get(self.index)
        command_focus_change.send(cmd.help or "")

    def get_next(self, pos):
        if pos >= len(self.cmds) - 1:
            return None, None

            

Reported by Pylint.

              
    def keypress(self, size, key):
        if key == "m_select":
            foc, idx = self.get_focus()
            signals.status_prompt_command.send(partial=foc.cmd.name + " ")
        elif key == "m_start":
            self.set_focus(0)
            self.walker._modified()
        elif key == "m_end":

            

Reported by Pylint.

                          signals.status_prompt_command.send(partial=foc.cmd.name + " ")
        elif key == "m_start":
            self.set_focus(0)
            self.walker._modified()
        elif key == "m_end":
            self.set_focus(len(self.walker.cmds) - 1)
            self.walker._modified()
        return super().keypress(size, key)


            

Reported by Pylint.

                          self.walker._modified()
        elif key == "m_end":
            self.set_focus(len(self.walker.cmds) - 1)
            self.walker._modified()
        return super().keypress(size, key)


class CommandHelp(urwid.Frame):
    def __init__(self, master):

            

Reported by Pylint.

              
    def keypress(self, size, key):
        if key == "m_next":
            self.focus_position = (
                self.focus_position + 1
            ) % len(self.widget_list)
            self.widget_list[1].set_active(self.focus_position == 1)
            key = None


            

Reported by Pylint.

              import urwid
import blinker
import textwrap

from mitmproxy import command
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import signals

HELP_HEIGHT = 5

            

Reported by Pylint.

              import urwid
import blinker
import textwrap

from mitmproxy import command
from mitmproxy.tools.console import layoutwidget
from mitmproxy.tools.console import signals

HELP_HEIGHT = 5

            

Reported by Pylint.

              command_focus_change = blinker.Signal()


class CommandItem(urwid.WidgetWrap):
    def __init__(self, walker, cmd: command.Command, focused: bool):
        self.walker, self.cmd, self.focused = walker, cmd, focused
        super().__init__(None)
        self._w = self.get_widget()


            

Reported by Pylint.

              import pytest

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


class TestModifyBody:
    def test_configure(self):

            

Reported by Pylint.

              import pytest

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


class TestModifyBody:
    def test_configure(self):

            

Reported by Pylint.

              from mitmproxy.test import tflow


class TestModifyBody:
    def test_configure(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(mb, modify_body=["one/two/three"])
            with pytest.raises(Exception, match="Cannot parse modify_body"):

            

Reported by Pylint.

              

class TestModifyBody:
    def test_configure(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(mb, modify_body=["one/two/three"])
            with pytest.raises(Exception, match="Cannot parse modify_body"):
                tctx.configure(mb, modify_body=["/"])

            

Reported by Pylint.

              

class TestModifyBody:
    def test_configure(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(mb, modify_body=["one/two/three"])
            with pytest.raises(Exception, match="Cannot parse modify_body"):
                tctx.configure(mb, modify_body=["/"])

            

Reported by Pylint.

              
class TestModifyBody:
    def test_configure(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(mb, modify_body=["one/two/three"])
            with pytest.raises(Exception, match="Cannot parse modify_body"):
                tctx.configure(mb, modify_body=["/"])


            

Reported by Pylint.

                          with pytest.raises(Exception, match="Cannot parse modify_body"):
                tctx.configure(mb, modify_body=["/"])

    def test_simple(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(
                mb,
                modify_body=[

            

Reported by Pylint.

                          with pytest.raises(Exception, match="Cannot parse modify_body"):
                tctx.configure(mb, modify_body=["/"])

    def test_simple(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(
                mb,
                modify_body=[

            

Reported by Pylint.

                              tctx.configure(mb, modify_body=["/"])

    def test_simple(self):
        mb = modifybody.ModifyBody()
        with taddons.context(mb) as tctx:
            tctx.configure(
                mb,
                modify_body=[
                    "/~q/foo/bar",

            

Reported by Pylint.

                                  "/~s/foo/bar",
                ]
            )
            f = tflow.tflow()
            f.request.content = b"foo"
            mb.request(f)
            assert f.request.content == b"bar"

            f = tflow.tflow(resp=True)

            

Reported by Pylint.

              #!/usr/bin/env python3
'''
@author: David Shaw, shawd@vmware.com

Inspired by EAS Inspector for Fiddler
https://easinspectorforfiddler.codeplex.com

----- The MIT License (MIT) ----- 
Filename: ASWBXMLCodePage.py

            

Reported by Pylint.

              Inspired by EAS Inspector for Fiddler
https://easinspectorforfiddler.codeplex.com

----- The MIT License (MIT) ----- 
Filename: ASWBXMLCodePage.py
Copyright (c) 2014, David P. Shaw

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

            

Reported by Pylint.

              OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
THE SOFTWARE.
'''
class ASWBXMLCodePage:
	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}

            

Reported by Pylint.

              THE SOFTWARE.
'''
class ASWBXMLCodePage:
	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	

            

Reported by Pylint.

              '''
class ASWBXMLCodePage:
	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):

            

Reported by Pylint.

              class ASWBXMLCodePage:
	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):
		self.tokenLookup[token] = tag

            

Reported by Pylint.

              	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):
		self.tokenLookup[token] = tag
		self.tagLookup[tag] = token

            

Reported by Pylint.

              	def __init__(self):
		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):
		self.tokenLookup[token] = tag
		self.tagLookup[tag] = token

            

Reported by Pylint.

              		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):
		self.tokenLookup[token] = tag
		self.tagLookup[tag] = token
	

            

Reported by Pylint.

              		self.namespace = ""
		self.xmlns = ""
		self.tokenLookup = {}
		self.tagLookup = {}
	
	def addToken(self, token, tag):
		self.tokenLookup[token] = tag
		self.tagLookup[tag] = token
	

            

Reported by Pylint.

              from mitmproxy.test import taddons
from mitmproxy.http import Headers

from ..mitmproxy import tservers


class TestScripts(tservers.MasterTest):
    def test_add_header(self, tdata):
        with taddons.context() as tctx:

            

Reported by Pylint.

              from mitmproxy import contentviews
from mitmproxy.test import tflow
from mitmproxy.test import tutils
from mitmproxy.test import taddons
from mitmproxy.http import Headers

from ..mitmproxy import tservers



            

Reported by Pylint.

              from ..mitmproxy import tservers


class TestScripts(tservers.MasterTest):
    def test_add_header(self, tdata):
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)

            

Reported by Pylint.

              

class TestScripts(tservers.MasterTest):
    def test_add_header(self, tdata):
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)
            assert f.request.headers["myheader"] == "value"

            

Reported by Pylint.

              

class TestScripts(tservers.MasterTest):
    def test_add_header(self, tdata):
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)
            assert f.request.headers["myheader"] == "value"

            

Reported by Pylint.

              class TestScripts(tservers.MasterTest):
    def test_add_header(self, tdata):
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)
            assert f.request.headers["myheader"] == "value"

    def test_custom_contentviews(self, tdata):

            

Reported by Pylint.

                  def test_add_header(self, tdata):
        with taddons.context() as tctx:
            a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)
            assert f.request.headers["myheader"] == "value"

    def test_custom_contentviews(self, tdata):
        with taddons.context() as tctx:

            

Reported by Pylint.

                          a = tctx.script(tdata.path("../examples/addons/anatomy2.py"))
            f = tflow.tflow()
            a.request(f)
            assert f.request.headers["myheader"] == "value"

    def test_custom_contentviews(self, tdata):
        with taddons.context() as tctx:
            tctx.script(tdata.path("../examples/addons/contentview.py"))
            swapcase = contentviews.get("swapcase")

            

Reported by Bandit.

                          a.request(f)
            assert f.request.headers["myheader"] == "value"

    def test_custom_contentviews(self, tdata):
        with taddons.context() as tctx:
            tctx.script(tdata.path("../examples/addons/contentview.py"))
            swapcase = contentviews.get("swapcase")
            _, fmt = swapcase(b"<html>Test!</html>")
            assert any(b'tEST!' in val[0][1] for val in fmt)

            

Reported by Pylint.

                          a.request(f)
            assert f.request.headers["myheader"] == "value"

    def test_custom_contentviews(self, tdata):
        with taddons.context() as tctx:
            tctx.script(tdata.path("../examples/addons/contentview.py"))
            swapcase = contentviews.get("swapcase")
            _, fmt = swapcase(b"<html>Test!</html>")
            assert any(b'tEST!' in val[0][1] for val in fmt)

            

Reported by Pylint.

              from typing import Optional
from typing import Tuple

import ldap3
import passlib.apache

from mitmproxy import ctx, connection
from mitmproxy import exceptions
from mitmproxy import http

            

Reported by Pylint.

              from typing import Tuple

import ldap3
import passlib.apache

from mitmproxy import ctx, connection
from mitmproxy import exceptions
from mitmproxy import http
from mitmproxy.net.http import status_codes

            

Reported by Pylint.

                  try:
        user, password = binascii.a2b_base64(authinfo.encode()).decode("utf8", "replace").split(":")
    except binascii.Error as e:
        raise ValueError(str(e))
    return scheme, user, password


class ProxyAuth:
    def __init__(self):

            

Reported by Pylint.

                      """
        auth_value = f.request.headers.get(self.which_auth_header(), "")
        try:
            scheme, username, password = parse_http_basic_auth(auth_value)
        except ValueError:
            return None

        if self.nonanonymous:
            return username, password

            

Reported by Pylint.

                                  try:
                        self.htpasswd = passlib.apache.HtpasswdFile(p)
                    except (ValueError, OSError):
                        raise exceptions.OptionsError(
                            "Could not open htpasswd file: %s" % p
                        )
                elif ctx.options.proxyauth.startswith("ldap"):
                    parts = ctx.options.proxyauth.split(':')
                    if len(parts) != 5:

            

Reported by Pylint.

                                  "Proxy Authentication not supported in SOCKS mode. "
                    "https://github.com/mitmproxy/mitmproxy/issues/738"
                )
                # TODO: check for multiple auth options

    def http_connect(self, f: http.HTTPFlow) -> None:
        if self.enabled():
            if self.authenticate(f):
                self.authenticated[f.client_conn] = f.metadata["proxyauth"]

            

Reported by Pylint.

              import binascii
import weakref
from typing import MutableMapping
from typing import Optional
from typing import Tuple

import ldap3
import passlib.apache


            

Reported by Pylint.

                  """
    Craft a basic auth string
    """
    v = binascii.b2a_base64(
        (username + ":" + password).encode("utf8")
    ).decode("ascii")
    return scheme + " " + v



            

Reported by Pylint.

                  return scheme + " " + v


def parse_http_basic_auth(s: str) -> Tuple[str, str, str]:
    """
    Parse a basic auth header.
    Raises a ValueError if the input is invalid.
    """
    scheme, authinfo = s.split()

            

Reported by Pylint.

                      raise ValueError("Unknown scheme")
    try:
        user, password = binascii.a2b_base64(authinfo.encode()).decode("utf8", "replace").split(":")
    except binascii.Error as e:
        raise ValueError(str(e))
    return scheme, user, password


class ProxyAuth:

            

Reported by Pylint.

              import asyncio
import io

import pytest
from unittest import mock

import mitmproxy.io
from mitmproxy import exceptions
from mitmproxy.addons import readfile

            

Reported by Pylint.

              

@pytest.fixture
def corrupt_data(data):
    f = io.BytesIO(data.getvalue())
    f.seek(0, io.SEEK_END)
    f.write(b"qibble")
    f.seek(0)
    return f

            

Reported by Pylint.

                              tctx.configure(rf, readfile_filter="~~")

    @pytest.mark.asyncio
    async def test_read(self, tmpdir, data, corrupt_data):
        rf = readfile.ReadFile()
        with taddons.context(rf) as tctx:
            assert not rf.reading()

            tf = tmpdir.join("tfile")

            

Reported by Pylint.

                              tctx.configure(rf, readfile_filter="~~")

    @pytest.mark.asyncio
    async def test_read(self, tmpdir, data, corrupt_data):
        rf = readfile.ReadFile()
        with taddons.context(rf) as tctx:
            assert not rf.reading()

            tf = tmpdir.join("tfile")

            

Reported by Pylint.

                          await tctx.master.await_log("corrupted")

    @pytest.mark.asyncio
    async def test_corrupt(self, corrupt_data):
        rf = readfile.ReadFile()
        with taddons.context(rf) as tctx:
            with pytest.raises(exceptions.FlowReadException):
                await rf.load_flows(io.BytesIO(b"qibble"))


            

Reported by Pylint.

              class TestReadFileStdin:
    @mock.patch('sys.stdin')
    @pytest.mark.asyncio
    async def test_stdin(self, stdin, data, corrupt_data):
        rf = readfile.ReadFileStdin()
        with taddons.context(rf):
            with mock.patch('mitmproxy.master.Master.load_flow') as mck:
                stdin.buffer = data
                mck.assert_not_awaited()

            

Reported by Pylint.

              class TestReadFileStdin:
    @mock.patch('sys.stdin')
    @pytest.mark.asyncio
    async def test_stdin(self, stdin, data, corrupt_data):
        rf = readfile.ReadFileStdin()
        with taddons.context(rf):
            with mock.patch('mitmproxy.master.Master.load_flow') as mck:
                stdin.buffer = data
                mck.assert_not_awaited()

            

Reported by Pylint.

                                  await rf.load_flows(stdin.buffer)

    @pytest.mark.asyncio
    async def test_normal(self, tmpdir, data):
        rf = readfile.ReadFileStdin()
        with taddons.context(rf) as tctx:
            tf = tmpdir.join("tfile")
            with mock.patch('mitmproxy.master.Master.load_flow') as mck:
                tf.write(data.getvalue())

            

Reported by Pylint.

              import asyncio
import io

import pytest
from unittest import mock

import mitmproxy.io
from mitmproxy import exceptions
from mitmproxy.addons import readfile

            

Reported by Pylint.

              import io

import pytest
from unittest import mock

import mitmproxy.io
from mitmproxy import exceptions
from mitmproxy.addons import readfile
from mitmproxy.test import taddons

            

Reported by Pylint.