(check)
Validate all recommended monitors definition files.

    If `check` is specified, only the check will be validated, if check value is 'changed' will only apply to changed
    checks, an 'all' or empty `check` value will validate all README files.
Validate all recommended monitors definition files.
Validate all recommended monitors definition files .
def recommended_monitors(check):
    """Validate all recommended monitors definition files.

    If `check` is specified, only the check will be validated, if check value is 'changed' will only apply to changed
    checks, an 'all' or empty `check` value will validate all README files.
    """

    checks = process_checks_option(check, source='integrations', extend_changed=True)
    echo_info(f"Validating recommended monitors for {len(checks)} checks ...")

    failed_checks = 0
    ok_checks = 0

    for check_name in checks:
        display_queue = []
        file_failed = False
        manifest = Manifest.load_manifest(check_name)
        monitors_relative_locations, invalid_files = get_assets_from_manifest(check_name, 'monitors')
        manifest_file = get_manifest_file(check_name)
        for file in invalid_files:
            echo_info(f'{check_name}... ', nl=False)
            echo_info(' FAILED')
            message = f'{file} does not exist'
            echo_failure('  ' + message)
            failed_checks += 1
            annotate_error(manifest_file, message)

        for monitor_file in monitors_relative_locations:
            monitor_filename = os.path.basename(monitor_file)
            try:
                decoded = json.loads(read_file(monitor_file).strip())
            except json.JSONDecodeError as e:
                failed_checks += 1
                echo_info(f'{check_name}... ', nl=False)
                echo_failure(' FAILED')
                message = f'invalid json: {e}'
                echo_failure('  ' + message)
                annotate_error(monitor_file, message)
                continue

            all_keys = set(decoded.keys())
            if not REQUIRED_ATTRIBUTES.issubset(all_keys):
                missing_fields = REQUIRED_ATTRIBUTES.difference(all_keys)
                file_failed = True
                display_queue.append(
                    (echo_failure, f"    {monitor_filename} does not contain the required fields: {missing_fields}"),
                )
            elif any([item for item in all_keys if item in EXTRA_NOT_ALLOWED_FIELDS]):
                file_failed = True
                display_queue.append(
                    (
                        echo_failure,
                        f"    {monitor_filename} contains unsupported field(s). Please ensure none of the following are"
                        f" in the file: {EXTRA_NOT_ALLOWED_FIELDS}",
                    ),
                )
            else:
                # If all required keys exist, validate value
                monitor_type = decoded.get('type')
                if monitor_type not in ALLOWED_MONITOR_TYPES:
                    file_failed = True
                    display_queue.append(
                        (
                            echo_failure,
                            f"    {monitor_filename} is of unsupported type: \"{monitor_type}\". Only"
                            f" the following types are allowed: {ALLOWED_MONITOR_TYPES}",
                        )
                    )

                description = decoded.get('recommended_monitor_metadata').get('description')
                if description is not None:
                    if len(description) > 300:
                        file_failed = True
                        display_queue.append(
                            (
                                echo_failure,
                                f"    {monitor_filename} has a description field that is too long, must be < 300 chars",
                            ),
                        )

                result = [i for i in decoded.get('tags') if i.startswith('integration:')]
                if len(result) < 1:
                    file_failed = True
                    display_queue.append((echo_failure, f"    {monitor_filename} must have an `integration` tag"))

                display_name = manifest.get_display_name().lower()
                monitor_name = decoded.get('name').lower()
                if not (check_name in monitor_name or display_name in monitor_name):
                    file_failed = True
                    display_queue.append(
                        (echo_failure, f"    {monitor_filename} name must contain the integration name"),
                    )

            if file_failed:
                failed_checks += 1
                # Display detailed info if file is invalid
                echo_info(f'{check_name}... ', nl=False)
                echo_failure(' FAILED')
                annotate_display_queue(monitor_file, display_queue)
                for display_func, message in display_queue:
                    display_func(message)
                display_queue = []
            else:
                ok_checks += 1

    if ok_checks:
        echo_success(f"{ok_checks} valid files")
    if failed_checks:
        echo_failure(f"{failed_checks} invalid files")
        abort()
def recommended_monitors ( check ) : checks = process_checks_option ( check , source = 'integrations' , extend_changed = True ) echo_info ( f"Validating recommended monitors for {len(checks)} checks ..." ) failed_checks = 0 ok_checks = 0 for check_name in checks : display_queue = [ ] file_failed = False manifest = Manifest . load_manifest ( check_name ) monitors_relative_locations , invalid_files = get_assets_from_manifest ( check_name , 'monitors' ) manifest_file = get_manifest_file ( check_name ) for file in invalid_files : echo_info ( f'{check_name}... ' , nl = False ) echo_info ( ' FAILED' ) message = f'{file} does not exist' echo_failure ( '  ' + message ) failed_checks += 1 annotate_error ( manifest_file , message ) for monitor_file in monitors_relative_locations : monitor_filename = os . path . basename ( monitor_file ) try : decoded = json . loads ( read_file ( monitor_file ) . strip ( ) ) except json . JSONDecodeError as e : failed_checks += 1 echo_info ( f'{check_name}... ' , nl = False ) echo_failure ( ' FAILED' ) message = f'invalid json: {e}' echo_failure ( '  ' + message ) annotate_error ( monitor_file , message ) continue all_keys = set ( decoded . keys ( ) ) if not REQUIRED_ATTRIBUTES . issubset ( all_keys ) : missing_fields = REQUIRED_ATTRIBUTES . difference ( all_keys ) file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} does not contain the required fields: {missing_fields}" ) , ) elif any ( [ item for item in all_keys if item in EXTRA_NOT_ALLOWED_FIELDS ] ) : file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} contains unsupported field(s). Please ensure none of the following are" f" in the file: {EXTRA_NOT_ALLOWED_FIELDS}" , ) , ) else : # If all required keys exist, validate value monitor_type = decoded . get ( 'type' ) if monitor_type not in ALLOWED_MONITOR_TYPES : file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} is of unsupported type: \"{monitor_type}\". Only" f" the following types are allowed: {ALLOWED_MONITOR_TYPES}" , ) ) description = decoded . get ( 'recommended_monitor_metadata' ) . get ( 'description' ) if description is not None : if len ( description ) > 300 : file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} has a description field that is too long, must be < 300 chars" , ) , ) result = [ i for i in decoded . get ( 'tags' ) if i . startswith ( 'integration:' ) ] if len ( result ) < 1 : file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} must have an `integration` tag" ) ) display_name = manifest . get_display_name ( ) . lower ( ) monitor_name = decoded . get ( 'name' ) . lower ( ) if not ( check_name in monitor_name or display_name in monitor_name ) : file_failed = True display_queue . append ( ( echo_failure , f"    {monitor_filename} name must contain the integration name" ) , ) if file_failed : failed_checks += 1 # Display detailed info if file is invalid echo_info ( f'{check_name}... ' , nl = False ) echo_failure ( ' FAILED' ) annotate_display_queue ( monitor_file , display_queue ) for display_func , message in display_queue : display_func ( message ) display_queue = [ ] else : ok_checks += 1 if ok_checks : echo_success ( f"{ok_checks} valid files" ) if failed_checks : echo_failure ( f"{failed_checks} invalid files" ) abort ( )