Function do_repo_list has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_repo_list(self, args, repository, manifest):
        """List the archives contained in a repository"""
        if args.format is not None:
            format = args.format
        elif args.short:

• Read up
  Read up

Function analyze_change has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def analyze_change(self, base, new):
        """for each directory path, sum up the changed (removed or added) chunks' sizes between base and new."""

        def analyze_path_change(path):
            base_chunks = base[path]

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if not dry_run:
                            item.chunks = chunks  # TODO: overwrite? IDs and sizes are same.
                            archive.stats.nfiles += 1
                    if not dry_run:

Function get_repository has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def get_repository(
    location, *, create, exclusive, lock_wait, lock, append_only, make_parent_dirs, storage_quota, args, v1_or_v2
):
    if location.proto in ("ssh", "socket"):
        RemoteRepoCls = LegacyRemoteRepository if v1_or_v2 else RemoteRepository

• Read up
  Read up

Function define_archive_filters_group has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def define_archive_filters_group(subparser, *, sort_by=True, first_last=True, oldest_newest=True, older_newer=True):

Avoid deeply nested control flow statements.

Open

                    if not is_win32:
                        # The S_IFBLK macro is broken on MINGW
                        assert stat.S_ISBLK(item.mode)
                    # looks like we can't use os.major/minor with data coming from another platform,
                    # thus we only do a rather rough check here:

Avoid deeply nested control flow statements.

Open

                            if not recurse_excluded_dir:
                                if keep_exclude_tags:
                                    if not dry_run:
                                        fso.process_dir_with_fd(
                                            path=path, fd=child_fd, st=st, strip_prefix=strip_prefix

Function do_help has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_help(self, parser, commands, args):
        if not args.topic:
            parser.print_help()
        elif args.topic in self.helptext:
            print(rst_to_terminal(self.helptext[args.topic]))

• Read up
  Read up

Function fuse_mount has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def fuse_mount(archiver, mountpoint=None, *options, fork=True, os_fork=False, **kwargs):

Function print_warning has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def print_warning(self, msg, *args, **kw):
        warning_code = kw.get("wc", EXIT_WARNING)  # note: wc=None can be used to not influence exit code
        warning_type = kw.get("wt", "percent")
        assert warning_type in ("percent", "curly")
        warning_msgid = kw.get("msgid")

• Read up
  Read up

Function test_stats_progress has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def test_stats_progress(item_path, update_size, expected_output, stats, monkeypatch, columns=80):

Avoid deeply nested control flow statements.

Open

                        for chunk_id, size in item.chunks:
                            chunk_present = cache.seen_chunk(chunk_id, size)
                            if not chunk_present:  # target repo does not yet have this chunk
                                if not dry_run:
                                    cdata = other_repository.get(chunk_id)

Avoid deeply nested control flow statements.

Open

                        for id, size in item.chunks_healthy:
                            use_it(id, wanted=True)
        pi.finish()

Function do_key_import has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def do_key_import(self, args, repository):
        """Import the repository key from backup"""
        manager = KeyManager(repository)
        if args.paper:
            if args.path:

• Read up
  Read up

Function _import_tar has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def _import_tar(self, args, repository, manifest, key, cache, tarstream):

Function parse has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

    def parse(
        self, id: bytes, cdata: bytes, decompress: bool = True, want_compressed: bool = False, ro_type: str = None
    ) -> tuple[dict, bytes]:
        """
        Parse a repo object into metadata and data (decrypt it, maybe decompress, maybe verify if the chunk plaintext

• Read up
  Read up

Function parse_inclexcl_command has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def parse_inclexcl_command(cmd_line_str, fallback=ShellPattern):
    """Read a --patterns-from command from string and return a CmdTuple object."""

    cmd_prefix_map = {
        "-": IECommand.Exclude,

• Read up
  Read up

Function test_remote_rpc_exception_transport has a Cognitive Complexity of 8 (exceeds 5 allowed). Consider refactoring.

Open

def test_remote_rpc_exception_transport(remote_repository):
    with remote_repository:
        s1 = "test string"

        try:

• Read up
  Read up

Function write_chunkindex_to_repo_cache has 6 arguments (exceeds 4 allowed). Consider refactoring.

Open

def write_chunkindex_to_repo_cache(repository, chunks, *, compact=False, clear=False, force_write=False):

Avoid deeply nested control flow statements.

Open

                    for seed in (1849058162, 1234567653):
                        fh = BytesIO(data)
                        chunker = Chunker(seed, minexp, maxexp, maskbits, winsize)
                        chunks = [blake2b_256(b"", c) for c in cf(chunker.chunkify(fh, -1))]
                        runs.append(blake2b_256(b"", b"".join(chunks)))