Function _matching_info_tuples has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def _matching_info_tuples(self, match_patterns, match_end):
        archive_infos = list(self._info_tuples())
        if match_patterns:
            assert isinstance(match_patterns, list), f"match_pattern is a {type(match_patterns)}"
            for match in match_patterns:

• Read up
  Read up

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

Open

    def __init__(

Function save_config has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def save_config(self, path, config):
        config_path = os.path.join(path, "config")
        old_config_path = os.path.join(path, "config.old")

        if os.path.isfile(old_config_path):

• Read up
  Read up

Function segment_iterator has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def segment_iterator(self, start_segment=None, end_segment=None, reverse=False):
        if start_segment is None:
            start_segment = MIN_SEGMENT_INDEX if not reverse else MAX_SEGMENT_INDEX
        if end_segment is None:
            end_segment = MAX_SEGMENT_INDEX if not reverse else MIN_SEGMENT_INDEX

• Read up
  Read up

Function is_committed_segment has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def is_committed_segment(self, segment):
        """Check if segment ends with a COMMIT_TAG tag"""
        try:
            iterator = self.iter_objects(segment)
        except IntegrityError:

• Read up
  Read up

Function check_can_create_repository has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

    def check_can_create_repository(self, path):
        """
        Raise an exception if a repository already exists at *path* or any parent directory.

        Checking parent directories is done for two reasons:

• Read up
  Read up

Function prune_split has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def prune_split(archives, rule, n, kept_because=None):
    last = None
    keep = []
    period_func = PRUNING_PATTERNS[rule]
    if kept_because is None:

• Read up
  Read up

Function test_sparse_file has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def test_sparse_file(archivers, request):
    archiver = request.getfixturevalue(archivers)

    def is_sparse(fn, total_size, hole_size):
        st = os.stat(fn)

• Read up
  Read up

Function create_test_files has a Cognitive Complexity of 14 (exceeds 5 allowed). Consider refactoring.

Open

def create_test_files(input_path, create_hardlinks=True):
    """Create a minimal test case including all supported file types"""
    # File
    create_regular_file(input_path, "file1", size=1024 * 80)
    create_regular_file(input_path, "flagfile", size=1024)

• Read up
  Read up

Function check has 13 arguments (exceeds 4 allowed). Consider refactoring.

Open

    def check(

Function test_get_item_uid_gid has 40 lines of code (exceeds 25 allowed). Consider refactoring.

Open

def test_get_item_uid_gid():
    # test requires that:
    # - a user/group name for the current process' real uid/gid exists.
    # - a system user/group udoesnotexist:gdoesnotexist does NOT exist.

Function get_many has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def get_many(self, keys, read_data=True, cache=True):
        # It could use different cache keys depending on read_data and cache full vs. meta-only chunks.
        unknown_keys = [key for key in keys if self.prefixed_key(key, complete=read_data) not in self.cache]
        repository_iterator = zip(unknown_keys, self.repository.get_many(unknown_keys, read_data=read_data))
        for key in keys:

• Read up
  Read up

Function list has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def list(self, limit=None, marker=None):
        """
        list <limit> infos starting from after id <marker>.
        each info is a tuple (id, storage_size).
        """

• Read up
  Read up

Function get_many has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def get_many(self, keys, read_data=True, cache=True):
        # It could use different cache keys depending on read_data and cache full vs. meta-only chunks.
        unknown_keys = [key for key in keys if self.prefixed_key(key, complete=read_data) not in self.cache]
        repository_iterator = zip(unknown_keys, self.repository.get_many(unknown_keys, read_data=read_data))
        for key in keys:

• Read up
  Read up

Function get_write_fd has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def get_write_fd(self, no_new=False, want_new=False, raise_full=False):
        if not no_new and (want_new or self.offset and self.offset > self.limit):
            if raise_full:
                raise self.SegmentFull
            self.close_segment()

• Read up
  Read up

Function _parse_braces has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

def _parse_braces(pat):
    """Returns the index values of paired braces in `pat` as a list of tuples.

    The dict's keys are the indexes corresponding to opening braces. Initially,
    they are set to a value of `None`. Once a corresponding closing brace is found,

• Read up
  Read up

Function add_common_group has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

        def add_common_group(self, parser, suffix, provide_defaults=False):
            """
            Add common options to *parser*.

            *provide_defaults* must only be True exactly once in a parser hierarchy,

• Read up
  Read up

Function do_debug_dump_archive has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def do_debug_dump_archive(self, args, repository, manifest):
        """dump decoded archive metadata (not: data)"""
        archive_info = manifest.archives.get_one([args.name])
        repo_objs = manifest.repo_objs
        try:

• Read up
  Read up

Function do_check has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def do_check(self, args, repository):
        """Check repository consistency"""
        if args.repair:
            msg = (
                "This is a potentially dangerous function.\n"

• Read up
  Read up

Function test_race_condition has a Cognitive Complexity of 13 (exceeds 5 allowed). Consider refactoring.

Open

    def test_race_condition(self, lockpath):
        class SynchronizedCounter:
            def __init__(self, count=0):
                self.lock = ThreadingLock()
                self.count = count

• Read up
  Read up