diff --git a/sql/utils/test_workflow_audit.py b/sql/utils/test_workflow_audit.py index ab0b1a5e64..816aeed424 100644 --- a/sql/utils/test_workflow_audit.py +++ b/sql/utils/test_workflow_audit.py @@ -445,21 +445,6 @@ def test_generate_audit_setting_empty_config(sql_query_apply): assert "未配置审流" in str(e.value) -def test_generate_audit_setting_auto_review( - sql_workflow, setup_sys_config, mocker: MockFixture -): - sql_workflow, _ = sql_workflow - setup_sys_config.set("auto_review", True) - - audit = AuditV2(workflow=sql_workflow, sys_config=setup_sys_config) - mock_is_auto_review = mocker.patch.object( - audit, "is_auto_review", return_value=True - ) - audit_setting = audit.generate_audit_setting() - assert audit_setting.auto_pass is True - mock_is_auto_review.assert_called() - - def test_get_workflow( archive_apply, sql_query_apply, @@ -611,3 +596,41 @@ def test_get_review_info_auto_pass( response = admin_client.get(f"/queryapplydetail/{sql_query_apply.apply_id}/") assert response.status_code == 200 assert "无需审批" in response.content.decode("utf-8") + + +def test_auto_review_with_auto_reject(sql_workflow, mocker: MockFixture): + """自动审和不通过时无法自动审批""" + mocker.patch.object(AuditV2, "is_auto_reject").return_value = True + sql_workflow, _ = sql_workflow + audit = AuditV2(workflow=sql_workflow) + assert audit.is_auto_review() is False + + +def test_auto_reject_non_sql_review(sql_query_apply): + """当前自动审核仅对 SQL 上线工单生效""" + audit = AuditV2(workflow=sql_query_apply) + assert audit.is_auto_reject() is False + + +def test_auto_reject_not_applicable(sql_workflow, setup_sys_config): + """测试自动拒绝场景""" + sql_workflow, _ = sql_workflow + audit = AuditV2(workflow=sql_workflow, sys_config=setup_sys_config) + # warning_count > 0 and auto_review_wrong == "1", + audit.sys_config.set("auto_review_wrong", "1") + audit.workflow.sqlworkflowcontent.review_content = json.dumps([{"errlevel": 1}]) + audit.workflow.sqlworkflowcontent.save() + assert audit.is_auto_reject() is True + # error_count > 0 and auto_review_wrong in ("", "1", "2") + audit.workflow.sqlworkflowcontent.review_content = json.dumps([{"errlevel": 2}]) + audit.workflow.sqlworkflowcontent.save() + audit.sys_config.set("auto_review_wrong", "") + assert audit.is_auto_reject() is True + audit.sys_config.set("auto_review_wrong", "1") + assert audit.is_auto_reject() is True + audit.sys_config.set("auto_review_wrong", "2") + assert audit.is_auto_reject() is True + # warning_count=0 error_count=0 + audit.workflow.sqlworkflowcontent.review_content = json.dumps([{"errlevel": 0}]) + audit.workflow.sqlworkflowcontent.save() + assert audit.is_auto_reject() is False diff --git a/sql/utils/workflow_audit.py b/sql/utils/workflow_audit.py index 6383465322..c2b2b283f4 100644 --- a/sql/utils/workflow_audit.py +++ b/sql/utils/workflow_audit.py @@ -99,9 +99,12 @@ class AuditSetting: audit_auth_groups: List = field(default_factory=list) auto_pass: bool = False + auto_reject: bool = False @property def audit_auth_group_in_db(self): + if self.auto_reject or self.auto_pass: + return "" return ",".join(str(x) for x in self.audit_auth_groups) @@ -192,7 +195,34 @@ def get_workflow(self): self.resource_group = self.audit.group_name self.resource_group_id = self.audit.group_id + def is_auto_reject(self): + """系统自动驳回工单""" + if self.workflow_type != WorkflowType.SQL_REVIEW: + return False + # 按照系统配置确定是自动驳回还是放行 + auto_review_wrong = self.sys_config.get( + "auto_review_wrong", "" + ) # 1表示出现警告就驳回，2和空表示出现错误才驳回 + review_content = self.workflow.sqlworkflowcontent.review_content or "[]" + warning_count, error_count = 0, 0 + for r in json.loads(review_content): + err_level = ReviewResult(**r).errlevel + if err_level == 1: + warning_count += 1 + if err_level == 2: + error_count += 1 + if any( + [ + warning_count > 0 and auto_review_wrong == "1", + error_count > 0 and auto_review_wrong in ("", "1", "2"), + ] + ): + return True + return False + def is_auto_review(self) -> bool: + if self.is_auto_reject(): + return False if self.workflow_type != WorkflowType.SQL_REVIEW: # 当前自动审核仅对 sql 上线工单有用 return False @@ -237,11 +267,8 @@ def is_auto_review(self) -> bool: return True def generate_audit_setting(self) -> AuditSetting: - if self.is_auto_review(): - return AuditSetting(auto_pass=True) if self.workflow_type in [WorkflowType.SQL_REVIEW, WorkflowType.QUERY]: group_id = self.workflow.group_id - else: # ArchiveConfig group_id = self.resource_group_id @@ -252,7 +279,9 @@ def generate_audit_setting(self) -> AuditSetting: except WorkflowAuditSetting.DoesNotExist: raise AuditException(f"审批类型 {self.workflow_type.label} 未配置审流") return AuditSetting( - audit_auth_groups=workflow_audit_setting.audit_auth_groups.split(",") + auto_pass=self.is_auto_review(), + auto_reject=self.is_auto_reject(), + audit_auth_groups=workflow_audit_setting.audit_auth_groups.split(","), ) def create_audit(self) -> str: @@ -300,8 +329,21 @@ def create_audit(self) -> str: create_user=create_user, create_user_display=create_user_display, ) - # 自动通过的情况 - if audit_setting.auto_pass: + # 自动审批的情况 + if audit_setting.auto_reject: + self.audit.current_status = WorkflowStatus.REJECTED + self.audit.save() + WorkflowLog.objects.create( + audit_id=self.audit.audit_id, + operation_type=WorkflowAction.SUBMIT, + operation_type_desc=WorkflowAction.SUBMIT.label, + operation_info="系统直接审核不通过", + operator=self.audit.create_user, + operator_display=self.audit.create_user_display, + ) + + return "直接审核不通过" + elif audit_setting.auto_pass: self.audit.current_status = WorkflowStatus.PASSED self.audit.save() WorkflowLog.objects.create( diff --git a/sql_api/serializers.py b/sql_api/serializers.py index 051d38a74f..14986cd25c 100644 --- a/sql_api/serializers.py +++ b/sql_api/serializers.py @@ -397,18 +397,8 @@ def create(self, validated_data): if not sys_config.get("enable_backup_switch") and check_engine.auto_backup: is_backup = True - # 按照系统配置确定是自动驳回还是放行 - auto_review_wrong = sys_config.get( - "auto_review_wrong", "" - ) # 1表示出现警告就驳回，2和空表示出现错误才驳回 - workflow_status = "workflow_manreviewing" - if check_result.warning_count > 0 and auto_review_wrong == "1": - workflow_status = "workflow_autoreviewwrong" - elif check_result.error_count > 0 and auto_review_wrong in ("", "1", "2"): - workflow_status = "workflow_autoreviewwrong" - workflow_data.update( - status=workflow_status, + status="workflow_manreviewing", is_backup=is_backup, is_manual=0, syntax_type=check_result.syntax_type, @@ -432,7 +422,9 @@ def create(self, validated_data): logger.error(f"提交工单报错，错误信息：{traceback.format_exc()}") raise serializers.ValidationError({"errors": str(e)}) # 有时候提交后自动审批通过, 在这里改写一下 workflow 状态 - if auditor.audit.current_status == WorkflowStatus.PASSED: + if auditor.audit.current_status == WorkflowStatus.REJECTED: + auditor.workflow.status = "workflow_autoreviewwrong" + elif auditor.audit.current_status == WorkflowStatus.PASSED: auditor.workflow.status = "workflow_review_pass" auditor.workflow.save() return workflow_content