from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import json
from . import full_eval


def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')

            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import json
from . import full_eval


def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')

            

Reported by Pylint.

              from hypothesis.strategies import binary

from mitmproxy.contentviews import json
from . import full_eval


def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None

            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import json
from . import full_eval


def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')

            

Reported by Pylint.

              from . import full_eval


def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR

            

Reported by Pylint.

              

def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR


            

Reported by Bandit.

              
def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR



            

Reported by Bandit.

              def test_parse_json():
    assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR


def test_format_json():

            

Reported by Bandit.

                  assert json.parse_json(b'{"foo": 1}')
    assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR


def test_format_json():
    assert list(json.format_json({

            

Reported by Bandit.

                  assert json.parse_json(b'null') is None
    assert json.parse_json(b"moo") is json.PARSE_ERROR
    assert json.parse_json(b'{"foo" : "\xe4\xb8\x96\xe7\x95\x8c"}')  # utf8 with chinese characters
    assert json.parse_json(b'{"foo" : "\xFF"}') is json.PARSE_ERROR


def test_format_json():
    assert list(json.format_json({
        "data": [

            

Reported by Bandit.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import graphql
from . import full_eval


def test_render_priority():
    v = graphql.ViewGraphQL()

            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import graphql
from . import full_eval


def test_render_priority():
    v = graphql.ViewGraphQL()

            

Reported by Pylint.

              from hypothesis.strategies import binary

from mitmproxy.contentviews import graphql
from . import full_eval


def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")

            

Reported by Pylint.

              from hypothesis import given
from hypothesis.strategies import binary

from mitmproxy.contentviews import graphql
from . import full_eval


def test_render_priority():
    v = graphql.ViewGraphQL()

            

Reported by Pylint.

              from . import full_eval


def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")

            

Reported by Pylint.

              

def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"}", content_type="application/json")

            

Reported by Pylint.

              
def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"}", content_type="application/json")


            

Reported by Pylint.

              
def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"}", content_type="application/json")


            

Reported by Pylint.

              
def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"}", content_type="application/json")


            

Reported by Bandit.

              def test_render_priority():
    v = graphql.ViewGraphQL()
    assert 2 == v.render_priority(b"""{"query": "query P { \\n }"}""", content_type="application/json")
    assert 2 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"""[{"query": "query P { \\n }"}]""", content_type="text/html")
    assert 0 == v.render_priority(b"""[{"xquery": "query P { \\n }"}]""", content_type="application/json")
    assert 0 == v.render_priority(b"}", content_type="application/json")



            

Reported by Pylint.

              
from mitmproxy.coretypes import multidict

"""
A flexible module for cookie parsing and manipulation.

This module differs from usual standards-compliant cookie modules in a number
of ways. We try to be as permissive as possible, and to retain even mal-formed
information. Duplicate cookies are preserved in parsing, and can be set in

            

Reported by Pylint.

              

TSetCookie = Tuple[str, str, CookieAttrs]
TPairs = List[List[str]]  # TODO: Should be List[Tuple[str,str]]?


def _read_until(s, start, term):
    """
        Read until one of the characters in term is reached.

            

Reported by Pylint.

                      Parse a Cookie header value.
        Returns a list of (lhs, rhs) tuples.
    """
    pairs, off_ = _read_cookie_pairs(line)
    return pairs


def parse_cookie_headers(cookie_headers):
    cookie_list = []

            

Reported by Pylint.

                      CookieAttrs dict of attributes. No attempt is made to parse attribute
        values - they are treated purely as strings.
    """
    cookie_pairs, off = _read_set_cookie_pairs(line)
    cookies = []
    for pairs in cookie_pairs:
        if pairs:
            cookie, *attrs = pairs
            cookies.append((

            

Reported by Pylint.

              import email.utils
import re
import time
from typing import Tuple, List, Iterable

from mitmproxy.coretypes import multidict

"""
A flexible module for cookie parsing and manipulation.

            

Reported by Pylint.

              ESCAPE = re.compile(r"([\"\\])")


class CookieAttrs(multidict.MultiDict):
    @staticmethod
    def _kconv(key):
        return key.lower()

    @staticmethod

            

Reported by Pylint.

              TPairs = List[List[str]]  # TODO: Should be List[Tuple[str,str]]?


def _read_until(s, start, term):
    """
        Read until one of the characters in term is reached.
    """
    if start == len(s):
        return "", start + 1

            

Reported by Pylint.

                  return s[start:i + 1], i + 1


def _read_quoted_string(s, start):
    """
        start: offset to the first quote of the string to be read

        A sort of loose super-set of the various quoted string specifications.


            

Reported by Pylint.

                  return "".join(ret), i + 1


def _read_key(s, start, delims=";="):
    """
        Read a key - the LHS of a token/value pair in a cookie.
    """
    return _read_until(s, start, delims)


            

Reported by Pylint.

                  return _read_until(s, start, delims)


def _read_value(s, start, delims):
    """
        Reads a value - the RHS of a token/value pair in a cookie.
    """
    if start >= len(s):
        return "", start

            

Reported by Pylint.

              import pytest

from mitmproxy import http
from mitmproxy import websocket
from mitmproxy.test import tflow
from wsproto.frame_protocol import Opcode


class TestWebSocketData:

            

Reported by Pylint.

              from mitmproxy import http
from mitmproxy import websocket
from mitmproxy.test import tflow
from wsproto.frame_protocol import Opcode


class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

            

Reported by Pylint.

              
    def test_text(self):
        txt = websocket.WebSocketMessage(Opcode.TEXT, True, b"foo")
        bin = websocket.WebSocketMessage(Opcode.BINARY, True, b"foo")

        assert txt.is_text
        assert txt.text == "foo"
        txt.text = "bar"
        assert txt.content == b"bar"

            

Reported by Pylint.

              import pytest

from mitmproxy import http
from mitmproxy import websocket
from mitmproxy.test import tflow
from wsproto.frame_protocol import Opcode


class TestWebSocketData:

            

Reported by Pylint.

              from mitmproxy import http
from mitmproxy import websocket
from mitmproxy.test import tflow
from wsproto.frame_protocol import Opcode


class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

            

Reported by Pylint.

              from wsproto.frame_protocol import Opcode


class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

    def test_state(self):
        f = tflow.twebsocketflow()

            

Reported by Pylint.

              

class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

    def test_state(self):
        f = tflow.twebsocketflow()
        f2 = http.HTTPFlow.from_state(f.get_state())

            

Reported by Pylint.

              

class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

    def test_state(self):
        f = tflow.twebsocketflow()
        f2 = http.HTTPFlow.from_state(f.get_state())

            

Reported by Pylint.

              
class TestWebSocketData:
    def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

    def test_state(self):
        f = tflow.twebsocketflow()
        f2 = http.HTTPFlow.from_state(f.get_state())
        f2.set_state(f.get_state())

            

Reported by Bandit.

                  def test_repr(self):
        assert repr(tflow.twebsocketflow().websocket) == "<WebSocketData (3 messages)>"

    def test_state(self):
        f = tflow.twebsocketflow()
        f2 = http.HTTPFlow.from_state(f.get_state())
        f2.set_state(f.get_state())



            

Reported by Pylint.

              import json

import flask
import pytest
from flask import request

from mitmproxy.addons import asgiapp
from mitmproxy.addons import next_layer
from mitmproxy.addons.proxyserver import Proxyserver

            

Reported by Pylint.

                  raise ValueError("errapp")


async def noresponseapp(scope, receive, send):
    return


@pytest.mark.asyncio
async def test_asgi_full():

            

Reported by Pylint.

                  raise ValueError("errapp")


async def noresponseapp(scope, receive, send):
    return


@pytest.mark.asyncio
async def test_asgi_full():

            

Reported by Pylint.

                  raise ValueError("errapp")


async def noresponseapp(scope, receive, send):
    return


@pytest.mark.asyncio
async def test_asgi_full():

            

Reported by Pylint.

                      proxy_addr = ps.server.sockets[0].getsockname()[:2]

        reader, writer = await asyncio.open_connection(*proxy_addr)
        req = f"GET http://testapp:80/ HTTP/1.1\r\n\r\n"
        writer.write(req.encode())
        header = await reader.readuntil(b"\r\n\r\n")
        assert header.startswith(b"HTTP/1.1 200 OK")
        body = await reader.readuntil(b"testapp")
        assert body == b"testapp"

            

Reported by Pylint.

                      assert body == b"testapp"

        reader, writer = await asyncio.open_connection(*proxy_addr)
        req = f"GET http://testapp:80/parameters?param1=1&param2=2 HTTP/1.1\r\n\r\n"
        writer.write(req.encode())
        header = await reader.readuntil(b"\r\n\r\n")
        assert header.startswith(b"HTTP/1.1 200 OK")
        body = await reader.readuntil(b"}")
        assert body == b'{"param1": "1", "param2": "2"}'

            

Reported by Pylint.

                      assert body == b'{"param1": "1", "param2": "2"}'

        reader, writer = await asyncio.open_connection(*proxy_addr)
        req = f"POST http://testapp:80/requestbody HTTP/1.1\r\nContent-Length: 6\r\n\r\nHello!"
        writer.write(req.encode())
        header = await reader.readuntil(b"\r\n\r\n")
        assert header.startswith(b"HTTP/1.1 200 OK")
        body = await reader.readuntil(b"}")
        assert body == b'{"body": "Hello!"}'

            

Reported by Pylint.

                      assert body == b'{"body": "Hello!"}'

        reader, writer = await asyncio.open_connection(*proxy_addr)
        req = f"GET http://errapp:80/?foo=bar HTTP/1.1\r\n\r\n"
        writer.write(req.encode())
        header = await reader.readuntil(b"\r\n\r\n")
        assert header.startswith(b"HTTP/1.1 500")
        body = await reader.readuntil(b"ASGI Error")
        assert body == b"ASGI Error"

            

Reported by Pylint.

                      assert body == b"ASGI Error"

        reader, writer = await asyncio.open_connection(*proxy_addr)
        req = f"GET http://noresponseapp:80/ HTTP/1.1\r\n\r\n"
        writer.write(req.encode())
        header = await reader.readuntil(b"\r\n\r\n")
        assert header.startswith(b"HTTP/1.1 500")
        body = await reader.readuntil(b"ASGI Error")
        assert body == b"ASGI Error"

            

Reported by Pylint.

              import asyncio
import json

import flask
import pytest
from flask import request

from mitmproxy.addons import asgiapp
from mitmproxy.addons import next_layer

            

Reported by Pylint.

                      f.close()
        ctx.log.alert("Saved %s flows." % len(flows))

    def tcp_start(self, flow):
        if self.stream:
            self.active_flows.add(flow)

    def tcp_end(self, flow):
        if self.stream:

            

Reported by Pylint.

                      if self.stream:
            self.active_flows.add(flow)

    def tcp_end(self, flow):
        if self.stream:
            self.stream.add(flow)
            self.active_flows.discard(flow)

    def tcp_error(self, flow):

            

Reported by Pylint.

                          self.stream.add(flow)
            self.active_flows.discard(flow)

    def tcp_error(self, flow):
        self.tcp_end(flow)

    def websocket_end(self, flow: http.HTTPFlow):
        if self.stream:
            self.stream.add(flow)

            

Reported by Pylint.

                  def tcp_error(self, flow):
        self.tcp_end(flow)

    def websocket_end(self, flow: http.HTTPFlow):
        if self.stream:
            self.stream.add(flow)
            self.active_flows.discard(flow)

    def request(self, flow: http.HTTPFlow):

            

Reported by Pylint.

                          self.stream.add(flow)
            self.active_flows.discard(flow)

    def request(self, flow: http.HTTPFlow):
        if self.stream:
            self.active_flows.add(flow)

    def response(self, flow: http.HTTPFlow):
        # websocket flows will receive a websocket_end,

            

Reported by Pylint.

                      if self.stream:
            self.active_flows.add(flow)

    def response(self, flow: http.HTTPFlow):
        # websocket flows will receive a websocket_end,
        # we don't want to persist them here already
        if self.stream and flow.websocket is None:
            self.stream.add(flow)
            self.active_flows.discard(flow)

            

Reported by Pylint.

                          self.stream.add(flow)
            self.active_flows.discard(flow)

    def error(self, flow: http.HTTPFlow):
        self.response(flow)

    def done(self):
        if self.stream:
            for f in self.active_flows:

            

Reported by Pylint.

              import os.path
import typing

from mitmproxy import command
from mitmproxy import exceptions
from mitmproxy import flowfilter
from mitmproxy import io
from mitmproxy import ctx
from mitmproxy import flow

            

Reported by Pylint.

              import mitmproxy.types


class Save:
    def __init__(self):
        self.stream = None
        self.filt = None
        self.active_flows: typing.Set[flow.Flow] = set()


            

Reported by Pylint.

                      self.filt = None
        self.active_flows: typing.Set[flow.Flow] = set()

    def load(self, loader):
        loader.add_option(
            "save_stream_file", typing.Optional[str], None,
            "Stream flows to file as they arrive. Prefix path with + to append."
        )
        loader.add_option(

            

Reported by Pylint.

              from dataclasses import dataclass
from typing import Iterator, List

import wsproto
import wsproto.extensions
import wsproto.frame_protocol
import wsproto.utilities
from mitmproxy import connection, http, websocket
from mitmproxy.proxy import commands, events, layer

            

Reported by Pylint.

              from typing import Iterator, List

import wsproto
import wsproto.extensions
import wsproto.frame_protocol
import wsproto.utilities
from mitmproxy import connection, http, websocket
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.commands import StartHook

            

Reported by Pylint.

              
import wsproto
import wsproto.extensions
import wsproto.frame_protocol
import wsproto.utilities
from mitmproxy import connection, http, websocket
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.commands import StartHook
from mitmproxy.proxy.context import Context

            

Reported by Pylint.

              import wsproto
import wsproto.extensions
import wsproto.frame_protocol
import wsproto.utilities
from mitmproxy import connection, http, websocket
from mitmproxy.proxy import commands, events, layer
from mitmproxy.proxy.commands import StartHook
from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import MessageInjected

            

Reported by Pylint.

              from mitmproxy.proxy.context import Context
from mitmproxy.proxy.events import MessageInjected
from mitmproxy.proxy.utils import expect
from wsproto import ConnectionState
from wsproto.frame_protocol import Opcode


@dataclass
class WebsocketStartHook(StartHook):

            

Reported by Pylint.

              from mitmproxy.proxy.events import MessageInjected
from mitmproxy.proxy.utils import expect
from wsproto import ConnectionState
from wsproto.frame_protocol import Opcode


@dataclass
class WebsocketStartHook(StartHook):
    """

            

Reported by Pylint.

                          src_ws.receive_data(None)
        elif isinstance(event, WebSocketMessageInjected):
            fragmentizer = Fragmentizer([], event.message.type == Opcode.TEXT)
            src_ws._events.extend(
                fragmentizer(event.message.content)
            )
        else:  # pragma: no cover
            raise AssertionError(f"Unexpected event: {event}")


            

Reported by Pylint.

              import time
from dataclasses import dataclass
from typing import Iterator, List

import wsproto
import wsproto.extensions
import wsproto.frame_protocol
import wsproto.utilities
from mitmproxy import connection, http, websocket

            

Reported by Pylint.

                  flow: http.HTTPFlow


class WebSocketMessageInjected(MessageInjected[websocket.WebSocketMessage]):
    """
    The user has injected a custom WebSocket message.
    """



            

Reported by Pylint.

                  frame_buf: List[bytes]

    def __init__(self, *args, conn: connection.Connection, **kwargs):
        super(WebsocketConnection, self).__init__(*args, **kwargs)
        self.conn = conn
        self.frame_buf = [b""]

    def send2(self, event: wsproto.events.Event) -> commands.SendData:
        data = self.send(event)

            

Reported by Pylint.

              from unittest import mock
import pytest

from mitmproxy.net import encoding


@pytest.mark.parametrize("encoder", [
    'identity',
    'none',

            

Reported by Pylint.

              from unittest import mock
import pytest

from mitmproxy.net import encoding


@pytest.mark.parametrize("encoder", [
    'identity',
    'none',

            

Reported by Pylint.

              @pytest.mark.parametrize("encoder", [
    'identity',
    'none',
])
def test_identity(encoder):
    assert b"string" == encoding.decode(b"string", encoder)
    assert b"string" == encoding.encode(b"string", encoder)
    with pytest.raises(ValueError):
        encoding.encode(b"string", "nonexistent encoding")

            

Reported by Pylint.

                  'none',
])
def test_identity(encoder):
    assert b"string" == encoding.decode(b"string", encoder)
    assert b"string" == encoding.encode(b"string", encoder)
    with pytest.raises(ValueError):
        encoding.encode(b"string", "nonexistent encoding")



            

Reported by Bandit.

                  'none',
])
def test_identity(encoder):
    assert b"string" == encoding.decode(b"string", encoder)
    assert b"string" == encoding.encode(b"string", encoder)
    with pytest.raises(ValueError):
        encoding.encode(b"string", "nonexistent encoding")



            

Reported by Pylint.

              ])
def test_identity(encoder):
    assert b"string" == encoding.decode(b"string", encoder)
    assert b"string" == encoding.encode(b"string", encoder)
    with pytest.raises(ValueError):
        encoding.encode(b"string", "nonexistent encoding")


@pytest.mark.parametrize("encoder", [

            

Reported by Pylint.

              ])
def test_identity(encoder):
    assert b"string" == encoding.decode(b"string", encoder)
    assert b"string" == encoding.encode(b"string", encoder)
    with pytest.raises(ValueError):
        encoding.encode(b"string", "nonexistent encoding")


@pytest.mark.parametrize("encoder", [

            

Reported by Bandit.

                  """
        This test is for testing byte->byte encoding/decoding
    """
    assert encoding.decode(None, encoder) is None
    assert encoding.encode(None, encoder) is None

    assert b"" == encoding.decode(b"", encoder)

    assert b"string" == encoding.decode(

            

Reported by Bandit.

                      This test is for testing byte->byte encoding/decoding
    """
    assert encoding.decode(None, encoder) is None
    assert encoding.encode(None, encoder) is None

    assert b"" == encoding.decode(b"", encoder)

    assert b"string" == encoding.decode(
        encoding.encode(

            

Reported by Bandit.

                  assert encoding.decode(None, encoder) is None
    assert encoding.encode(None, encoder) is None

    assert b"" == encoding.decode(b"", encoder)

    assert b"string" == encoding.decode(
        encoding.encode(
            b"string",
            encoder

            

Reported by Bandit.

              import pytest

from mitmproxy.addons import blocklist
from mitmproxy.exceptions import OptionsError
from mitmproxy.test import taddons
from mitmproxy.test import tflow


@pytest.mark.parametrize("filter,err", [

            

Reported by Pylint.

                  ("/~u index.html/999", "Invalid HTTP status code"),
    ("/~u index.html/abc", "Invalid HTTP status code"),
])
def test_parse_spec_err(filter, err):
    with pytest.raises(ValueError, match=err):
        blocklist.parse_spec(filter)


class TestBlockList:

            

Reported by Pylint.

                      ("/!png/418", 418),

    ])
    def test_block(self, filter, status_code):
        bl = blocklist.BlockList()
        with taddons.context(bl) as tctx:
            tctx.configure(bl, block_list=[filter])
            f = tflow.tflow()
            f.request.url = b"https://example.org/images/test.jpg"

            

Reported by Pylint.

              import pytest

from mitmproxy.addons import blocklist
from mitmproxy.exceptions import OptionsError
from mitmproxy.test import taddons
from mitmproxy.test import tflow


@pytest.mark.parametrize("filter,err", [

            

Reported by Pylint.

                  (":~d ~d ~d:200", "Invalid filter"),
    ("/~u index.html/999", "Invalid HTTP status code"),
    ("/~u index.html/abc", "Invalid HTTP status code"),
])
def test_parse_spec_err(filter, err):
    with pytest.raises(ValueError, match=err):
        blocklist.parse_spec(filter)



            

Reported by Pylint.

                      blocklist.parse_spec(filter)


class TestBlockList:
    @pytest.mark.parametrize("filter,status_code", [
        (":~u example.org:404", 404),
        (":~u example.com:404", None),
        ("/!jpg/418", None),
        ("/!png/418", 418),

            

Reported by Pylint.

                      (":~u example.com:404", None),
        ("/!jpg/418", None),
        ("/!png/418", 418),

    ])
    def test_block(self, filter, status_code):
        bl = blocklist.BlockList()
        with taddons.context(bl) as tctx:
            tctx.configure(bl, block_list=[filter])

            

Reported by Pylint.

                      (":~u example.com:404", None),
        ("/!jpg/418", None),
        ("/!png/418", 418),

    ])
    def test_block(self, filter, status_code):
        bl = blocklist.BlockList()
        with taddons.context(bl) as tctx:
            tctx.configure(bl, block_list=[filter])

            

Reported by Pylint.

              
    ])
    def test_block(self, filter, status_code):
        bl = blocklist.BlockList()
        with taddons.context(bl) as tctx:
            tctx.configure(bl, block_list=[filter])
            f = tflow.tflow()
            f.request.url = b"https://example.org/images/test.jpg"
            bl.request(f)

            

Reported by Pylint.

                      bl = blocklist.BlockList()
        with taddons.context(bl) as tctx:
            tctx.configure(bl, block_list=[filter])
            f = tflow.tflow()
            f.request.url = b"https://example.org/images/test.jpg"
            bl.request(f)
            if status_code is not None:
                assert f.response.status_code == status_code
                assert f.metadata['blocklisted']

            

Reported by Pylint.

              import typing
from typing import Dict

from bs4 import BeautifulSoup

from mitmproxy.http import HTTPFlow
from examples.contrib.webscanner_helper.urldict import URLDict

NO_CONTENT = object()

            

Reported by Pylint.

              
from bs4 import BeautifulSoup

from mitmproxy.http import HTTPFlow
from examples.contrib.webscanner_helper.urldict import URLDict

NO_CONTENT = object()



            

Reported by Pylint.

              from bs4 import BeautifulSoup

from mitmproxy.http import HTTPFlow
from examples.contrib.webscanner_helper.urldict import URLDict

NO_CONTENT = object()


class MappingAddonConfig:

            

Reported by Pylint.

                  def replace(self, soup: BeautifulSoup, css_sel: str, replace: BeautifulSoup) -> None:
        """Replaces the content of soup that matches the css selector with the given replace content."""
        for content in soup.select(css_sel):
            self.logger.debug(f"replace \"{content}\" with \"{replace}\"")
            content.replace_with(copy.copy(replace))

    def apply_template(self, soup: BeautifulSoup, template: Dict[str, typing.Union[BeautifulSoup]]) -> None:
        """Applies the given mapping template to the given soup."""
        for css_sel, replace in template.items():

            

Reported by Pylint.

                      for css_sel, replace in template.items():
            mapped = soup.select(css_sel)
            if not mapped:
                self.logger.warning(f"Could not find \"{css_sel}\", can not freeze anything.")
            else:
                self.replace(soup, css_sel, BeautifulSoup(replace, features=MappingAddonConfig.HTML_PARSER))

    def response(self, flow: HTTPFlow) -> None:
        """If a response is received, check if we should replace some content. """

            

Reported by Pylint.

                                      self.apply_template(content, template)
                    res.content = content.encode(encoding)
                else:
                    self.logger.warning(f"Unsupported content type '{content_type}' or content encoding '{encoding}'")
        except KeyError:
            pass

    def done(self) -> None:
        """Dumps all new content into the configuration file if self.persistent is set."""

            

Reported by Pylint.

                                  for css_sel, soup in value.items():
                        store[css_sel] = str(soup)
                except:
                    raise RuntimeError(value)
                return store

            with open(self.filename, "w") as f:
                self.mapping_templates.dump(f, value_dumper)

            

Reported by Pylint.

              import copy
import logging
import typing
from typing import Dict

from bs4 import BeautifulSoup

from mitmproxy.http import HTTPFlow
from examples.contrib.webscanner_helper.urldict import URLDict

            

Reported by Pylint.

              NO_CONTENT = object()


class MappingAddonConfig:
    HTML_PARSER = "html.parser"


class MappingAddon:
    """ The mapping add-on can be used in combination with web application scanners to reduce their false positives.

            

Reported by Pylint.

              NO_CONTENT = object()


class MappingAddonConfig:
    HTML_PARSER = "html.parser"


class MappingAddon:
    """ The mapping add-on can be used in combination with web application scanners to reduce their false positives.

            

Reported by Pylint.