File ynl.py has 767 lines of code (exceeds 250 allowed). Consider refactoring.

Open

# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause

from collections import namedtuple
import functools
import os

Function _genl_load_families has a Cognitive Complexity of 47 (exceeds 5 allowed). Consider refactoring.

Open

def _genl_load_families():
    with socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, Netlink.NETLINK_GENERIC) as sock:
        sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_CAP_ACK, 1)

        msg = _genl_msg(Netlink.GENL_ID_CTRL,

• Read up
  Read up

Function _add_attr has a Cognitive Complexity of 41 (exceeds 5 allowed). Consider refactoring.

Open

    def _add_attr(self, space, name, value, search_attrs):
        try:
            attr = self.attr_sets[space][name]
        except KeyError:
            raise Exception(f"Space '{space}' has no attribute '{name}'")

• Read up
  Read up

Function _op has a Cognitive Complexity of 35 (exceeds 5 allowed). Consider refactoring.

Open

    def _op(self, method, vals, flags=None, dump=False):
        op = self.ops[method]

        nl_flags = Netlink.NLM_F_REQUEST | Netlink.NLM_F_ACK
        for flag in flags or []:

• Read up
  Read up

Function __init__ has a Cognitive Complexity of 30 (exceeds 5 allowed). Consider refactoring.

Open

    def __init__(self, msg, offset, attr_space=None):
        self.hdr = msg[offset : offset + 16]

        self.nl_len, self.nl_type, self.nl_flags, self.nl_seq, self.nl_portid = \
            struct.unpack("IHHII", self.hdr)

• Read up
  Read up

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

Open

    def _decode(self, attrs, space, outer_attrs = None):
        rsp = dict()
        if space:
            attr_space = self.attr_sets[space]
            search_attrs = SpaceAttrs(attr_space, rsp, outer_attrs)

• Read up
  Read up

YnlFamily has 28 functions (exceeds 20 allowed). Consider refactoring.

Open

class YnlFamily(SpecFamily):
    def __init__(self, def_path, schema=None, process_unknown=False,
                 recv_size=0):
        super().__init__(def_path, schema)

Function _encode_struct has a Cognitive Complexity of 23 (exceeds 5 allowed). Consider refactoring.

Open

    def _encode_struct(self, name, vals):
        members = self.consts[name].members
        attr_payload = b''
        for m in members:
            value = vals.pop(m.name) if m.name in vals else None

• Read up
  Read up

Function _decode_struct has a Cognitive Complexity of 15 (exceeds 5 allowed). Consider refactoring.

Open

    def _decode_struct(self, data, name):
        members = self.consts[name].members
        attrs = dict()
        offset = 0
        for m in members:

• Read up
  Read up

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

Open

    def check_ntf(self):
        while True:
            try:
                reply = self.sock.recv(self._recv_size, socket.MSG_DONTWAIT)
            except BlockingIOError:

• Read up
  Read up

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

Open

    def _struct_size(self, name):
        if name:
            members = self.consts[name].members
            size = 0
            for m in members:

• Read up
  Read up

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

Open

    def _decode_extack_path(self, attrs, attr_set, offset, target):
        for attr in attrs:
            try:
                attr_spec = attr_set.attrs_by_val[attr.type]
            except KeyError:

• Read up
  Read up

Function __init__ has a Cognitive Complexity of 9 (exceeds 5 allowed). Consider refactoring.

Open

    def __init__(self, def_path, schema=None, process_unknown=False,
                 recv_size=0):
        super().__init__(def_path, schema)

        self.include_raw = False

• Read up
  Read up

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

Open

    def _decode_enum(self, raw, attr_spec):
        enum = self.consts[attr_spec['enum']]
        if enum.type == 'flags' or attr_spec.get('enum-as-flags', False):
            i = 0
            value = set()

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                        if 'doc' in spec:
                            desc += f" ({spec['doc']})"
                        self.extack['miss-type'] = desc

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

Open

    def _rsp_add(self, rsp, name, is_multi, decoded):
        if is_multi == None:
            if name in rsp and type(rsp[name]) is not list:
                rsp[name] = [rsp[name]]
                is_multi = True

• Read up
  Read up

Avoid deeply nested control flow statements.

Open

                    if attr.type == Netlink.CTRL_ATTR_FAMILY_ID:
                        fam['id'] = attr.as_scalar('u16')
                    elif attr.type == Netlink.CTRL_ATTR_FAMILY_NAME:
                        fam['name'] = attr.as_strz()
                    elif attr.type == Netlink.CTRL_ATTR_MAXATTR:

Function _genl_msg has 5 arguments (exceeds 4 allowed). Consider refactoring.

Open

def _genl_msg(nl_type, nl_flags, genl_cmd, genl_version, seq=None):

Function _encode_enum has a Cognitive Complexity of 7 (exceeds 5 allowed). Consider refactoring.

Open

    def _encode_enum(self, attr_spec, value):
        enum = self.consts[attr_spec['enum']]
        if enum.type == 'flags' or attr_spec.get('enum-as-flags', False):
            scalar = 0
            if isinstance(value, str):

• Read up
  Read up

Function lookup has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.

Open

    def lookup(self, name):
        for scope in self.scopes:
            if name in scope.spec:
                if name in scope.values:
                    return scope.values[name]

• Read up
  Read up