fix: After setting the function library to private, the workflow used should report an error stating that the current function is not authorized to be used

[shaohuzhang1]

fix: After setting the function library to private, the workflow used should report an error stating that the current function is not authorized to be used

[f2c-ci-robot]

Adding the "do-not-merge/release-note-label-needed" label because no release-note block was detected, please follow our release note process to remove it.

Details

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes/test-infra repository.

[f2c-ci-robot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[shaohuzhang1]

The provided code has several issues and areas for improvement:

Issues:

1. Duplicate get_field_value Method: There seems to be a duplicate get_field_value method that could cause confusion. It's better to remove one of them.

2. String Translation Missing Parameter:

   • In valid_function, there is an exception message where a parameter _type is missing in the translation string: 'Field: {name} Type: {_type} Unsupported Types'.

3. Improper Exception Handling: Some exceptions are raised without providing context messages which can make debug harder.

4. Inconsistent Code Structure:

   • The code structure can be cleaner by grouping related methods together and ensuring consistent formatting.

Optimization Suggestions:

1. Use More Consistent Variable Names: Use descriptive variable names to enhance readability, such as field_name, reference_value, input_fields, etc.

2. Remove Unnecessary Imports:

   • If _ (gettext) isn't needed throughout the function, consider removing it to reduce clutter.

3. Simplify JSON Parsing Error Handling: Extract common handling logic for parsing errors into a separate function or method.

Here's an improved version of the code with these considerations:

from typing import Dict

from django.db.models import QuerySet
import json

from application.flow.i_step_node import NodeResult
from application.flow.step_node.function_lib_node.i_function_lib_node import IFunctionLibNode


def get_field_value(field_list, name, is_required):
    """Get the value from a list of fields."""
    results = [entry['value'] for entry in field_list if entry['name'] == name]
    return results[-1] if len(results) > 0 else (
        f"{name}字段未设置值" if is_required else None
    )


def convert_value(value, _type, is_required=None):
    try:
        if _type == 'object':
            v = json.loads(value)
            if isinstance(v, dict):
                return v
            raise ValueError("类型错误")
        if _type == 'array':
            v = json.loads(value)
            if isinstance(v, list):
                return v
            raise ValueError("类型错误")
        # Add other types if necessary
        return value
    except (json.JSONDecodeError, TypeError):
        if is_required:
            raise ValueError(f'字段:{value} 类型错误')
        return None


class BaseFunctionLibNode(IFunctionLibNode):
    def __init__(self, flow_params_serializer):
        self.flow_params_serializer = flow_params_serializer

    def save_context(self, details, workflow_manage):
        pass

    def execute(self, function_lib_id, input_field_list, **kwargs) -> NodeResult:
        function_lib = QuerySet(FunctionLib).filter(id=function_lib_id).first()
        valid_function(function_lib, self.flow_params_serializer.data.get('user_id'))

        params = {}
        for field in input_field_list:
            field_name = field['name']
            reference_value = get_field_value(input_field_list, field_name, field['is_required'])

            field_result = convert_value(
                reference_value,
                field['type'],
                field['is_required'],
                field['source'],
                self
            )
            
            # Assuming you want to store the resulting values in the dictionary
            params[field_name] = field_result

        return NodeResult(params)


def valid_function(function_lib, user_id):
    if function_lib is None:
        raise ValueError(_("Function does not exist"))
    if function_lib.permission_type == "PRIVATE" and str(function_lib.user_id) != str(user_id):
        raise ValueError(_("No permission to use this function %s").format(function_lib.name))
    if not function_lib.is_active:
        raise ValueError(_("Function %s is unavailable").format(function_lib.name))

This version cleans up the code, improves naming consistency, and provides more meaningful error messages.