perf: gateway auto (#9529)

Co-authored-by: feng <1304903146@qq.com>

apps/assets/api/domain.py

@@ -5,7 +5,7 @@ from rest_framework.serializers import ValidationError
 from rest_framework.views import APIView, Response
 
 from common.utils import get_logger
-from assets.tasks import test_assets_connectivity_manual
+from assets.tasks import test_gateways_connectivity_manual
 from orgs.mixins.api import OrgBulkModelViewSet
 from .. import serializers
 from ..models import Domain, Gateway
@@ -55,5 +55,5 @@ class GatewayTestConnectionApi(SingleObjectMixin, APIView):
             local_port = int(local_port)
         except ValueError:
             raise ValidationError({'port': _('Number required')})
-        task = test_assets_connectivity_manual.delay([gateway.id], local_port)
+        task = test_gateways_connectivity_manual([gateway.id], local_port)
         return Response({'task': task.id})

apps/assets/tasks/__init__.py

@@ -6,3 +6,4 @@ from .common import *
 from .automation import *
 from .gather_facts import *
 from .nodes_amount import *
+from .ping_gateway import *

apps/assets/tasks/ping_gateway.py

@@ -0,0 +1,34 @@
+# ~*~ coding: utf-8 ~*~
+from celery import shared_task
+from django.utils.translation import gettext_noop
+
+from assets.const import AutomationTypes
+from common.utils import get_logger
+from orgs.utils import org_aware_func
+from .common import quickstart_automation
+
+logger = get_logger(__file__)
+
+__all__ = [
+    'test_gateways_connectivity_task',
+    'test_gateways_connectivity_manual',
+]
+
+
+@shared_task
+@org_aware_func('assets')
+def test_gateways_connectivity_task(assets, local_port, task_name=None):
+    from assets.models import PingAutomation
+    if task_name is None:
+        task_name = gettext_noop("Test gateways connectivity ")
+
+    task_name = PingAutomation.generate_unique_name(task_name)
+    task_snapshot = {'assets': [str(asset.id) for asset in assets], 'local_port': local_port}
+    quickstart_automation(task_name, AutomationTypes.ping_gateway, task_snapshot)
+
+
+def test_gateways_connectivity_manual(gateway_ids, local_port):
+    from assets.models import Asset
+    gateways = Asset.objects.filter(id__in=gateway_ids)
+    task_name = gettext_noop("Test gateways connectivity ")
+    return test_gateways_connectivity_task.delay(gateways, local_port, task_name)