def _convert_add_related_integrations()

in detection_rules/rule.py [0:0]

• 29 lines of code
• 15 McCabe index (conditional complexity)

    def _convert_add_related_integrations(self, obj: dict) -> None:
        """Add restricted field related_integrations to the obj."""
        field_name = "related_integrations"
        package_integrations = obj.get(field_name, [])

        if not package_integrations and self.metadata.integration:
            packages_manifest = load_integrations_manifests()
            current_stack_version = load_current_package_version()

            if self.check_restricted_field_version(field_name):
                if (isinstance(self.data, QueryRuleData) or isinstance(self.data, MachineLearningRuleData)):
                    if (self.data.get('language') is not None and self.data.get('language') != 'lucene') or \
                            self.data.get('type') == 'machine_learning':
                        package_integrations = self.get_packaged_integrations(self.data, self.metadata,
                                                                              packages_manifest)

                        if not package_integrations:
                            return

                        for package in package_integrations:
                            package["version"] = find_least_compatible_version(
                                package=package["package"],
                                integration=package["integration"],
                                current_stack_version=current_stack_version,
                                packages_manifest=packages_manifest)

                            # if integration is not a policy template remove
                            if package["version"]:
                                version_data = packages_manifest.get(package["package"],
                                                                     {}).get(package["version"].strip("^"), {})
                                policy_templates = version_data.get("policy_templates", [])

                                if package["integration"] not in policy_templates:
                                    del package["integration"]

                    # remove duplicate entries
                    package_integrations = list({json.dumps(d, sort_keys=True):
                                                d for d in package_integrations}.values())
                    obj.setdefault("related_integrations", package_integrations)