Function info has a Cognitive Complexity of 40 (exceeds 5 allowed). Consider refactoring.

Open

def info(csar_or_rootdir: Optional[PurePath], storage: Storage) -> dict:  # pylint: disable=too-many-statements
    info_dict: Dict[str, Optional[Union[str, dict, bool]]] = dict(
        service_template=None,
        content_root=None,
        inputs=None,

• Read up
  Read up

Function _parser_callback has a Cognitive Complexity of 39 (exceeds 5 allowed). Consider refactoring.

Open

def _parser_callback(args):  # pylint: disable=too-many-statements
    if args.instance_path and not path.isdir(args.instance_path):
        raise argparse.ArgumentTypeError(f"Directory {args.instance_path} is not a valid path!")

    if args.workers < 1:

• Read up
  Read up

File node.py has 343 lines of code (exceeds 250 allowed). Consider refactoring.

Open

from typing import Optional
from pathlib import Path

from opera_tosca_parser.parser.tosca.v_1_3.template.node import Node as Template
from opera_tosca_parser.parser.tosca.v_1_3.template.trigger import Trigger

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

Open

    def join(self, params, node=None):
        if 1 <= len(params) <= 2:
            if not isinstance(params[0], list):
                raise DataError(f"Concat or join intrinsic function parameters '{params}' should be a list")

• Read up
  Read up

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

Open

    def get_attribute(self, params):
        host, attr, *rest = params

        if host == OperationHost.SELF.value:
            # TODO: Add support for nested attribute values once we have data type support.

• Read up
  Read up

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

Open

    def get_property(self, params):
        host, prop, *rest = params

        if host == OperationHost.SOURCE.value:
            return self.source.get_property([OperationHost.SELF.value, prop] + rest)

• Read up
  Read up

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

Open

    def get_property(self, params):
        host, prop, *rest = params

        if host == OperationHost.SELF.value:
            # TODO: Add support for nested property values.

• Read up
  Read up

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

Open

def _parser_callback(args):
    if args.instance_path and not path.isdir(args.instance_path):
        raise argparse.ArgumentTypeError(f"Directory {args.instance_path} is not a valid path!")

    storage = Storage.create(args.instance_path)

• Read up
  Read up

File test_node_policies.py has 298 lines of code (exceeds 250 allowed). Consider refactoring.

Open

import pathlib

import pytest
from opera_tosca_parser.commands.parse import parse_service_template

Function notify has a Cognitive Complexity of 21 (exceeds 5 allowed). Consider refactoring.

Open

def notify(storage: Storage, verbose_mode: bool, trigger_name_or_event: str,
           notification_file_contents: typing.Optional[str]):
    if storage.exists("inputs"):
        inputs = yaml.safe_load(storage.read("inputs"))
    else:

• Read up
  Read up

Function _parser_callback has a Cognitive Complexity of 19 (exceeds 5 allowed). Consider refactoring.

Open

def _parser_callback(args):
    if args.instance_path and not path.isdir(args.instance_path):
        raise argparse.ArgumentTypeError(f"Directory {args.instance_path} is not a valid path!")

    if args.workers < 1:

• Read up
  Read up

Function compare has a Cognitive Complexity of 19 (exceeds 5 allowed). Consider refactoring.

Open

    def compare(self, collection1, collection2, context):  # pylint: disable=arguments-differ
        if collection1 is None:
            collection1 = {}
        if collection2 is None:
            collection2 = {}

• Read up
  Read up

Function _parser_callback has a Cognitive Complexity of 17 (exceeds 5 allowed). Consider refactoring.

Open

def _parser_callback(args):
    if args.instance_path and not path.isdir(args.instance_path):
        raise argparse.ArgumentTypeError(f"Directory {args.instance_path} is not a valid path!")

    storage_old = Storage.create(args.instance_path)

• Read up
  Read up

Topology has 21 functions (exceeds 20 allowed). Consider refactoring.

Open

class Topology:  # pylint: disable=too-many-public-methods
    def __init__(self, template, storage=None):
        self.storage = storage
        self.nodes = {n.tosca_id: n for n in (Node.instantiate(node, self) for node in template.nodes.values())}
        self.relationships = {r.tosca_id: r for r in (Relationship.instantiate(relationship, self)

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

Open

    def _check_dependencies(self, nodes, changed_nodes, parent_name, parent_changed):
        dependency_changes: Dict[str, set] = {}
        # dependency graph is acyclic so we may not care about
        # marking processed nodes
        for node in nodes:

• Read up
  Read up

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

Open

    def invoke_trigger(self, verbose: bool, workdir: str, trigger: Trigger,
                       notification_file_contents: Optional[str]):
        for interface_name, operation_name, _ in trigger.action:
            # get node's interface and interface operations that are referenced in a trigger action
            interface = self.template.interfaces.get(interface_name, None)

• Read up
  Read up

Function token has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def token(params):
        if isinstance(params, list) and len(params) == 3:
            string_with_tokens = params[0]
            string_of_token_chars = params[1]
            substring_index = params[2]

• Read up
  Read up

Function notify has a Cognitive Complexity of 12 (exceeds 5 allowed). Consider refactoring.

Open

    def notify(self, verbose: bool, workdir: str, trigger_name_or_event: Optional[str],
               notification_file_contents: Optional[str]):
        thread_utils.print_thread(f"  Notifying {self.tosca_id}")

        for policy in self.template.policies:

• Read up
  Read up

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

Open

def run(host, primary, dependencies, artifacts, variables, verbose, workdir, validate):
    # pylint: disable=too-many-locals
    with tempfile.TemporaryDirectory() as dir_path:
        playbook = os.path.join(dir_path, os.path.basename(primary))
        utils.copy(os.path.join(workdir, primary), playbook)

• Read up
  Read up

Function update has 9 arguments (exceeds 4 allowed). Consider refactoring.

Open

def update(