import abc
from typing import Callable, Optional, Type, Union

import h11
from h11._readers import ChunkedReader, ContentLengthReader, Http10Reader
from h11._receivebuffer import ReceiveBuffer

from mitmproxy import http, version
from mitmproxy.connection import Connection, ConnectionState

            

Reported by Pylint.

              from typing import Callable, Optional, Type, Union

import h11
from h11._readers import ChunkedReader, ContentLengthReader, Http10Reader
from h11._receivebuffer import ReceiveBuffer

from mitmproxy import http, version
from mitmproxy.connection import Connection, ConnectionState
from mitmproxy.net.http import http1, status_codes

            

Reported by Pylint.

              
import h11
from h11._readers import ChunkedReader, ContentLengthReader, Http10Reader
from h11._receivebuffer import ReceiveBuffer

from mitmproxy import http, version
from mitmproxy.connection import Connection, ConnectionState
from mitmproxy.net.http import http1, status_codes
from mitmproxy.proxy import commands, events, layer

            

Reported by Pylint.

              from mitmproxy.proxy.layers.http._base import ReceiveHttp, StreamId
from mitmproxy.proxy.utils import expect
from mitmproxy.utils import human
from ._base import HttpConnection, format_error
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, ResponseData, \
    ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError
from ...context import Context

TBodyReader = Union[ChunkedReader, Http10Reader, ContentLengthReader]

            

Reported by Pylint.

              from mitmproxy.proxy.utils import expect
from mitmproxy.utils import human
from ._base import HttpConnection, format_error
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, ResponseData, \
    ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError
from ...context import Context

TBodyReader = Union[ChunkedReader, Http10Reader, ContentLengthReader]


            

Reported by Pylint.

              from ._base import HttpConnection, format_error
from ._events import HttpEvent, RequestData, RequestEndOfMessage, RequestHeaders, RequestProtocolError, ResponseData, \
    ResponseEndOfMessage, ResponseHeaders, ResponseProtocolError
from ...context import Context

TBodyReader = Union[ChunkedReader, Http10Reader, ContentLengthReader]


class Http1Connection(HttpConnection, metaclass=abc.ABCMeta):

            

Reported by Pylint.

                      if isinstance(event, HttpEvent):
            yield from self.send(event)
        else:
            if isinstance(event, events.DataReceived) and self.state != self.passthrough:
                self.buf += event.data
            yield from self.state(event)

    @expect(events.Start)
    def start(self, _) -> layer.CommandGenerator[None]:

            

Reported by Pylint.

                          elif isinstance(h11_event, h11.EndOfMessage):
                assert self.request
                if h11_event.headers:
                    raise NotImplementedError(f"HTTP trailers are not implemented yet.")
                if self.request.data.method.upper() != b"CONNECT":
                    yield ReceiveHttp(self.ReceiveEndOfMessage(self.stream_id))
                is_request = isinstance(self, Http1Server)
                yield from self.mark_done(
                    request=is_request,

            

Reported by Pylint.

                          # see https://github.com/httpwg/http-core/issues/22
            if event.connection.state is not ConnectionState.CLOSED:
                yield commands.CloseConnection(event.connection)
            yield ReceiveHttp(self.ReceiveProtocolError(self.stream_id, f"Client disconnected.",
                                                        code=status_codes.CLIENT_CLOSED_REQUEST))
        else:  # pragma: no cover
            raise AssertionError(f"Unexpected event: {event}")

    def done(self, event: events.ConnectionEvent) -> layer.CommandGenerator[None]:

            

Reported by Pylint.

                      else:  # pragma: no cover
            raise AssertionError(f"Unexpected event: {event}")

    def done(self, event: events.ConnectionEvent) -> layer.CommandGenerator[None]:
        yield from ()  # pragma: no cover

    def make_pipe(self) -> layer.CommandGenerator[None]:
        self.state = self.passthrough
        if self.buf:

            

Reported by Pylint.

              from mitmproxy import certs
from mitmproxy.test import taddons
from mitmproxy.test import tflow
import pytest
import pyperclip
from unittest import mock


def test_extract(tdata):

            

Reported by Pylint.

              from mitmproxy.test import taddons
from mitmproxy.test import tflow
import pytest
import pyperclip
from unittest import mock


def test_extract(tdata):
    tf = tflow.tflow(resp=True)

            

Reported by Pylint.

              from mitmproxy.addons import cut
from mitmproxy.addons import view
from mitmproxy import exceptions
from mitmproxy import certs
from mitmproxy.test import taddons
from mitmproxy.test import tflow
import pytest
import pyperclip
from unittest import mock

            

Reported by Pylint.

              from mitmproxy.test import tflow
import pytest
import pyperclip
from unittest import mock


def test_extract(tdata):
    tf = tflow.tflow(resp=True)
    tests = [

            

Reported by Pylint.

              from unittest import mock


def test_extract(tdata):
    tf = tflow.tflow(resp=True)
    tests = [
        ["request.method", "GET"],
        ["request.scheme", "http"],
        ["request.host", "address"],

            

Reported by Pylint.

              

def test_extract(tdata):
    tf = tflow.tflow(resp=True)
    tests = [
        ["request.method", "GET"],
        ["request.scheme", "http"],
        ["request.host", "address"],
        ["request.http_version", "HTTP/1.1"],

            

Reported by Pylint.

                  ]
    for spec, expected in tests:
        ret = cut.extract(spec, tf)
        assert spec and ret == expected

    with open(tdata.path("mitmproxy/net/data/text_cert"), "rb") as f:
        d = f.read()
    c1 = certs.Cert.from_pem(d)
    tf.server_conn.certificate_list = [c1]

            

Reported by Bandit.

                      ret = cut.extract(spec, tf)
        assert spec and ret == expected

    with open(tdata.path("mitmproxy/net/data/text_cert"), "rb") as f:
        d = f.read()
    c1 = certs.Cert.from_pem(d)
    tf.server_conn.certificate_list = [c1]
    assert "CERTIFICATE" in cut.extract("server_conn.certificate_list", tf)


            

Reported by Pylint.

                      assert spec and ret == expected

    with open(tdata.path("mitmproxy/net/data/text_cert"), "rb") as f:
        d = f.read()
    c1 = certs.Cert.from_pem(d)
    tf.server_conn.certificate_list = [c1]
    assert "CERTIFICATE" in cut.extract("server_conn.certificate_list", tf)



            

Reported by Pylint.

              
    with open(tdata.path("mitmproxy/net/data/text_cert"), "rb") as f:
        d = f.read()
    c1 = certs.Cert.from_pem(d)
    tf.server_conn.certificate_list = [c1]
    assert "CERTIFICATE" in cut.extract("server_conn.certificate_list", tf)


def test_headername():

            

Reported by Pylint.

              import os.path
from typing import Optional

import urwid

import mitmproxy.tools.console.master  # noqa
from mitmproxy.tools.console import commandexecutor
from mitmproxy.tools.console import common
from mitmproxy.tools.console import signals

            

Reported by Pylint.

              
import urwid

import mitmproxy.tools.console.master  # noqa
from mitmproxy.tools.console import commandexecutor
from mitmproxy.tools.console import common
from mitmproxy.tools.console import signals
from mitmproxy.tools.console.commander import commander


            

Reported by Pylint.

              
        self.onekey = False

    def sig_message(self, sender, message, expire=1):
        if self.prompting:
            return
        cols, _ = self.master.ui.get_cols_rows()
        w = urwid.Text(self.shorten_message(message, cols))
        self._w = w

            

Reported by Pylint.

                          return
        cols, _ = self.master.ui.get_cols_rows()
        w = urwid.Text(self.shorten_message(message, cols))
        self._w = w
        if expire:
            def cb(*args):
                if w == self._w:
                    self.clear()


            

Reported by Pylint.

                      w = urwid.Text(self.shorten_message(message, cols))
        self._w = w
        if expire:
            def cb(*args):
                if w == self._w:
                    self.clear()

            signals.call_in.send(seconds=expire, callback=cb)


            

Reported by Pylint.

              
        return [(disp_attr, first_line), ("warn", prompt)]

    def sig_prompt(self, sender, prompt, text, callback, args=()):
        signals.focus.send(self, section="footer")
        self._w = urwid.Edit(self.prep_prompt(prompt), text or "")
        self.prompting = PromptStub(callback, args)

    def sig_prompt_command(self, sender, partial: str = "", cursor: Optional[int] = None):

            

Reported by Pylint.

              
    def sig_prompt(self, sender, prompt, text, callback, args=()):
        signals.focus.send(self, section="footer")
        self._w = urwid.Edit(self.prep_prompt(prompt), text or "")
        self.prompting = PromptStub(callback, args)

    def sig_prompt_command(self, sender, partial: str = "", cursor: Optional[int] = None):
        signals.focus.send(self, section="footer")
        self._w = commander.CommandEdit(

            

Reported by Pylint.

                      self._w = urwid.Edit(self.prep_prompt(prompt), text or "")
        self.prompting = PromptStub(callback, args)

    def sig_prompt_command(self, sender, partial: str = "", cursor: Optional[int] = None):
        signals.focus.send(self, section="footer")
        self._w = commander.CommandEdit(
            self.master,
            partial,
        )

            

Reported by Pylint.

              
    def sig_prompt_command(self, sender, partial: str = "", cursor: Optional[int] = None):
        signals.focus.send(self, section="footer")
        self._w = commander.CommandEdit(
            self.master,
            partial,
        )
        if cursor is not None:
            self._w.cbuf.cursor = cursor

            

Reported by Pylint.

                      execute = commandexecutor.CommandExecutor(self.master)
        execute(txt)

    def sig_prompt_onekey(self, sender, prompt, keys, callback, args=()):
        """
            Keys are a set of (word, key) tuples. The appropriate key in the
            word is highlighted.
        """
        signals.focus.send(self, section="footer")

            

Reported by Pylint.

              import pytest

from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.context import Context
from test.mitmproxy.proxy import tutils


class TestLayer:
    def test_continue(self, tctx: Context):

            

Reported by Pylint.

                  def test_debug_shorten(self, tctx):
        t = layer.Layer(tctx)
        t.debug = "  "
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 512 + "…"
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 256 + "…"
        assert t._Layer__debug("foo").message == "  foo"


class TestNextLayer:

            

Reported by Pylint.

                      t = layer.Layer(tctx)
        t.debug = "  "
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 512 + "…"
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 256 + "…"
        assert t._Layer__debug("foo").message == "  foo"


class TestNextLayer:
    def test_simple(self, tctx: Context):

            

Reported by Pylint.

                      t.debug = "  "
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 512 + "…"
        assert t._Layer__debug("x" * 600).message == "  " + "x" * 256 + "…"
        assert t._Layer__debug("foo").message == "  foo"


class TestNextLayer:
    def test_simple(self, tctx: Context):
        nl = layer.NextLayer(tctx, ask_on_start=True)

            

Reported by Pylint.

              import pytest

from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.context import Context
from test.mitmproxy.proxy import tutils


class TestLayer:
    def test_continue(self, tctx: Context):

            

Reported by Pylint.

              
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.context import Context
from test.mitmproxy.proxy import tutils


class TestLayer:
    def test_continue(self, tctx: Context):
        class TLayer(layer.Layer):

            

Reported by Pylint.

              from test.mitmproxy.proxy import tutils


class TestLayer:
    def test_continue(self, tctx: Context):
        class TLayer(layer.Layer):
            def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
                yield commands.OpenConnection(self.context.server)
                yield commands.OpenConnection(self.context.server)

            

Reported by Pylint.

              

class TestLayer:
    def test_continue(self, tctx: Context):
        class TLayer(layer.Layer):
            def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
                yield commands.OpenConnection(self.context.server)
                yield commands.OpenConnection(self.context.server)


            

Reported by Pylint.

              
class TestLayer:
    def test_continue(self, tctx: Context):
        class TLayer(layer.Layer):
            def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
                yield commands.OpenConnection(self.context.server)
                yield commands.OpenConnection(self.context.server)

        assert (

            

Reported by Pylint.

                              yield commands.OpenConnection(self.context.server)
                yield commands.OpenConnection(self.context.server)

        assert (
            tutils.Playbook(TLayer(tctx))
            << commands.OpenConnection(tctx.server)
            >> tutils.reply(None)
            << commands.OpenConnection(tctx.server)
            >> tutils.reply(None)

            

Reported by Bandit.

              from mitmproxy.tools.console import keymap
from mitmproxy.test import taddons
from unittest import mock
import pytest


def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "

            

Reported by Pylint.

              from mitmproxy.tools.console import keymap
from mitmproxy.test import taddons
from unittest import mock
import pytest


def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "

            

Reported by Pylint.

              from mitmproxy.tools.console import keymap
from mitmproxy.test import taddons
from unittest import mock
import pytest


def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "

            

Reported by Pylint.

              import pytest


def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "


def test_bind():

            

Reported by Pylint.

              

def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "


def test_bind():
    with taddons.context() as tctx:

            

Reported by Pylint.

              
def test_binding():
    b = keymap.Binding("space", "cmd", ["options"], "")
    assert b.keyspec() == " "


def test_bind():
    with taddons.context() as tctx:
        km = keymap.Keymap(tctx.master)

            

Reported by Bandit.

                  assert b.keyspec() == " "


def test_bind():
    with taddons.context() as tctx:
        km = keymap.Keymap(tctx.master)
        km.executor = mock.Mock()

        with pytest.raises(ValueError):

            

Reported by Pylint.

              
def test_bind():
    with taddons.context() as tctx:
        km = keymap.Keymap(tctx.master)
        km.executor = mock.Mock()

        with pytest.raises(ValueError):
            km.add("foo", "bar", ["unsupported"])


            

Reported by Pylint.

                          km.add("foo", "bar", ["unsupported"])

        km.add("key", "str", ["options", "commands"])
        assert km.get("options", "key")
        assert km.get("commands", "key")
        assert not km.get("flowlist", "key")
        assert len(km.list("commands")) == 1

        km.handle("unknown", "unknown")

            

Reported by Bandit.

              
        km.add("key", "str", ["options", "commands"])
        assert km.get("options", "key")
        assert km.get("commands", "key")
        assert not km.get("flowlist", "key")
        assert len(km.list("commands")) == 1

        km.handle("unknown", "unknown")
        assert not km.executor.called

            

Reported by Bandit.

                      time.sleep(seconds)

    def run_external(self, command: str) -> None:
        subprocess.run(command, shell=True)

    def message(self, msg: str, duration: typing.Optional[int] = None, add_instruction: bool = True, instruction_html: str = "") -> None:
        if duration is None:
            duration = len(msg) * 0.08  # seconds
        self.tmux_session.set_option("display-time", int(duration * 1000))  # milliseconds

            

Reported by Bandit.

              import json
import libtmux
import random
import subprocess
import threading
import time
import typing



            

Reported by Pylint.

                      return self.tmux_session

    def start_session(self, width: int = 0, height: int = 0) -> libtmux.Session:
        self.tmux_server = libtmux.Server()
        self.tmux_session = self.tmux_server.new_session(session_name="asciinema_recorder", kill_session=True)
        self.tmux_pane = self.tmux_session.attached_window.attached_pane
        self.tmux_version = self.tmux_pane.display_message("#{version}", True)
        if width and height:
            self.resize_window(width, height)

            

Reported by Pylint.

              
    def start_session(self, width: int = 0, height: int = 0) -> libtmux.Session:
        self.tmux_server = libtmux.Server()
        self.tmux_session = self.tmux_server.new_session(session_name="asciinema_recorder", kill_session=True)
        self.tmux_pane = self.tmux_session.attached_window.attached_pane
        self.tmux_version = self.tmux_pane.display_message("#{version}", True)
        if width and height:
            self.resize_window(width, height)
        self.pause(3)

            

Reported by Pylint.

                  def start_session(self, width: int = 0, height: int = 0) -> libtmux.Session:
        self.tmux_server = libtmux.Server()
        self.tmux_session = self.tmux_server.new_session(session_name="asciinema_recorder", kill_session=True)
        self.tmux_pane = self.tmux_session.attached_window.attached_pane
        self.tmux_version = self.tmux_pane.display_message("#{version}", True)
        if width and height:
            self.resize_window(width, height)
        self.pause(3)
        return self.tmux_session

            

Reported by Pylint.

                      self.tmux_server = libtmux.Server()
        self.tmux_session = self.tmux_server.new_session(session_name="asciinema_recorder", kill_session=True)
        self.tmux_pane = self.tmux_session.attached_window.attached_pane
        self.tmux_version = self.tmux_pane.display_message("#{version}", True)
        if width and height:
            self.resize_window(width, height)
        self.pause(3)
        return self.tmux_session


            

Reported by Pylint.

                      return self.tmux_session

    def start_recording(self, filename: str) -> None:
        self.asciinema_proc = subprocess.Popen([
            "asciinema", "rec", "-y", "--overwrite", "-c", "tmux attach -t asciinema_recorder", filename])
        self.pause(1.5)
        self.record_start = time.time()

    def resize_window(self, width: int, height: int) -> None:

            

Reported by Pylint.

                          pause = self.pause_between_keys
        if target is None:
            target = self.tmux_pane
        for i in range(count):
            if keys == " ":
                keys = "Space"
            target.send_keys(cmd=keys, enter=False, suppress_history=False)

            # inspired by https://github.com/dmotz/TuringType

            

Reported by Pylint.

                  def focus_pane(self, pane: libtmux.Pane, set_active_pane: bool = True) -> None:
        pane.select_pane()
        if set_active_pane:
            self.tmux_pane = pane

    def pause(self, seconds: float) -> None:
        time.sleep(seconds)

    def run_external(self, command: str) -> None:

            

Reported by Pylint.

                      time.sleep(seconds)

    def run_external(self, command: str) -> None:
        subprocess.run(command, shell=True)

    def message(self, msg: str, duration: typing.Optional[int] = None, add_instruction: bool = True, instruction_html: str = "") -> None:
        if duration is None:
            duration = len(msg) * 0.08  # seconds
        self.tmux_session.set_option("display-time", int(duration * 1000))  # milliseconds

            

Reported by Pylint.

              import typing

import pytest

from mitmproxy.stateobject import StateObject


class TObject(StateObject):
    def __init__(self, x):

            

Reported by Pylint.

              import typing

import pytest

from mitmproxy.stateobject import StateObject


class TObject(StateObject):
    def __init__(self, x):

            

Reported by Pylint.

              from mitmproxy.stateobject import StateObject


class TObject(StateObject):
    def __init__(self, x):
        self.x = x

    @classmethod
    def from_state(cls, state):

            

Reported by Pylint.

              
class TObject(StateObject):
    def __init__(self, x):
        self.x = x

    @classmethod
    def from_state(cls, state):
        obj = cls(None)
        obj.set_state(state)

            

Reported by Pylint.

                      return obj


class Child(TObject):
    _stateobject_attributes = dict(
        x=int
    )

    def __eq__(self, other):

            

Reported by Pylint.

                      return isinstance(other, Child) and self.x == other.x


class TTuple(TObject):
    _stateobject_attributes = dict(
        x=typing.Tuple[int, Child]
    )



            

Reported by Pylint.

                  )


class TList(TObject):
    _stateobject_attributes = dict(
        x=typing.List[Child]
    )



            

Reported by Pylint.

                  )


class TDict(TObject):
    _stateobject_attributes = dict(
        x=typing.Dict[str, Child]
    )



            

Reported by Pylint.

                  )


class TAny(TObject):
    _stateobject_attributes = dict(
        x=typing.Any
    )



            

Reported by Pylint.

                  )


class TSerializableChild(TObject):
    _stateobject_attributes = dict(
        x=Child
    )



            

Reported by Pylint.

              from typing import Tuple, Optional

import pytest

from mitmproxy.proxy import tunnel, layer
from mitmproxy.proxy.commands import SendData, Log, CloseConnection, OpenConnection
from mitmproxy.connection import Server, ConnectionState
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import Event, DataReceived, Start, ConnectionClosed

            

Reported by Pylint.

                  assert repr(tl)

    playbook = Playbook(tl, logs=True)
    (
            playbook
            << SendData(server, b"handshake-hello")
            >> DataReceived(tctx.client, b"client-hello")
            >> DataReceived(server, b"handshake-" + success.encode())
            << SendData(server, b"handshake-" + success.encode())

            

Reported by Pylint.

                          << SendData(server, b"handshake-" + success.encode())
    )
    if success == "success":
        playbook << Log("Got start. Server state: OPEN")
    else:
        playbook << CloseConnection(server)
        playbook << Log("Got start. Server state: CLOSED")

    playbook << SendData(tctx.client, b"client-hello-reply")

            

Reported by Pylint.

                  if success == "success":
        playbook << Log("Got start. Server state: OPEN")
    else:
        playbook << CloseConnection(server)
        playbook << Log("Got start. Server state: CLOSED")

    playbook << SendData(tctx.client, b"client-hello-reply")
    if success == "success":
        playbook >> DataReceived(server, b"tunneled-server-hello")

            

Reported by Pylint.

                      playbook << Log("Got start. Server state: OPEN")
    else:
        playbook << CloseConnection(server)
        playbook << Log("Got start. Server state: CLOSED")

    playbook << SendData(tctx.client, b"client-hello-reply")
    if success == "success":
        playbook >> DataReceived(server, b"tunneled-server-hello")
        playbook << SendData(server, b"tunneled-server-hello-reply")

            

Reported by Pylint.

                      playbook << CloseConnection(server)
        playbook << Log("Got start. Server state: CLOSED")

    playbook << SendData(tctx.client, b"client-hello-reply")
    if success == "success":
        playbook >> DataReceived(server, b"tunneled-server-hello")
        playbook << SendData(server, b"tunneled-server-hello-reply")

    assert playbook

            

Reported by Pylint.

              
    playbook << SendData(tctx.client, b"client-hello-reply")
    if success == "success":
        playbook >> DataReceived(server, b"tunneled-server-hello")
        playbook << SendData(server, b"tunneled-server-hello-reply")

    assert playbook



            

Reported by Pylint.

                  playbook << SendData(tctx.client, b"client-hello-reply")
    if success == "success":
        playbook >> DataReceived(server, b"tunneled-server-hello")
        playbook << SendData(server, b"tunneled-server-hello-reply")

    assert playbook


@pytest.mark.parametrize("success", ["success", "fail"])

            

Reported by Pylint.

                  tl.child_layer = TChildLayer(tctx)

    playbook = Playbook(tl, logs=True)
    (
            playbook
            << Log("Got start. Server state: CLOSED")
            >> DataReceived(tctx.client, b"client-hello")
            << SendData(tctx.client, b"client-hello-reply")
            >> DataReceived(tctx.client, b"open")

            

Reported by Pylint.

                  if success == "success":
        assert (
                playbook
                << Log(f"Opened: err=None. Server state: OPEN")
                >> DataReceived(server, b"tunneled-server-hello")
                << SendData(server, b"tunneled-server-hello-reply")
                >> ConnectionClosed(tctx.client)
                << Log("Got client close.")
                << CloseConnection(tctx.client)

            

Reported by Pylint.

              import pytest

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

from mitmproxy.addons import stickycookie
from mitmproxy.test import tutils as ntutils



            

Reported by Pylint.

              import pytest

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

from mitmproxy.addons import stickycookie
from mitmproxy.test import tutils as ntutils



            

Reported by Pylint.

              from mitmproxy.test import tutils as ntutils


def test_domain_match():
    assert stickycookie.domain_match("www.google.com", ".google.com")
    assert stickycookie.domain_match("google.com", ".google.com")


class TestStickyCookie:

            

Reported by Pylint.

              

def test_domain_match():
    assert stickycookie.domain_match("www.google.com", ".google.com")
    assert stickycookie.domain_match("google.com", ".google.com")


class TestStickyCookie:
    def test_config(self):

            

Reported by Bandit.

              
def test_domain_match():
    assert stickycookie.domain_match("www.google.com", ".google.com")
    assert stickycookie.domain_match("google.com", ".google.com")


class TestStickyCookie:
    def test_config(self):
        sc = stickycookie.StickyCookie()

            

Reported by Bandit.

                  assert stickycookie.domain_match("google.com", ".google.com")


class TestStickyCookie:
    def test_config(self):
        sc = stickycookie.StickyCookie()
        with taddons.context(sc) as tctx:
            with pytest.raises(Exception, match="invalid filter"):
                tctx.configure(sc, stickycookie="~b")

            

Reported by Pylint.

              

class TestStickyCookie:
    def test_config(self):
        sc = stickycookie.StickyCookie()
        with taddons.context(sc) as tctx:
            with pytest.raises(Exception, match="invalid filter"):
                tctx.configure(sc, stickycookie="~b")


            

Reported by Pylint.

              

class TestStickyCookie:
    def test_config(self):
        sc = stickycookie.StickyCookie()
        with taddons.context(sc) as tctx:
            with pytest.raises(Exception, match="invalid filter"):
                tctx.configure(sc, stickycookie="~b")


            

Reported by Pylint.

              
class TestStickyCookie:
    def test_config(self):
        sc = stickycookie.StickyCookie()
        with taddons.context(sc) as tctx:
            with pytest.raises(Exception, match="invalid filter"):
                tctx.configure(sc, stickycookie="~b")

            tctx.configure(sc, stickycookie="foo")

            

Reported by Pylint.

                              tctx.configure(sc, stickycookie="~b")

            tctx.configure(sc, stickycookie="foo")
            assert sc.flt
            tctx.configure(sc, stickycookie=None)
            assert not sc.flt

    def test_simple(self):
        sc = stickycookie.StickyCookie()

            

Reported by Bandit.

              import urwid
import blinker
import textwrap
import pprint
from typing import Optional, Sequence

from mitmproxy import exceptions
from mitmproxy import optmanager
from mitmproxy.tools.console import layoutwidget

            

Reported by Pylint.

                      self.set_focus(0)
        self.master.options.changed.connect(self.sig_mod)

    def sig_mod(self, *args, **kwargs):
        self.opts = sorted(self.master.options.keys())
        self.maxlen = max(len(i) for i in self.opts)
        self._modified()
        self.set_focus(self.index)


            

Reported by Pylint.

                      self.set_focus(0)
        self.master.options.changed.connect(self.sig_mod)

    def sig_mod(self, *args, **kwargs):
        self.opts = sorted(self.master.options.keys())
        self.maxlen = max(len(i) for i in self.opts)
        self._modified()
        self.set_focus(self.index)


            

Reported by Pylint.

              
    def start_editing(self):
        self.editing = True
        self.focus_obj = self._get(self.index, True)
        self._modified()

    def stop_editing(self):
        self.editing = False
        self.focus_obj = self._get(self.index, False)

            

Reported by Pylint.

              
    def stop_editing(self):
        self.editing = False
        self.focus_obj = self._get(self.index, False)
        self.set_focus(self.index)
        self._modified()

    def get_edit_text(self):
        return self.focus_obj.get_edit_text()

            

Reported by Pylint.

              
    def _get(self, pos, editing):
        name = self.opts[pos]
        opt = self.master.options._options[name]
        return OptionItem(
            self, opt, pos == self.index, self.maxlen, editing
        )

    def get_focus(self):

            

Reported by Pylint.

                  def set_focus(self, index):
        self.editing = False
        name = self.opts[index]
        opt = self.master.options._options[name]
        self.index = index
        self.focus_obj = self._get(self.index, self.editing)
        option_focus_change.send(opt.help)
        self._modified()


            

Reported by Pylint.

                  def keypress(self, size, key):
        if self.walker.editing:
            if key == "enter":
                foc, idx = self.get_focus()
                v = self.walker.get_edit_text()
                try:
                    d = self.master.options.parse_setval(foc.opt, v)
                    self.master.options.update(**{foc.opt.name: d})
                except exceptions.OptionsError as v:

            

Reported by Pylint.

                      else:
            if key == "m_start":
                self.set_focus(0)
                self.walker._modified()
            elif key == "m_end":
                self.set_focus(len(self.walker.opts) - 1)
                self.walker._modified()
            elif key == "m_select":
                foc, idx = self.get_focus()

            

Reported by Pylint.

                              self.walker._modified()
            elif key == "m_end":
                self.set_focus(len(self.walker.opts) - 1)
                self.walker._modified()
            elif key == "m_select":
                foc, idx = self.get_focus()
                if foc.opt.typespec == bool:
                    self.master.options.toggler(foc.opt.name)()
                    # Bust the focus widget cache

            

Reported by Pylint.