from mitmproxy import flowfilter


def _match_all(flow) -> bool:
    return True


def parse_spec(option: str) -> typing.Tuple[flowfilter.TFilter, str, str]:
    """

            

Reported by Pylint.

              import typing
from mitmproxy import flowfilter


def _match_all(flow) -> bool:
    return True


def parse_spec(option: str) -> typing.Tuple[flowfilter.TFilter, str, str]:

            

Reported by Pylint.

                  """
    sep, rem = option[0], option[1:]
    parts = rem.split(sep, 2)
    if len(parts) == 2:
        subject, replacement = parts
        return _match_all, subject, replacement
    elif len(parts) == 3:
        patt, subject, replacement = parts
        flow_filter = flowfilter.parse(patt)

            

Reported by Pylint.

              #!/usr/bin/env python3

import re
from pathlib import Path

here = Path(__file__).absolute().parent
example_dir = here / ".." / "src" / "examples" / "addons"
examples = example_dir.glob('*.py')


            

Reported by Pylint.

              
for example in examples:
    code = example.read_text()
    slug = str(example.with_suffix("").relative_to(example_dir))
    slug = re.sub(r"[^a-zA-Z]", "-", slug)
    match = re.search(r'''
        ^
        (?:[#][^\n]*\n)?  # there might be a shebang
        """

            

Reported by Pylint.

                  if match:
        comment = " — " + match.group(1)
    else:
        comment = ""
    overview.append(
        f"  * [{example.name}](#{slug}){comment}\n"
    )
    listings.append(f"""
<h3 id="{slug}">Example: {example.name}</h3>

            

Reported by Pylint.

              import os.path
import importlib
import inspect


class Data:

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

            

Reported by Pylint.

              import inspect


class Data:

    def __init__(self, name):
        self.name = name
        m = importlib.import_module(name)
        dirname = os.path.dirname(inspect.getsourcefile(m))

            

Reported by Pylint.

              
    def __init__(self, name):
        self.name = name
        m = importlib.import_module(name)
        dirname = os.path.dirname(inspect.getsourcefile(m))
        self.dirname = os.path.abspath(dirname)

    def push(self, subpath):
        """

            

Reported by Pylint.

              def setbit(byte, offset, value):
    """
        Set a bit in a byte to 1 if value is truthy, 0 if not.
    """
    if value:
        return byte | (1 << offset)
    else:
        return byte & ~(1 << offset)


            

Reported by Pylint.

                  """
        Set a bit in a byte to 1 if value is truthy, 0 if not.
    """
    if value:
        return byte | (1 << offset)
    else:
        return byte & ~(1 << offset)



            

Reported by Pylint.

                      return byte & ~(1 << offset)


def getbit(byte, offset):
    mask = 1 << offset
    return bool(byte & mask)

            

Reported by Pylint.

              import asyncio
import sys
import time
from collections.abc import Coroutine
from typing import Optional

from mitmproxy.utils import human



            

Reported by Pylint.

                  Ideally we stop closing the event loop during shutdown and then remove this parameter.
    """
    try:
        t = asyncio.create_task(coro, name=name)
    except RuntimeError:
        if ignore_closed_loop:
            coro.close()
            return None
        else:

            

Reported by Pylint.

                  try:
        t = asyncio.create_task(coro, name=name)
    except RuntimeError:
        if ignore_closed_loop:
            coro.close()
            return None
        else:
            raise
    set_task_debug_info(t, name=name, client=client)

            

Reported by Pylint.

              from .editors import *  # noqa
from . import base  # noqa

            

Reported by Pylint.

              from .editors import *  # noqa
from . import base  # noqa

            

Reported by Pylint.

              from .editors import *  # noqa
from . import base  # noqa

            

Reported by Pylint.

              from mitmproxy.coretypes import basethread


class ScriptThread(basethread.BaseThread):
    name = "ScriptThread"


def concurrent(fn):
    if fn.__name__ not in set(hooks.all_hooks.keys()) - {"load", "configure"}:

            

Reported by Pylint.

                  name = "ScriptThread"


def concurrent(fn):
    if fn.__name__ not in set(hooks.all_hooks.keys()) - {"load", "configure"}:
        raise NotImplementedError(
            "Concurrent decorator not supported for '%s' method." % fn.__name__
        )


            

Reported by Pylint.

                  name = "ScriptThread"


def concurrent(fn):
    if fn.__name__ not in set(hooks.all_hooks.keys()) - {"load", "configure"}:
        raise NotImplementedError(
            "Concurrent decorator not supported for '%s' method." % fn.__name__
        )


            

Reported by Pylint.

                  If another event is passed, an AssertionError is raised.
    """

    def decorator(f):
        if __debug__ is True:
            @functools.wraps(f)
            def _check_event_type(self, event: events.Event):
                if isinstance(event, event_types):
                    return f(self, event)

            

Reported by Pylint.

                  """

    def decorator(f):
        if __debug__ is True:
            @functools.wraps(f)
            def _check_event_type(self, event: events.Event):
                if isinstance(event, event_types):
                    return f(self, event)
                else:

            

Reported by Pylint.

                      if __debug__ is True:
            @functools.wraps(f)
            def _check_event_type(self, event: events.Event):
                if isinstance(event, event_types):
                    return f(self, event)
                else:
                    event_types_str = '|'.join(e.__name__ for e in event_types) or "no events"
                    raise AssertionError(
                        f"Unexpected event type at {f.__qualname__}: "

            

Reported by Pylint.

              import re
import sys


def lookup(address, port, s):
    """
        Parse the pfctl state output s, to look up the destination host
        matching the client (address, port).


            

Reported by Pylint.

              import sys


def lookup(address, port, s):
    """
        Parse the pfctl state output s, to look up the destination host
        matching the client (address, port).

        Returns an (address, port) tuple, or None.

            

Reported by Pylint.

                  # ALL tcp 192.168.1.13:57474 -> 23.205.82.58:443       ESTABLISHED:ESTABLISHED
    specv4 = f"{address}:{port}"

    # ALL tcp 2a01:e35:8bae:50f0:9d9b:ef0d:2de3:b733[58505] -> 2606:4700:30::681f:4ad0[443]       ESTABLISHED:ESTABLISHED
    specv6 = f"{address}[{port}]"

    for i in s.split("\n"):
        if "ESTABLISHED:ESTABLISHED" in i and specv4 in i:
            s = i.split()

            

Reported by Pylint.

              from .read import (
    read_request_head,
    read_response_head,
    connection_close,
    expected_http_body_size,
)
from .assemble import (
    assemble_request, assemble_request_head,
    assemble_response, assemble_response_head,

            

Reported by Pylint.

                  connection_close,
    expected_http_body_size,
)
from .assemble import (
    assemble_request, assemble_request_head,
    assemble_response, assemble_response_head,
    assemble_body,
)


            

Reported by Pylint.

              from .read import (
    read_request_head,
    read_response_head,
    connection_close,
    expected_http_body_size,
)
from .assemble import (
    assemble_request, assemble_request_head,
    assemble_response, assemble_response_head,

            

Reported by Pylint.