from ctypes import Structure, Union, windll, POINTER
from ctypes.wintypes import BOOL, DWORD, WCHAR, WORD, SHORT, UINT, HANDLE, LPDWORD, CHAR

# https://docs.microsoft.com/de-de/windows/console/getstdhandle
STD_INPUT_HANDLE = -10
STD_OUTPUT_HANDLE = -11

# https://docs.microsoft.com/de-de/windows/console/setconsolemode
ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004

            

Reported by Pylint.

              ENABLE_WINDOW_INPUT = 0x0008


class COORD(Structure):
    """https://docs.microsoft.com/en-us/windows/console/coord-str"""

    _fields_ = [
        ("X", SHORT),
        ("Y", SHORT),

            

Reported by Pylint.

                  ]


class SMALL_RECT(Structure):
    """https://docs.microsoft.com/en-us/windows/console/small-rect-str"""

    _fields_ = [
        ("Left", SHORT),
        ("Top", SHORT),

            

Reported by Pylint.

                  ]


class SMALL_RECT(Structure):
    """https://docs.microsoft.com/en-us/windows/console/small-rect-str"""

    _fields_ = [
        ("Left", SHORT),
        ("Top", SHORT),

            

Reported by Pylint.

                  ]


class CONSOLE_SCREEN_BUFFER_INFO(Structure):
    """https://docs.microsoft.com/en-us/windows/console/console-screen-buffer-info-str"""

    _fields_ = [
        ("dwSize", COORD),
        ("dwCursorPosition", COORD),

            

Reported by Pylint.

                  ]


class CONSOLE_SCREEN_BUFFER_INFO(Structure):
    """https://docs.microsoft.com/en-us/windows/console/console-screen-buffer-info-str"""

    _fields_ = [
        ("dwSize", COORD),
        ("dwCursorPosition", COORD),

            

Reported by Pylint.

                  ]


class uChar(Union):
    """https://docs.microsoft.com/en-us/windows/console/key-event-record-str"""
    _fields_ = [
        ("AsciiChar", CHAR),
        ("UnicodeChar", WCHAR),
    ]

            

Reported by Pylint.

                  ]


class uChar(Union):
    """https://docs.microsoft.com/en-us/windows/console/key-event-record-str"""
    _fields_ = [
        ("AsciiChar", CHAR),
        ("UnicodeChar", WCHAR),
    ]

            

Reported by Pylint.

                  ]


class KEY_EVENT_RECORD(Structure):
    """https://docs.microsoft.com/en-us/windows/console/key-event-record-str"""

    _fields_ = [
        ("bKeyDown", BOOL),
        ("wRepeatCount", WORD),

            

Reported by Pylint.

                  ]


class KEY_EVENT_RECORD(Structure):
    """https://docs.microsoft.com/en-us/windows/console/key-event-record-str"""

    _fields_ = [
        ("bKeyDown", BOOL),
        ("wRepeatCount", WORD),

            

Reported by Pylint.

              import pytest

from mitmproxy.addons import onboarding
from mitmproxy.test import taddons


@pytest.fixture
def client():
    with onboarding.app.test_client() as client:

            

Reported by Pylint.

              
@pytest.fixture
def client():
    with onboarding.app.test_client() as client:
        yield client


class TestApp:
    def addons(self):

            

Reported by Pylint.

                      return [onboarding.Onboarding()]

    @pytest.mark.asyncio
    async def test_basic(self, client):
        ob = onboarding.Onboarding()
        with taddons.context(ob) as tctx:
            tctx.configure(ob)
            assert client.get("/").status_code == 200


            

Reported by Pylint.

              
    @pytest.mark.parametrize("ext", ["pem", "p12", "cer"])
    @pytest.mark.asyncio
    async def test_cert(self, client, ext, tdata):
        ob = onboarding.Onboarding()
        with taddons.context(ob) as tctx:
            tctx.configure(ob, confdir=tdata.path("mitmproxy/data/confdir"))
            resp = client.get(f"/cert/{ext}")
            assert resp.status_code == 200

            

Reported by Pylint.

              
    @pytest.mark.parametrize("ext", ["pem", "p12", "cer"])
    @pytest.mark.asyncio
    async def test_head(self, client, ext, tdata):
        ob = onboarding.Onboarding()
        with taddons.context(ob) as tctx:
            tctx.configure(ob, confdir=tdata.path("mitmproxy/data/confdir"))
            resp = client.head(f"http://{tctx.options.onboarding_host}/cert/{ext}")
            assert resp.status_code == 200

            

Reported by Pylint.

              import pytest

from mitmproxy.addons import onboarding
from mitmproxy.test import taddons


@pytest.fixture
def client():
    with onboarding.app.test_client() as client:

            

Reported by Pylint.

              

@pytest.fixture
def client():
    with onboarding.app.test_client() as client:
        yield client


class TestApp:

            

Reported by Pylint.

                      yield client


class TestApp:
    def addons(self):
        return [onboarding.Onboarding()]

    @pytest.mark.asyncio
    async def test_basic(self, client):

            

Reported by Pylint.

              

class TestApp:
    def addons(self):
        return [onboarding.Onboarding()]

    @pytest.mark.asyncio
    async def test_basic(self, client):
        ob = onboarding.Onboarding()

            

Reported by Pylint.

              

class TestApp:
    def addons(self):
        return [onboarding.Onboarding()]

    @pytest.mark.asyncio
    async def test_basic(self, client):
        ob = onboarding.Onboarding()

            

Reported by Pylint.

                          # content_view
            t = time.time()
            if message:
                description, lines, error = contentviews.get_message_content_view(
                    'Auto', message, f
                )
            else:
                description, lines = 'No content.', []
            if time.time() - t > 0.1:

            

Reported by Pylint.

              

class StaticViewer:
    # TODO: make this a command at some point.
    def load(self, loader):
        loader.add_option(
            "web_static_viewer", typing.Optional[str], "",
            "The path to output a static viewer."
        )

            

Reported by Pylint.

              import json
import os.path
import pathlib
import shutil
import time
import typing

from mitmproxy import contentviews, http
from mitmproxy import ctx

            

Reported by Pylint.

                  shutil.copytree(str(web_dir / "static"), str(path / "static"))
    shutil.copyfile(str(web_dir / 'templates' / 'index.html'), str(path / "index.html"))

    with open(str(path / "static" / "static.js"), "w") as f:
        f.write("MITMWEB_STATIC = true;")


def save_filter_help(path: pathlib.Path) -> None:
    with open(str(path / 'filter-help.json'), 'w') as f:

            

Reported by Pylint.

                      f.write("MITMWEB_STATIC = true;")


def save_filter_help(path: pathlib.Path) -> None:
    with open(str(path / 'filter-help.json'), 'w') as f:
        json.dump(dict(commands=flowfilter.help), f)


def save_settings(path: pathlib.Path) -> None:

            

Reported by Pylint.

              

def save_filter_help(path: pathlib.Path) -> None:
    with open(str(path / 'filter-help.json'), 'w') as f:
        json.dump(dict(commands=flowfilter.help), f)


def save_settings(path: pathlib.Path) -> None:
    with open(str(path / 'settings.json'), 'w') as f:

            

Reported by Pylint.

                      json.dump(dict(commands=flowfilter.help), f)


def save_settings(path: pathlib.Path) -> None:
    with open(str(path / 'settings.json'), 'w') as f:
        json.dump(dict(version=version.VERSION), f)


def save_flows(path: pathlib.Path, flows: typing.Iterable[flow.Flow]) -> None:

            

Reported by Pylint.

              

def save_settings(path: pathlib.Path) -> None:
    with open(str(path / 'settings.json'), 'w') as f:
        json.dump(dict(version=version.VERSION), f)


def save_flows(path: pathlib.Path, flows: typing.Iterable[flow.Flow]) -> None:
    with open(str(path / 'flows.json'), 'w') as f:

            

Reported by Pylint.

                      json.dump(dict(version=version.VERSION), f)


def save_flows(path: pathlib.Path, flows: typing.Iterable[flow.Flow]) -> None:
    with open(str(path / 'flows.json'), 'w') as f:
        json.dump(
            [flow_to_json(f) for f in flows],
            f
        )

            

Reported by Pylint.

              

def save_flows(path: pathlib.Path, flows: typing.Iterable[flow.Flow]) -> None:
    with open(str(path / 'flows.json'), 'w') as f:
        json.dump(
            [flow_to_json(f) for f in flows],
            f
        )


            

Reported by Pylint.

              
def mitmproxy(args=None) -> typing.Optional[int]:  # pragma: no cover
    if os.name == "nt":
        import urwid
        urwid.set_encoding("utf8")
    else:
        assert_utf8_env()
    from mitmproxy.tools import console
    run(console.master.ConsoleMaster, cmdline.mitmproxy, args)

            

Reported by Pylint.

                      sys.exit(1)


def process_options(parser, opts, args):
    if args.version:
        print(debug.dump_system_info())
        sys.exit(0)
    if args.quiet or args.options or args.commands:
        # also reduce log verbosity if --options or --commands is passed,

            

Reported by Pylint.

                  debug.register_info_dumpers()

    opts = options.Options()
    master = master_cls(opts)

    parser = make_parser(opts)

    # To make migration from 2.x to 3.0 bearable.
    if "-R" in sys.argv and sys.argv[sys.argv.index("-R") + 1].startswith("http"):

            

Reported by Pylint.

                      if extra:
            if args.filter_args:
                master.log.info(f"Only processing flows that match \"{' & '.join(args.filter_args)}\"")
            opts.update(**extra(args))

        loop = asyncio.get_event_loop()
        try:
            loop.add_signal_handler(signal.SIGINT, getattr(master, "prompt_for_exit", master.shutdown))
            loop.add_signal_handler(signal.SIGTERM, master.shutdown)

            

Reported by Bandit.

              import argparse
import asyncio
import os
import signal
import sys
import typing

from mitmproxy import exceptions, master
from mitmproxy import options

            

Reported by Pylint.

              from mitmproxy.utils import debug, arg_check


def assert_utf8_env():
    spec = ""
    for i in ["LANG", "LC_CTYPE", "LC_ALL"]:
        spec += os.environ.get(i, "").lower()
    if "utf" not in spec:
        print(

            

Reported by Pylint.

                      sys.exit(1)


def process_options(parser, opts, args):
    if args.version:
        print(debug.dump_system_info())
        sys.exit(0)
    if args.quiet or args.options or args.commands:
        # also reduce log verbosity if --options or --commands is passed,

            

Reported by Pylint.

                      args.flow_detail = 2

    adict = {}
    for n in dir(args):
        if n in opts:
            adict[n] = getattr(args, n)
    opts.merge(adict)



            

Reported by Pylint.

                          sys.exit(0)
        if extra:
            if args.filter_args:
                master.log.info(f"Only processing flows that match \"{' & '.join(args.filter_args)}\"")
            opts.update(**extra(args))

        loop = asyncio.get_event_loop()
        try:
            loop.add_signal_handler(signal.SIGINT, getattr(master, "prompt_for_exit", master.shutdown))

            

Reported by Pylint.

              
        loop = asyncio.get_event_loop()
        try:
            loop.add_signal_handler(signal.SIGINT, getattr(master, "prompt_for_exit", master.shutdown))
            loop.add_signal_handler(signal.SIGTERM, master.shutdown)
        except NotImplementedError:
            # Not supported on Windows
            pass


            

Reported by Pylint.

              
    def __init__(
            self,
            context: context.Context,
            tunnel_connection: connection.Connection,
            conn: connection.Connection,
    ):
        super().__init__(context)
        self.tunnel_connection = tunnel_connection

            

Reported by Pylint.

                  def start_handshake(self) -> layer.CommandGenerator[None]:
        yield from self._handle_event(events.DataReceived(self.tunnel_connection, b""))

    def receive_handshake_data(self, data: bytes) -> layer.CommandGenerator[Tuple[bool, Optional[str]]]:
        """returns a (done, err) tuple"""
        yield from ()
        return True, None

    def on_handshake_error(self, err: str) -> layer.CommandGenerator[None]:

            

Reported by Pylint.

                      yield from ()
        return True, None

    def on_handshake_error(self, err: str) -> layer.CommandGenerator[None]:
        """Called if either receive_handshake_data returns an error or we receive a close during handshake."""
        yield commands.CloseConnection(self.tunnel_connection)

    def receive_data(self, data: bytes) -> layer.CommandGenerator[None]:
        yield from self.event_to_child(

            

Reported by Pylint.

              from enum import Enum, auto
from typing import List, Optional, Tuple, Union

from mitmproxy import connection
from mitmproxy.proxy import commands, context, events, layer
from mitmproxy.proxy.layer import Layer


class TunnelState(Enum):

            

Reported by Pylint.

              from mitmproxy.proxy.layer import Layer


class TunnelState(Enum):
    INACTIVE = auto()
    ESTABLISHING = auto()
    OPEN = auto()
    CLOSED = auto()


            

Reported by Pylint.

              
class TunnelLayer(layer.Layer):
    """
    A specialized layer that simplifies the implementation of tunneling protocols such as SOCKS, upstream HTTP proxies,
    or TLS.
    """
    child_layer: layer.Layer
    tunnel_connection: connection.Connection
    """The 'outer' connection which provides the tunnel protocol I/O"""

            

Reported by Pylint.

                  def __repr__(self):
        return f"{type(self).__name__}({self.tunnel_state.name.lower()})"

    def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
        if isinstance(event, events.Start):
            if self.tunnel_connection.state is not connection.ConnectionState.CLOSED:
                # we might be in the interesting state here where the connection is already half-closed,
                # for example because next_layer buffered events and the client disconnected in the meantime.
                # we still expect a close event to arrive, so we carry on here as normal for now.

            

Reported by Pylint.

                  def _handle_event(self, event: events.Event) -> layer.CommandGenerator[None]:
        if isinstance(event, events.Start):
            if self.tunnel_connection.state is not connection.ConnectionState.CLOSED:
                # we might be in the interesting state here where the connection is already half-closed,
                # for example because next_layer buffered events and the client disconnected in the meantime.
                # we still expect a close event to arrive, so we carry on here as normal for now.
                self.tunnel_state = TunnelState.ESTABLISHING
                yield from self.start_handshake()
            yield from self.event_to_child(event)

            

Reported by Pylint.

                      if isinstance(event, events.Start):
            if self.tunnel_connection.state is not connection.ConnectionState.CLOSED:
                # we might be in the interesting state here where the connection is already half-closed,
                # for example because next_layer buffered events and the client disconnected in the meantime.
                # we still expect a close event to arrive, so we carry on here as normal for now.
                self.tunnel_state = TunnelState.ESTABLISHING
                yield from self.start_handshake()
            yield from self.event_to_child(event)
        elif isinstance(event, events.ConnectionEvent) and event.connection == self.tunnel_connection:

            

Reported by Pylint.

                              self.tunnel_state = TunnelState.ESTABLISHING
                yield from self.start_handshake()
            yield from self.event_to_child(event)
        elif isinstance(event, events.ConnectionEvent) and event.connection == self.tunnel_connection:
            if isinstance(event, events.DataReceived):
                if self.tunnel_state is TunnelState.ESTABLISHING:
                    done, err = yield from self.receive_handshake_data(event.data)
                    if done:
                        if self.conn != self.tunnel_connection:

            

Reported by Pylint.

                  def __init__(self, fo):
        self.fo = fo

    def add(self, flow):
        d = flow.get_state()
        tnetstring.dump(d, self.fo)


class FlowReader:

            

Reported by Pylint.

                      """
        try:
            while True:
                # FIXME: This cast hides a lack of dynamic type checking
                loaded = cast(
                    Dict[Union[bytes, str], Any],
                    tnetstring.load(self.fo),
                )
                try:

            

Reported by Pylint.

              import os
from typing import Any, Dict, IO, Iterable, Type, Union, cast

from mitmproxy import exceptions
from mitmproxy import flow
from mitmproxy import flowfilter
from mitmproxy import http
from mitmproxy import tcp
from mitmproxy.io import compat

            

Reported by Pylint.

              )


class FlowWriter:
    def __init__(self, fo):
        self.fo = fo

    def add(self, flow):
        d = flow.get_state()

            

Reported by Pylint.

              )


class FlowWriter:
    def __init__(self, fo):
        self.fo = fo

    def add(self, flow):
        d = flow.get_state()

            

Reported by Pylint.

              
class FlowWriter:
    def __init__(self, fo):
        self.fo = fo

    def add(self, flow):
        d = flow.get_state()
        tnetstring.dump(d, self.fo)


            

Reported by Pylint.

                  def __init__(self, fo):
        self.fo = fo

    def add(self, flow):
        d = flow.get_state()
        tnetstring.dump(d, self.fo)


class FlowReader:

            

Reported by Pylint.

                      self.fo = fo

    def add(self, flow):
        d = flow.get_state()
        tnetstring.dump(d, self.fo)


class FlowReader:
    def __init__(self, fo: IO[bytes]):

            

Reported by Pylint.

                      tnetstring.dump(d, self.fo)


class FlowReader:
    def __init__(self, fo: IO[bytes]):
        self.fo: IO[bytes] = fo

    def stream(self) -> Iterable[flow.Flow]:
        """

            

Reported by Pylint.

                      tnetstring.dump(d, self.fo)


class FlowReader:
    def __init__(self, fo: IO[bytes]):
        self.fo: IO[bytes] = fo

    def stream(self) -> Iterable[flow.Flow]:
        """

            

Reported by Pylint.

              # This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild

from pkg_resources import parse_version
from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO


if parse_version(ks_version) < parse_version('0.7'):
    raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version))


            

Reported by Pylint.

                      def has_next(self):
            """If true, then we have more bytes to read."""
            if hasattr(self, '_m_has_next'):
                return self._m_has_next if hasattr(self, '_m_has_next') else None

            self._m_has_next = (self.b & 128) != 0
            return self._m_has_next if hasattr(self, '_m_has_next') else None

        @property

            

Reported by Pylint.

                      def value(self):
            """The 7-bit (base128) numeric value of this group."""
            if hasattr(self, '_m_value'):
                return self._m_value if hasattr(self, '_m_value') else None

            self._m_value = (self.b & 127)
            return self._m_value if hasattr(self, '_m_value') else None



            

Reported by Pylint.

                  @property
    def len(self):
        if hasattr(self, '_m_len'):
            return self._m_len if hasattr(self, '_m_len') else None

        self._m_len = len(self.groups)
        return self._m_len if hasattr(self, '_m_len') else None

    @property

            

Reported by Pylint.

                  def value(self):
        """Resulting value as normal integer."""
        if hasattr(self, '_m_value'):
            return self._m_value if hasattr(self, '_m_value') else None

        self._m_value = (((((((self.groups[0].value + ((self.groups[1].value << 7) if self.len >= 2 else 0)) + ((self.groups[2].value << 14) if self.len >= 3 else 0)) + ((self.groups[3].value << 21) if self.len >= 4 else 0)) + ((self.groups[4].value << 28) if self.len >= 5 else 0)) + ((self.groups[5].value << 35) if self.len >= 6 else 0)) + ((self.groups[6].value << 42) if self.len >= 7 else 0)) + ((self.groups[7].value << 49) if self.len >= 8 else 0))
        return self._m_value if hasattr(self, '_m_value') else None



            

Reported by Pylint.

              # This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild

from pkg_resources import parse_version
from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO


if parse_version(ks_version) < parse_version('0.7'):
    raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version))


            

Reported by Pylint.

              # This is a generated file! Please edit source .ksy file and use kaitai-struct-compiler to rebuild

from pkg_resources import parse_version
from kaitaistruct import __version__ as ks_version, KaitaiStruct, KaitaiStream, BytesIO


if parse_version(ks_version) < parse_version('0.7'):
    raise Exception("Incompatible Kaitai Struct Python API: 0.7 or later is required, but you have %s" % (ks_version))


            

Reported by Pylint.

                          if hasattr(self, '_m_has_next'):
                return self._m_has_next if hasattr(self, '_m_has_next') else None

            self._m_has_next = (self.b & 128) != 0
            return self._m_has_next if hasattr(self, '_m_has_next') else None

        @property
        def value(self):
            """The 7-bit (base128) numeric value of this group."""

            

Reported by Pylint.

                          if hasattr(self, '_m_value'):
                return self._m_value if hasattr(self, '_m_value') else None

            self._m_value = (self.b & 127)
            return self._m_value if hasattr(self, '_m_value') else None


    @property
    def len(self):

            

Reported by Pylint.

                      if hasattr(self, '_m_len'):
            return self._m_len if hasattr(self, '_m_len') else None

        self._m_len = len(self.groups)
        return self._m_len if hasattr(self, '_m_len') else None

    @property
    def value(self):
        """Resulting value as normal integer."""

            

Reported by Pylint.

              from unittest.mock import Mock

import pytest

from mitmproxy import connection
from mitmproxy.proxy import events, commands


@pytest.fixture

            

Reported by Pylint.

                  return connection.Server(None)


def test_dataclasses(tconn):
    assert repr(events.Start())
    assert repr(events.DataReceived(tconn, b"foo"))
    assert repr(events.ConnectionClosed(tconn))



            

Reported by Pylint.

                      pass

    with pytest.warns(RuntimeWarning, match="properly annotated"):
        class FooCompleted(events.CommandCompleted):
            pass

    class FooCompleted1(events.CommandCompleted):
        command: FooCommand


            

Reported by Pylint.

                      class FooCompleted(events.CommandCompleted):
            pass

    class FooCompleted1(events.CommandCompleted):
        command: FooCommand

    with pytest.warns(RuntimeWarning, match="conflicting subclasses"):
        class FooCompleted2(events.CommandCompleted):
            command: FooCommand

            

Reported by Pylint.

                      command: FooCommand

    with pytest.warns(RuntimeWarning, match="conflicting subclasses"):
        class FooCompleted2(events.CommandCompleted):
            command: FooCommand

            

Reported by Pylint.

              from unittest.mock import Mock

import pytest

from mitmproxy import connection
from mitmproxy.proxy import events, commands


@pytest.fixture

            

Reported by Pylint.

              

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


def test_dataclasses(tconn):
    assert repr(events.Start())

            

Reported by Pylint.

                  return connection.Server(None)


def test_dataclasses(tconn):
    assert repr(events.Start())
    assert repr(events.DataReceived(tconn, b"foo"))
    assert repr(events.ConnectionClosed(tconn))



            

Reported by Pylint.

              

def test_dataclasses(tconn):
    assert repr(events.Start())
    assert repr(events.DataReceived(tconn, b"foo"))
    assert repr(events.ConnectionClosed(tconn))


def test_command_completed():

            

Reported by Bandit.

              
def test_dataclasses(tconn):
    assert repr(events.Start())
    assert repr(events.DataReceived(tconn, b"foo"))
    assert repr(events.ConnectionClosed(tconn))


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

            

Reported by Bandit.

              import pytest

from mitmproxy.contentviews import javascript
from . import full_eval


def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")

            

Reported by Pylint.

              import pytest

from mitmproxy.contentviews import javascript
from . import full_eval


def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")

            

Reported by Pylint.

              def test_format_xml(filename, tdata):
    path = tdata.path("mitmproxy/contentviews/test_js_data/" + filename)
    with open(path) as f:
        input = f.read()
    with open("-formatted.".join(path.rsplit(".", 1))) as f:
        expected = f.read()
    js = javascript.beautify(input)
    assert js == expected


            

Reported by Pylint.

              import pytest

from mitmproxy.contentviews import javascript
from . import full_eval


def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")

            

Reported by Pylint.

              from . import full_eval


def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")
    assert v(b"[1, 2, 3")
    assert v(b"function(a){[1, 2, 3]}") == ("JavaScript", [
        [('text', 'function(a) {')],

            

Reported by Pylint.

              

def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")
    assert v(b"[1, 2, 3")
    assert v(b"function(a){[1, 2, 3]}") == ("JavaScript", [
        [('text', 'function(a) {')],
        [('text', '  [1, 2, 3]')],

            

Reported by Pylint.

              
def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")
    assert v(b"[1, 2, 3")
    assert v(b"function(a){[1, 2, 3]}") == ("JavaScript", [
        [('text', 'function(a) {')],
        [('text', '  [1, 2, 3]')],
        [('text', '}')]

            

Reported by Bandit.

              def test_view_javascript():
    v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")
    assert v(b"[1, 2, 3")
    assert v(b"function(a){[1, 2, 3]}") == ("JavaScript", [
        [('text', 'function(a) {')],
        [('text', '  [1, 2, 3]')],
        [('text', '}')]
    ])

            

Reported by Bandit.

                  v = full_eval(javascript.ViewJavaScript())
    assert v(b"[1, 2, 3]")
    assert v(b"[1, 2, 3")
    assert v(b"function(a){[1, 2, 3]}") == ("JavaScript", [
        [('text', 'function(a) {')],
        [('text', '  [1, 2, 3]')],
        [('text', '}')]
    ])
    assert v(b"\xfe")  # invalid utf-8

            

Reported by Bandit.

                      [('text', '  [1, 2, 3]')],
        [('text', '}')]
    ])
    assert v(b"\xfe")  # invalid utf-8


@pytest.mark.parametrize("filename", [
    "simple.js",
])

            

Reported by Bandit.

              
import gzip
import zlib
import brotli
import zstandard as zstd

from typing import Union, Optional, AnyStr, overload  # noqa

# We have a shared single-element cache for encoding and decoding.

            

Reported by Pylint.

              import gzip
import zlib
import brotli
import zstandard as zstd

from typing import Union, Optional, AnyStr, overload  # noqa

# We have a shared single-element cache for encoding and decoding.
# This is quite useful in practice, e.g.

            

Reported by Pylint.

              import brotli
import zstandard as zstd

from typing import Union, Optional, AnyStr, overload  # noqa

# We have a shared single-element cache for encoding and decoding.
# This is quite useful in practice, e.g.
# flow.request.content = flow.request.content.replace(b"foo", b"bar")
# does not require an .encode() call if content does not contain b"foo"

            

Reported by Pylint.

              import brotli
import zstandard as zstd

from typing import Union, Optional, AnyStr, overload  # noqa

# We have a shared single-element cache for encoding and decoding.
# This is quite useful in practice, e.g.
# flow.request.content = flow.request.content.replace(b"foo", b"bar")
# does not require an .encode() call if content does not contain b"foo"

            

Reported by Pylint.

                      return None
    encoding = encoding.lower()

    global _cache
    cached = (
            isinstance(encoded, bytes) and
            _cache.encoded == encoded and
            _cache.encoding == encoding and
            _cache.errors == errors

            

Reported by Pylint.

                  except TypeError:
        raise
    except Exception as e:
        raise ValueError("{} when decoding {} with {}: {}".format(
            type(e).__name__,
            repr(encoded)[:10],
            repr(encoding),
            repr(e),
        ))

            

Reported by Pylint.

                      return None
    encoding = encoding.lower()

    global _cache
    cached = (
            isinstance(decoded, bytes) and
            _cache.decoded == decoded and
            _cache.encoding == encoding and
            _cache.errors == errors

            

Reported by Pylint.

                  except TypeError:
        raise
    except Exception as e:
        raise ValueError("{} when encoding {} with {}: {}".format(
            type(e).__name__,
            repr(decoded)[:10],
            repr(encoding),
            repr(e),
        ))

            

Reported by Pylint.

              import brotli
import zstandard as zstd

from typing import Union, Optional, AnyStr, overload  # noqa

# We have a shared single-element cache for encoding and decoding.
# This is quite useful in practice, e.g.
# flow.request.content = flow.request.content.replace(b"foo", b"bar")
# does not require an .encode() call if content does not contain b"foo"

            

Reported by Pylint.

                      return None
    encoding = encoding.lower()

    global _cache
    cached = (
            isinstance(encoded, bytes) and
            _cache.encoded == encoded and
            _cache.encoding == encoding and
            _cache.errors == errors

            

Reported by Pylint.