Function _read has a Cognitive Complexity of 28 (exceeds 5 allowed). Consider refactoring.

Open

    def _read(self, fd, header, segment, offset, acceptable_tags, read_data=True):
        """
        Code shared by read() and iter_objects().

        Confidence in returned data:

• Read up
  Read up

Function do_prune has a Cognitive Complexity of 27 (exceeds 5 allowed). Consider refactoring.

Open

    def do_prune(self, args, repository, manifest):
        """Prune repository archives according to specified rules"""
        if not any(
            (
                args.secondly,

• Read up
  Read up

File prune_cmd.py has 329 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import argparse
from collections import OrderedDict
from datetime import datetime, timezone, timedelta
import logging
from operator import attrgetter

File process.py has 328 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import contextlib
import os
import os.path
import shlex
import signal

File check_cmd_test.py has 325 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from datetime import datetime, timezone, timedelta
import shutil
from unittest.mock import patch

import pytest

Function create_helper has a Cognitive Complexity of 26 (exceeds 5 allowed). Consider refactoring.

Open

    def create_helper(self, tarinfo, status=None, type=None):
        ph = tarinfo.pax_headers
        if ph and "BORG.item.version" in ph:
            assert ph["BORG.item.version"] == "1"
            meta_bin = base64.b64decode(ph["BORG.item.meta"])

• Read up
  Read up

Repository has 30 functions (exceeds 20 allowed). Consider refactoring.

Open

class Repository:
    """borgstore based key value store"""

    class AlreadyExists(Error):
        """A repository already exists at {}."""

LegacyRemoteRepository has 30 functions (exceeds 20 allowed). Consider refactoring.

Open

class LegacyRemoteRepository:
    extra_test_args = []  # type: ignore

    class RPCError(Exception):
        def __init__(self, unpacked):

File mount_cmds_test.py has 317 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import errno
import os
import stat
import sys

Function unpack_many has a Cognitive Complexity of 25 (exceeds 5 allowed). Consider refactoring.

Open

    def unpack_many(self, ids, *, filter=None, preload=False):
        """
        Return iterator of items.

        *ids* is a chunk ID list of an item stream. *filter* is a callable

• Read up
  Read up

Function verify_data has a Cognitive Complexity of 25 (exceeds 5 allowed). Consider refactoring.

Open

    def verify_data(self):
        logger.info("Starting cryptographic data integrity verification...")
        chunks_count = len(self.chunks)
        errors = 0
        defect_chunks = []

• Read up
  Read up

Function iter_archive_items has a Cognitive Complexity of 25 (exceeds 5 allowed). Consider refactoring.

Open

    def iter_archive_items(self, archive_item_ids, filter=None):
        unpacker = msgpack.Unpacker()

        # Current offset in the metadata stream, which consists of all metadata chunks glued together
        stream_offset = 0

• Read up
  Read up

File fslocking_test.py has 311 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import random
import time
from threading import Thread, Lock as ThreadingLock
from traceback import format_exc

Function create_filter_process has a Cognitive Complexity of 24 (exceeds 5 allowed). Consider refactoring.

Open

def create_filter_process(cmd, stream, stream_close, inbound=True):
    if cmd:
        # put a filter process between stream and us (e.g. a [de]compression command)
        # inbound: <stream> --> filter --> us
        # outbound: us --> filter --> <stream>

• Read up
  Read up

File hashindex_test.py has 302 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# Note: these tests are part of the self test, do not use or import pytest functionality here.
#       See borg.selftest for details. If you add/remove test methods, update SELFTEST_COUNT

import base64
import hashlib

Function borg_cmd has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

    def borg_cmd(self, args, testing):
        """return a borg serve command line"""
        # give some args/options to 'borg serve' process as they were given to us
        opts = []
        if args is not None:

• Read up
  Read up

Function borg_cmd has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

    def borg_cmd(self, args, testing):
        """return a borg serve command line"""
        # give some args/options to 'borg serve' process as they were given to us
        opts = []
        if args is not None:

• Read up
  Read up

Function _assert_dirs_equal_cmp has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

def _assert_dirs_equal_cmp(diff, ignore_flags=False, ignore_xattrs=False, ignore_ns=False):
    assert diff.left_only == []
    assert diff.right_only == []
    assert diff.diff_files == []
    assert diff.funny_files == []

• Read up
  Read up

Function exec_cmd has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

def exec_cmd(*args, archiver=None, fork=False, exe=None, input=b"", binary_output=False, **kw):
    if fork:
        try:
            if exe is None:
                borg = (sys.executable, "-m", "borg")

• Read up
  Read up

Function _process_leaf has a Cognitive Complexity of 22 (exceeds 5 allowed). Consider refactoring.

Open

    def _process_leaf(self, name, item, parent, prefix, is_dir, item_inode, hlm):
        path = item.path
        del item.path  # save some space

        def file_version(item, path):

• Read up
  Read up