Function add_chunk has 11 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def add_chunk(

Function __new__ has 11 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __new__(

Function create has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def create(cls, repository, args, *, other_key=None):
        key = cls(repository)
        key.repository_id = repository.id
        if other_key is not None:
            if isinstance(other_key, PlaintextKey):

• Read up
  Read up

Function upgrade_archive_metadata has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def upgrade_archive_metadata(self, *, metadata):
        new_metadata = {}
        # keep all metadata except archive version and stats. also do not keep
        # recreate_source_id, recreate_args, recreate_partial_chunks which were used only in 1.1.0b1 .. b2.
        for attr in ("hostname", "username", "comment", "chunker_params"):

• Read up
  Read up

Function calculate_relative_offset has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def calculate_relative_offset(format_string, from_ts, earlier=False):
    """
    Calculates offset based on a relative marker. 7d (7 days), 8m (8 months)
    earlier: whether offset should be calculated to an earlier time.
    """

• Read up
  Read up

Function __init__ has 10 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(

Function __init__ has 10 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(

Function extract_helper has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def extract_helper(self, item, path, hlm, *, dry_run=False):
        hardlink_set = False
        # Hard link?
        if "hlid" in item:
            link_target = hlm.retrieve(id=item.hlid)

• Read up
  Read up

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

Open

    def create_helper(self, path, st, status=None, hardlinkable=True, strip_prefix=None):
        if strip_prefix is not None:
            assert not path.endswith(os.sep)
            if strip_prefix.startswith(path + os.sep):
                # still on a directory level that shall be stripped - do not create an item for this!

• Read up
  Read up

Function matcher_add_tagged_dirs has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def matcher_add_tagged_dirs(self, archive):
        """Add excludes to the matcher created by exclude_cache and exclude_if_present."""

        def exclude(dir, tag_item):
            if self.keep_exclude_tags:

• Read up
  Read up

Function open has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def open(self, path, exclusive, lock_wait=None, lock=True):
        self.path = path
        try:
            st = os.stat(path)
        except FileNotFoundError:

• Read up
  Read up

Function extract_item has 10 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def extract_item(

Function __init__ has 10 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def __init__(

Function fuse_mount has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def fuse_mount(archiver, mountpoint=None, *options, fork=True, os_fork=False, **kwargs):
    # For a successful mount, `fork = True` is required for
    # the borg mount daemon to work properly or the tests
    # will just freeze. Therefore, if argument `fork` is not
    # specified, the default value is `True`, regardless of

• Read up
  Read up

Function test_disk_full has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def test_disk_full(test_pass, cmd_fixture, monkeypatch):
    monkeypatch.setenv("BORG_CHECK_I_KNOW_WHAT_I_AM_DOING", "YES")
    monkeypatch.setenv("BORG_DELETE_I_KNOW_WHAT_I_AM_DOING", "YES")
    repo = os.path.join(DF_MOUNT, "repo")
    input = os.path.join(DF_MOUNT, "input")

• Read up
  Read up

Function _process_any has 10 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def _process_any(self, *, path, parent_fd, name, st, fso, cache, read_special, dry_run, strip_prefix):

Function do_repo_info has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def do_repo_info(self, args, repository, manifest, cache):
        """Show repository infos"""
        key = manifest.key
        info = basic_json_data(manifest, cache=cache, extra={"security_dir": cache.security_manager.dir})

• Read up
  Read up

Function compat_check has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def compat_check(*, create, manifest, key, cache, compatibility, decorator_name):
    if not create and (manifest or key or cache):
        if compatibility is None:
            raise AssertionError(f"{decorator_name} decorator used without compatibility argument")
        if type(compatibility) is not tuple:

• Read up
  Read up

Function test_migrate_lock_alive has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

def test_migrate_lock_alive(archivers, request):
    """Both old_id and new_id must not be stale during lock migration / daemonization."""
    archiver = request.getfixturevalue(archivers)
    if archiver.get_kind() == "remote":
        pytest.skip("only works locally")

• Read up
  Read up

Function get_args has a Cognitive Complexity of 11 (exceeds 5 allowed). Consider refactoring.

Open

    def get_args(self, argv, cmd):
        """usually, just returns argv, except if we deal with a ssh forced command for borg serve."""
        result = self.parse_args(argv[1:])
        if cmd is not None and result.func == self.do_serve:
            # borg serve case:

• Read up
  Read up