feat: random password api support data_source_id arg (TencentBlueKing…

…#1493)

src/bk-user/bkuser/apis/web/data_source/serializers.py

@@ -297,15 +297,29 @@ class DataSourceTestConnectionOutputSLZ(serializers.Serializer):
 class DataSourceRandomPasswordInputSLZ(serializers.Serializer):
     """生成随机密码"""
 
+    data_source_id = serializers.IntegerField(help_text="数据源 ID", required=False)
     password_rule_config = serializers.JSONField(help_text="密码规则配置", required=False)
 
     def validate(self, attrs):
-        passwd_rule_cfg = attrs.get("password_rule_config")
-        if passwd_rule_cfg:
+        if passwd_rule_cfg := attrs.get("password_rule_config"):
             try:
                 attrs["password_rule"] = PasswordRuleConfig(**passwd_rule_cfg).to_rule()
             except PDValidationError as e:
                 raise ValidationError(_("密码规则配置不合法: {}").format(stringify_pydantic_error(e)))
+        elif data_source_id := attrs.get("data_source_id"):
+            data_source = DataSource.objects.filter(
+                id=data_source_id,
+                plugin_id=DataSourcePluginEnum.LOCAL,
+                owner_tenant_id=self.context["tenant_id"],
+            ).first()
+            if not data_source:
+                raise ValidationError(_("指定数据源不存在或不可用"))
+
+            plugin_config = data_source.get_plugin_cfg()
+            if not plugin_config.enable_account_password_login:
+                raise ValidationError(_("无法使用该数据源生成随机密码"))
+
+            attrs["password_rule"] = plugin_config.password_rule.to_rule()
         else:
             attrs["password_rule"] = get_default_plugin_cfg(  # type: ignore
                 DataSourcePluginEnum.LOCAL,

src/bk-user/bkuser/apis/web/data_source/views.py

@@ -219,15 +219,18 @@ def put(self, request, *args, **kwargs):
         return Response(status=status.HTTP_204_NO_CONTENT)
 
 
-class DataSourceRandomPasswordApi(generics.CreateAPIView):
+class DataSourceRandomPasswordApi(CurrentUserTenantMixin, generics.CreateAPIView):
     @swagger_auto_schema(
         tags=["data_source"],
         operation_description="生成数据源用户随机密码",
         request_body=DataSourceRandomPasswordInputSLZ(),
         responses={status.HTTP_200_OK: DataSourceRandomPasswordOutputSLZ()},
     )
     def post(self, request, *args, **kwargs):
-        slz = DataSourceRandomPasswordInputSLZ(data=request.data)
+        slz = DataSourceRandomPasswordInputSLZ(
+            data=request.data,
+            context={"tenant_id": self.get_current_tenant_id()},
+        )
         slz.is_valid(raise_exception=True)
         data = slz.validated_data
 

src/bk-user/bkuser/apis/web/data_source_organization/views.py

@@ -13,6 +13,7 @@
 
 from django.db import transaction
 from django.db.models import Q
+from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import generics, status
@@ -293,6 +294,7 @@ def put(self, request, *args, **kwargs):
 
         with transaction.atomic():
             identify_info.password = make_password(raw_password)
+            identify_info.password_updated_at = timezone.now()
             identify_info.save(update_fields=["password", "password_updated_at", "updated_at"])
 
             DataSourceUserDeprecatedPasswordRecord.objects.create(

src/bk-user/tests/apis/web/data_source/test_data_source.py

@@ -109,6 +109,29 @@ def test_retrieve_not_exists(self, api_client):
         assert resp.status_code == status.HTTP_404_NOT_FOUND
 
 
+class TestDataSourceRandomPasswordApi:
+    def test_generate_with_config(self, api_client, local_ds_plugin_cfg):
+        resp = api_client.post(
+            reverse("data_source.random_passwords"),
+            data={"password_rule_config": local_ds_plugin_cfg["password_rule"]},
+        )
+        assert resp.status_code == status.HTTP_200_OK
+        assert resp.data["password"] != ""
+
+    def test_generate_with_data_source_id(self, api_client, bare_local_data_source):
+        resp = api_client.post(
+            reverse("data_source.random_passwords"),
+            data={"data_source_id": bare_local_data_source.id},
+        )
+        assert resp.status_code == status.HTTP_200_OK
+        assert resp.data["password"] != ""
+
+    def test_generate_with_default_config(self, api_client):
+        resp = api_client.post(reverse("data_source.random_passwords"))
+        assert resp.status_code == status.HTTP_200_OK
+        assert resp.data["password"] != ""
+
+
 class TestDataSourceCreateApi:
     def test_create_local_data_source(self, api_client, local_ds_plugin_cfg):
         resp = api_client.post(