from collections.abc import MutableMapping
from typing import Any, Dict, Generator, List, TextIO, Callable

from mitmproxy import flowfilter
from mitmproxy.http import HTTPFlow


def f_id(x):
    return x

            

Reported by Pylint.

              from typing import Any, Dict, Generator, List, TextIO, Callable

from mitmproxy import flowfilter
from mitmproxy.http import HTTPFlow


def f_id(x):
    return x


            

Reported by Pylint.

                          if flowfilter.match(fltr, flow):
                yield value

    def get(self, flow: HTTPFlow, default=None, *, count=0) -> List[Any]:
        try:
            return self.__getitem__(flow, count=count)
        except KeyError:
            return default


            

Reported by Pylint.

              import itertools
import json
import typing
from collections.abc import MutableMapping
from typing import Any, Dict, Generator, List, TextIO, Callable

from mitmproxy import flowfilter
from mitmproxy.http import HTTPFlow


            

Reported by Pylint.

              from mitmproxy.http import HTTPFlow


def f_id(x):
    return x


class URLDict(MutableMapping):
    """Data structure to store information using filters as keys."""

            

Reported by Pylint.

              from mitmproxy.http import HTTPFlow


def f_id(x):
    return x


class URLDict(MutableMapping):
    """Data structure to store information using filters as keys."""

            

Reported by Pylint.

                      else:
            ret = list(self.get_generator(key))

        if ret:
            return ret
        else:
            raise KeyError

    def __setitem__(self, key: str, value):

            

Reported by Pylint.

                  def __len__(self):
        return self.store.__len__()

    def get_generator(self, flow: HTTPFlow) -> Generator[Any, None, None]:

        for fltr, value in self.store.items():
            if flowfilter.match(fltr, flow):
                yield value


            

Reported by Pylint.

                      return url_dict

    @classmethod
    def load(cls, f: TextIO, value_loader: Callable = f_id):
        json_obj = json.load(f)
        return cls._load(json_obj, value_loader)

    @classmethod
    def loads(cls, json_str: str, value_loader: Callable = f_id):

            

Reported by Pylint.

                      return url_dict

    @classmethod
    def load(cls, f: TextIO, value_loader: Callable = f_id):
        json_obj = json.load(f)
        return cls._load(json_obj, value_loader)

    @classmethod
    def loads(cls, json_str: str, value_loader: Callable = f_id):

            

Reported by Pylint.

              from mitmproxy.tools.console.grideditor import col_bytes


class Column(col_bytes.Column):
    def __init__(self, heading, encoding="utf8", errors="surrogateescape"):
        super().__init__(heading)
        self.encoding_args = encoding, errors

    def Display(self, data):

            

Reported by Pylint.

                      super().__init__(heading)
        self.encoding_args = encoding, errors

    def Display(self, data):
        return TDisplay(data, self.encoding_args)

    def Edit(self, data):
        return TEdit(data, self.encoding_args)


            

Reported by Pylint.

                      super().__init__(heading)
        self.encoding_args = encoding, errors

    def Display(self, data):
        return TDisplay(data, self.encoding_args)

    def Edit(self, data):
        return TEdit(data, self.encoding_args)


            

Reported by Pylint.

                  def Display(self, data):
        return TDisplay(data, self.encoding_args)

    def Edit(self, data):
        return TEdit(data, self.encoding_args)

    def blank(self):
        return ""


            

Reported by Pylint.

                  def Display(self, data):
        return TDisplay(data, self.encoding_args)

    def Edit(self, data):
        return TEdit(data, self.encoding_args)

    def blank(self):
        return ""


            

Reported by Pylint.

                  def Edit(self, data):
        return TEdit(data, self.encoding_args)

    def blank(self):
        return ""


# This is the same for both edit and display.
class EncodingMixin:

            

Reported by Pylint.

                  def Edit(self, data):
        return TEdit(data, self.encoding_args)

    def blank(self):
        return ""


# This is the same for both edit and display.
class EncodingMixin:

            

Reported by Pylint.

              

# This is the same for both edit and display.
class EncodingMixin:
    def __init__(self, data, encoding_args):
        self.encoding_args = encoding_args
        super().__init__(data.__str__().encode(*self.encoding_args))

    def get_data(self):

            

Reported by Pylint.

              

# This is the same for both edit and display.
class EncodingMixin:
    def __init__(self, data, encoding_args):
        self.encoding_args = encoding_args
        super().__init__(data.__str__().encode(*self.encoding_args))

    def get_data(self):

            

Reported by Pylint.

                      self.encoding_args = encoding_args
        super().__init__(data.__str__().encode(*self.encoding_args))

    def get_data(self):
        data = super().get_data()
        try:
            return data.decode(*self.encoding_args)
        except ValueError:
            signals.status_message.send(

            

Reported by Pylint.

              
    unknown_test_files = check_test_files_have_src()
    if unknown_test_files:
        # TODO: enable this in the future
        # exitcode += 1
        for f, p in sorted(unknown_test_files):
            print(f"{f} DOES NOT MATCH a source file! Expected to find: {p}")

    sys.exit(exitcode)

            

Reported by Pylint.

              #!/usr/bin/env python3

import os
import re
import glob
import sys


def check_src_files_have_test():

            

Reported by Pylint.

              import sys


def check_src_files_have_test():
    missing_test_files = []

    excluded = [
        'mitmproxy/contrib/',
        'mitmproxy/io/proto/',

            

Reported by Pylint.

                  src_files = glob.glob('mitmproxy/**/*.py', recursive=True)
    src_files = [f for f in src_files if os.path.basename(f) != '__init__.py']
    src_files = [f for f in src_files if not any(os.path.normpath(p) in f for p in excluded)]
    for f in src_files:
        p = os.path.join("test", os.path.dirname(f), "test_" + os.path.basename(f))
        if not os.path.isfile(p):
            missing_test_files.append((f, p))

    return missing_test_files

            

Reported by Pylint.

                  src_files = [f for f in src_files if os.path.basename(f) != '__init__.py']
    src_files = [f for f in src_files if not any(os.path.normpath(p) in f for p in excluded)]
    for f in src_files:
        p = os.path.join("test", os.path.dirname(f), "test_" + os.path.basename(f))
        if not os.path.isfile(p):
            missing_test_files.append((f, p))

    return missing_test_files


            

Reported by Pylint.

                  return missing_test_files


def check_test_files_have_src():
    unknown_test_files = []

    excluded = [
        'test/mitmproxy/data/',
        'test/mitmproxy/net/data/',

            

Reported by Pylint.

                  test_files = glob.glob('test/mitmproxy/**/*.py', recursive=True)
    test_files = [f for f in test_files if os.path.basename(f) != '__init__.py']
    test_files = [f for f in test_files if not any(os.path.normpath(p) in f for p in excluded)]
    for f in test_files:
        p = os.path.join(re.sub('^test/', '', os.path.dirname(f)), re.sub('^test_', '', os.path.basename(f)))
        if not os.path.isfile(p):
            unknown_test_files.append((f, p))

    return unknown_test_files

            

Reported by Pylint.

                  test_files = [f for f in test_files if os.path.basename(f) != '__init__.py']
    test_files = [f for f in test_files if not any(os.path.normpath(p) in f for p in excluded)]
    for f in test_files:
        p = os.path.join(re.sub('^test/', '', os.path.dirname(f)), re.sub('^test_', '', os.path.basename(f)))
        if not os.path.isfile(p):
            unknown_test_files.append((f, p))

    return unknown_test_files


            

Reported by Pylint.

                  test_files = [f for f in test_files if os.path.basename(f) != '__init__.py']
    test_files = [f for f in test_files if not any(os.path.normpath(p) in f for p in excluded)]
    for f in test_files:
        p = os.path.join(re.sub('^test/', '', os.path.dirname(f)), re.sub('^test_', '', os.path.basename(f)))
        if not os.path.isfile(p):
            unknown_test_files.append((f, p))

    return unknown_test_files


            

Reported by Pylint.

                  return unknown_test_files


def main():
    exitcode = 0

    missing_test_files = check_src_files_have_test()
    if missing_test_files:
        exitcode += 1

            

Reported by Pylint.

              import pytest

from mitmproxy.contentviews.image import image_parser


@pytest.mark.parametrize("filename, metadata", {
    # no textual data
    "mitmproxy/data/image_parser/ct0n0g04.png": [
        ('Format', 'Portable network graphics'),

            

Reported by Pylint.

              import pytest

from mitmproxy.contentviews.image import image_parser


@pytest.mark.parametrize("filename, metadata", {
    # no textual data
    "mitmproxy/data/image_parser/ct0n0g04.png": [
        ('Format', 'Portable network graphics'),

            

Reported by Pylint.

                      ('aspect', '72 x 72'),
        ('date:create', '2012-07-11T14:04:52-07:00'),
        ('date:modify', '2012-07-11T14:04:52-07:00')
    ],
}.items())
def test_parse_png(filename, metadata, tdata):
    with open(tdata.path(filename), "rb") as f:
        assert metadata == image_parser.parse_png(f.read())


            

Reported by Pylint.

                  ],
}.items())
def test_parse_png(filename, metadata, tdata):
    with open(tdata.path(filename), "rb") as f:
        assert metadata == image_parser.parse_png(f.read())


@pytest.mark.parametrize("filename, metadata", {
    # check comment

            

Reported by Pylint.

              }.items())
def test_parse_png(filename, metadata, tdata):
    with open(tdata.path(filename), "rb") as f:
        assert metadata == image_parser.parse_png(f.read())


@pytest.mark.parametrize("filename, metadata", {
    # check comment
    "mitmproxy/data/image_parser/hopper.gif": [

            

Reported by Bandit.

                      ('Version', 'GIF89a'),
        ('Size', '245 x 245 px'),
        ('background', '0')
    ],
}.items())
def test_parse_gif(filename, metadata, tdata):
    with open(tdata.path(filename), 'rb') as f:
        assert metadata == image_parser.parse_gif(f.read())


            

Reported by Pylint.

                  ],
}.items())
def test_parse_gif(filename, metadata, tdata):
    with open(tdata.path(filename), 'rb') as f:
        assert metadata == image_parser.parse_gif(f.read())


@pytest.mark.parametrize("filename, metadata", {
    # check app0

            

Reported by Pylint.

              }.items())
def test_parse_gif(filename, metadata, tdata):
    with open(tdata.path(filename), 'rb') as f:
        assert metadata == image_parser.parse_gif(f.read())


@pytest.mark.parametrize("filename, metadata", {
    # check app0
    "mitmproxy/data/image_parser/example.jpg": [

            

Reported by Bandit.

                       '\\r\\nTitle: Inferno, from the Divine Comedy by Dante (Folio 1v)\\r\\nTime-line:'
         ' 1401-1450\\r\\nSchool: Italian\\r\\nForm: illumination\\r\\nType: other\\r\\n\''),
        ('Size', '750 x 1055 px')
    ],
}.items())
def test_parse_jpeg(filename, metadata, tdata):
    with open(tdata.path(filename), 'rb') as f:
        assert metadata == image_parser.parse_jpeg(f.read())


            

Reported by Pylint.

                  ],
}.items())
def test_parse_jpeg(filename, metadata, tdata):
    with open(tdata.path(filename), 'rb') as f:
        assert metadata == image_parser.parse_jpeg(f.read())


@pytest.mark.parametrize("filename, metadata", {
    "mitmproxy/data/image.ico": [

            

Reported by Pylint.

              import multiprocessing
import configparser
import itertools
import pytest


def run_tests(src, test, fail):
    stderr = io.StringIO()
    stdout = io.StringIO()

            

Reported by Pylint.

              
    if any(e != 0 for _, _, e in result):
        sys.exit(1)
        pass


if __name__ == '__main__':
    main()

            

Reported by Pylint.

              #!/usr/bin/env python3

import io
import contextlib
import os
import sys
import glob
import multiprocessing
import configparser

            

Reported by Pylint.

              import pytest


def run_tests(src, test, fail):
    stderr = io.StringIO()
    stdout = io.StringIO()
    with contextlib.redirect_stderr(stderr):
        with contextlib.redirect_stdout(stdout):
            e = pytest.main([

            

Reported by Pylint.

                  stdout = io.StringIO()
    with contextlib.redirect_stderr(stderr):
        with contextlib.redirect_stdout(stdout):
            e = pytest.main([
                '-qq',
                '--disable-pytest-warnings',
                '--cov', src.replace('.py', '').replace('/', '.'),
                '--cov-fail-under', '100',
                '--cov-report', 'term-missing:skip-covered',

            

Reported by Pylint.

              
    if e == 0:
        if fail:
            print("FAIL DUE TO UNEXPECTED SUCCESS:", src, "Please remove this file from setup.cfg tool:individual_coverage/exclude.")
            e = 42
        else:
            print(".")
    else:
        if fail:

            

Reported by Pylint.

                  if e == 0:
        if fail:
            print("FAIL DUE TO UNEXPECTED SUCCESS:", src, "Please remove this file from setup.cfg tool:individual_coverage/exclude.")
            e = 42
        else:
            print(".")
    else:
        if fail:
            print("Ignoring allowed fail:", src)

            

Reported by Pylint.

                  else:
        if fail:
            print("Ignoring allowed fail:", src)
            e = 0
        else:
            cov = [l for l in stdout.getvalue().split("\n") if (src in l) or ("was never imported" in l)]
            if len(cov) == 1:
                print("FAIL:", cov[0])
            else:

            

Reported by Pylint.

                          print("Ignoring allowed fail:", src)
            e = 0
        else:
            cov = [l for l in stdout.getvalue().split("\n") if (src in l) or ("was never imported" in l)]
            if len(cov) == 1:
                print("FAIL:", cov[0])
            else:
                print("FAIL:", src, test, stdout.getvalue(), stdout.getvalue())
                print(stderr.getvalue())

            

Reported by Pylint.

                  sys.exit(e)


def start_pytest(src, test, fail):
    # run pytest in a new process, otherwise imports and modules might conflict
    proc = multiprocessing.Process(target=run_tests, args=(src, test, fail))
    proc.start()
    proc.join()
    return (src, test, proc.exitcode)

            

Reported by Pylint.

              import pyparsing
import pytest
from hypothesis import given, example
from hypothesis.strategies import text

from mitmproxy import command_lexer


@pytest.mark.parametrize(

            

Reported by Pylint.

              import pyparsing
import pytest
from hypothesis import given, example
from hypothesis.strategies import text

from mitmproxy import command_lexer


@pytest.mark.parametrize(

            

Reported by Pylint.

              import pyparsing
import pytest
from hypothesis import given, example
from hypothesis.strategies import text

from mitmproxy import command_lexer


@pytest.mark.parametrize(

            

Reported by Pylint.

              import pyparsing
import pytest
from hypothesis import given, example
from hypothesis.strategies import text

from mitmproxy import command_lexer


@pytest.mark.parametrize(

            

Reported by Pylint.

                      ('''"foo    ''', True),
        ('''"foo 'bar'   ''', True),
        ('"foo\\', True),
    ]
)
def test_partial_quoted_string(test_input, valid):
    if valid:
        assert command_lexer.PartialQuotedString.parseString(test_input, parseAll=True)[0] == test_input
    else:

            

Reported by Pylint.

              )
def test_partial_quoted_string(test_input, valid):
    if valid:
        assert command_lexer.PartialQuotedString.parseString(test_input, parseAll=True)[0] == test_input
    else:
        with pytest.raises(pyparsing.ParseException):
            command_lexer.PartialQuotedString.parseString(test_input, parseAll=True)



            

Reported by Bandit.

              )
def test_partial_quoted_string(test_input, valid):
    if valid:
        assert command_lexer.PartialQuotedString.parseString(test_input, parseAll=True)[0] == test_input
    else:
        with pytest.raises(pyparsing.ParseException):
            command_lexer.PartialQuotedString.parseString(test_input, parseAll=True)



            

Reported by Pylint.

                      ('''"foo''', ['"foo']),
        ('''"foo 'bar' ''', ['''"foo 'bar' ''']),
        ('"foo\\', ['"foo\\']),
    ]
)
def test_expr(test_input, expected):
    assert list(command_lexer.expr.parseString(test_input, parseAll=True)) == expected



            

Reported by Pylint.

                  ]
)
def test_expr(test_input, expected):
    assert list(command_lexer.expr.parseString(test_input, parseAll=True)) == expected


@given(text())
@example(r"foo")
@example(r"'foo\''")

            

Reported by Bandit.

              @example("\"foo\\\\'\"")
@example('\'foo\\"\'')
@example(r"\\\foo")
def test_quote_unquote_cycle(s):
    assert command_lexer.unquote(command_lexer.quote(s)).replace(r"\x22", '"') == s


@given(text())
@example("'foo\\'")

            

Reported by Pylint.

              from mitmproxy import log


def test_logentry():
    e = log.LogEntry("foo", "info")
    assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e

            

Reported by Pylint.

              from mitmproxy import log


def test_logentry():
    e = log.LogEntry("foo", "info")
    assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e

            

Reported by Pylint.

              

def test_logentry():
    e = log.LogEntry("foo", "info")
    assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f

            

Reported by Pylint.

              
def test_logentry():
    e = log.LogEntry("foo", "info")
    assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42

            

Reported by Bandit.

                  e = log.LogEntry("foo", "info")
    assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42



            

Reported by Pylint.

                  assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42


def test_dont_pick_up_mutations():

            

Reported by Pylint.

                  assert repr(e) == "LogEntry(foo, info)"

    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42


def test_dont_pick_up_mutations():

            

Reported by Bandit.

              
    f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42


def test_dont_pick_up_mutations():
    x = {"foo": "bar"}

            

Reported by Bandit.

                  f = log.LogEntry("foo", "warning")
    assert e == e
    assert e != f
    assert e != 42


def test_dont_pick_up_mutations():
    x = {"foo": "bar"}
    e = log.LogEntry(x, "info")

            

Reported by Bandit.

                  assert e != 42


def test_dont_pick_up_mutations():
    x = {"foo": "bar"}
    e = log.LogEntry(x, "info")
    x["foo"] = "baz"  # this should not affect the log entry anymore.
    assert repr(e) == "LogEntry({'foo': 'bar'}, info)"

            

Reported by Pylint.

              import collections

from mitmproxy.net.http.headers import parse_content_type, assemble_content_type


def test_parse_content_type():
    p = parse_content_type
    assert p("text/html") == ("text", "html", {})
    assert p("text") is None

            

Reported by Pylint.

              from mitmproxy.net.http.headers import parse_content_type, assemble_content_type


def test_parse_content_type():
    p = parse_content_type
    assert p("text/html") == ("text", "html", {})
    assert p("text") is None

    v = p("text/html; charset=UTF-8")

            

Reported by Pylint.

              

def test_parse_content_type():
    p = parse_content_type
    assert p("text/html") == ("text", "html", {})
    assert p("text") is None

    v = p("text/html; charset=UTF-8")
    assert v == ('text', 'html', {'charset': 'UTF-8'})

            

Reported by Pylint.

              
def test_parse_content_type():
    p = parse_content_type
    assert p("text/html") == ("text", "html", {})
    assert p("text") is None

    v = p("text/html; charset=UTF-8")
    assert v == ('text', 'html', {'charset': 'UTF-8'})


            

Reported by Bandit.

              def test_parse_content_type():
    p = parse_content_type
    assert p("text/html") == ("text", "html", {})
    assert p("text") is None

    v = p("text/html; charset=UTF-8")
    assert v == ('text', 'html', {'charset': 'UTF-8'})



            

Reported by Bandit.

                  assert p("text/html") == ("text", "html", {})
    assert p("text") is None

    v = p("text/html; charset=UTF-8")
    assert v == ('text', 'html', {'charset': 'UTF-8'})


def test_assemble_content_type():
    p = assemble_content_type

            

Reported by Pylint.

                  assert p("text") is None

    v = p("text/html; charset=UTF-8")
    assert v == ('text', 'html', {'charset': 'UTF-8'})


def test_assemble_content_type():
    p = assemble_content_type
    assert p("text", "html", {}) == "text/html"

            

Reported by Bandit.

                  assert v == ('text', 'html', {'charset': 'UTF-8'})


def test_assemble_content_type():
    p = assemble_content_type
    assert p("text", "html", {}) == "text/html"
    assert p("text", "html", {"charset": "utf8"}) == "text/html; charset=utf8"
    assert p("text", "html",
             collections.OrderedDict([("charset", "utf8"), ("foo", "bar")])) == "text/html; charset=utf8; foo=bar"

            

Reported by Pylint.

              

def test_assemble_content_type():
    p = assemble_content_type
    assert p("text", "html", {}) == "text/html"
    assert p("text", "html", {"charset": "utf8"}) == "text/html; charset=utf8"
    assert p("text", "html",
             collections.OrderedDict([("charset", "utf8"), ("foo", "bar")])) == "text/html; charset=utf8; foo=bar"

            

Reported by Pylint.

              
def test_assemble_content_type():
    p = assemble_content_type
    assert p("text", "html", {}) == "text/html"
    assert p("text", "html", {"charset": "utf8"}) == "text/html; charset=utf8"
    assert p("text", "html",
             collections.OrderedDict([("charset", "utf8"), ("foo", "bar")])) == "text/html; charset=utf8; foo=bar"

            

Reported by Bandit.

              import re
import typing

from mitmproxy import ctx, exceptions
from mitmproxy.addons.modifyheaders import parse_modify_spec, ModifySpec


class ModifyBody:
    def __init__(self):

            

Reported by Pylint.

              from mitmproxy.addons.modifyheaders import parse_modify_spec, ModifySpec


class ModifyBody:
    def __init__(self):
        self.replacements: typing.List[ModifySpec] = []

    def load(self, loader):
        loader.add_option(

            

Reported by Pylint.

                  def __init__(self):
        self.replacements: typing.List[ModifySpec] = []

    def load(self, loader):
        loader.add_option(
            "modify_body", typing.Sequence[str], [],
            """
            Replacement pattern of the form "[/flow-filter]/regex/[@]replacement", where
            the separator can be any character. The @ allows to provide a file path that

            

Reported by Pylint.

                  def __init__(self):
        self.replacements: typing.List[ModifySpec] = []

    def load(self, loader):
        loader.add_option(
            "modify_body", typing.Sequence[str], [],
            """
            Replacement pattern of the form "[/flow-filter]/regex/[@]replacement", where
            the separator can be any character. The @ allows to provide a file path that

            

Reported by Pylint.

                          """
        )

    def configure(self, updated):
        if "modify_body" in updated:
            self.replacements = []
            for option in ctx.options.modify_body:
                try:
                    spec = parse_modify_spec(option, True)

            

Reported by Pylint.

                          for option in ctx.options.modify_body:
                try:
                    spec = parse_modify_spec(option, True)
                except ValueError as e:
                    raise exceptions.OptionsError(f"Cannot parse modify_body option {option}: {e}") from e

                self.replacements.append(spec)

    def request(self, flow):

            

Reported by Pylint.

                              try:
                    spec = parse_modify_spec(option, True)
                except ValueError as e:
                    raise exceptions.OptionsError(f"Cannot parse modify_body option {option}: {e}") from e

                self.replacements.append(spec)

    def request(self, flow):
        if flow.response or flow.error or flow.reply.state == "taken":

            

Reported by Pylint.

              
                self.replacements.append(spec)

    def request(self, flow):
        if flow.response or flow.error or flow.reply.state == "taken":
            return
        self.run(flow)

    def response(self, flow):

            

Reported by Pylint.

                          return
        self.run(flow)

    def response(self, flow):
        if flow.error or flow.reply.state == "taken":
            return
        self.run(flow)

    def run(self, flow):

            

Reported by Pylint.

                          return
        self.run(flow)

    def run(self, flow):
        for spec in self.replacements:
            if spec.matches(flow):
                try:
                    replacement = spec.read_replacement()
                except OSError as e:

            

Reported by Pylint.

              import datetime
import functools
import ipaddress
import time
import typing


SIZE_UNITS = {
    "b": 1024 ** 0,

            

Reported by Pylint.

              
    len(return value) <= 5 always holds true.
    """
    s: float = size  # type cast for mypy
    if s < 1024:
        return f"{s}b"
    for suffix in ["k", "m", "g", "t"]:
        s /= 1024
        if s < 99.95:

            

Reported by Pylint.

                  if s < 1024:
        return f"{s}b"
    for suffix in ["k", "m", "g", "t"]:
        s /= 1024
        if s < 99.95:
            return f"{s:.1f}{suffix}"
        if s < 1024 or suffix == "t":
            return f"{s:.0f}{suffix}"
    raise AssertionError

            

Reported by Pylint.

              

@functools.lru_cache()
def parse_size(s: typing.Optional[str]) -> typing.Optional[int]:
    """
    Parse a size with an optional k/m/... suffix.
    Invalid values raise a ValueError. For added convenience, passing `None` returns `None`.
    """
    if s is None:

            

Reported by Pylint.

                      return int(s)
    except ValueError:
        pass
    for i in SIZE_UNITS.keys():
        if s.endswith(i):
            try:
                return int(s[:-1]) * SIZE_UNITS[i]
            except ValueError:
                break

            

Reported by Pylint.

                  raise ValueError("Invalid size specification.")


def pretty_duration(secs: typing.Optional[float]) -> str:
    formatters = [
        (100, "{:.0f}s"),
        (10, "{:2.1f}s"),
        (1, "{:1.2f}s"),
    ]

            

Reported by Pylint.

                  return "{:.0f}ms".format(secs * 1000)


def format_timestamp(s):
    s = time.localtime(s)
    d = datetime.datetime.fromtimestamp(time.mktime(s))
    return d.strftime("%Y-%m-%d %H:%M:%S")



            

Reported by Pylint.

                  return "{:.0f}ms".format(secs * 1000)


def format_timestamp(s):
    s = time.localtime(s)
    d = datetime.datetime.fromtimestamp(time.mktime(s))
    return d.strftime("%Y-%m-%d %H:%M:%S")



            

Reported by Pylint.

              
def format_timestamp(s):
    s = time.localtime(s)
    d = datetime.datetime.fromtimestamp(time.mktime(s))
    return d.strftime("%Y-%m-%d %H:%M:%S")


def format_timestamp_with_milli(s):
    d = datetime.datetime.fromtimestamp(s)

            

Reported by Pylint.

                  return d.strftime("%Y-%m-%d %H:%M:%S")


def format_timestamp_with_milli(s):
    d = datetime.datetime.fromtimestamp(s)
    return d.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]


@functools.lru_cache()

            

Reported by Pylint.