feature: 支持bk_agent_id (close TencentBlueKing#562)

apps/mock_data/views_mkd/job.py

@@ -16,7 +16,7 @@
 from apps.node_man import constants
 
 JOB_INSTALL_REQUEST_PARAMS = {
-    "job_type": "INSTALL_AGENT",
+    "job_type": constants.JobType.INSTALL_AGENT,
     "hosts": [
         {
             "bk_cloud_id": constants.DEFAULT_CLOUD,

apps/node_man/tests/test_views/test_job_views.py

@@ -9,21 +9,45 @@
 specific language governing permissions and limitations under the License.
 """
 import copy
+from unittest.mock import patch
 
 from apps.exceptions import ValidationError
 from apps.mock_data.views_mkd import job
+from apps.node_man import constants
+from apps.node_man.tests.utils import Subscription
 from apps.utils.unittest.testcase import CustomAPITestCase
 
 
 class JobViewsTestCase(CustomAPITestCase):
+    @patch("apps.node_man.handlers.job.JobHandler.create_subscription", Subscription.create_subscription)
     def test_install(self):
         data = copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS)
         result = self.client.post(path="/api/job/install/", data=data)
         self.assertEqual(result["result"], True)
 
+
+class TestJobValidationError(JobViewsTestCase):
+    @staticmethod
+    def modify_request_params(data):
+        return data
+
+    def test_install_raise_validation_error(self):
+        pass
+
     def test_install_with_empty_inner_ip_at_the_same_time(self):
-        data = copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS)
+        data = self.modify_request_params(copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS))
+
         data["hosts"][0]["inner_ip"] = ""
         data["hosts"][0]["inner_ipv6"] = ""
         result = self.client.post(path="/api/job/install/", data=data)
         self.assertEqual(result["code"], ValidationError().code)
+        self.assertTrue("inner_ip 和 inner_ipv6 不能同时为空" in result["message"])
+
+    def test_install_with_empty_outer_ip_at_the_same_time(self):
+        data = copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS)
+        data["hosts"][0]["outer_ip"] = ""
+        data["hosts"][0]["outer_ipv6"] = ""
+        data["job_type"] = constants.JobType.INSTALL_PROXY
+        result = self.client.post(path="/api/job/install/", data=data)
+        self.assertEqual(result["code"], ValidationError().code)
+        self.assertTrue("outer_ip 和 outer_ipv6 不能同时为空" in result["message"])

apps/utils/unittest/testcase.py

@@ -13,8 +13,8 @@
 from typing import Any, Dict, List, Optional, Union
 from unittest import mock
 
-from blueapps.account.models import User
-from django.db import IntegrityError, transaction
+from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.test import Client, RequestFactory, TestCase
 
 from apps.utils.unittest import base
@@ -139,21 +139,15 @@ def setUpTestData(cls):
         super().setUpTestData()
 
         # 创建用于测试的超管
-        try:
-            with transaction.atomic():
-                cls.superuser = User.objects.create_user(
-                    username=cls.SUPERUSER_NAME,
-                    password=cls.SUPERUSER_PASSWORD,
-                    # is_superuser=True,
-                    # is_staff=True,
-                    # is_active=True,
-                )
-        except IntegrityError:
-            pass
+        user_model = get_user_model()
+        user = user_model.objects.first()
+        user.set_password(cls.SUPERUSER_PASSWORD)
+        user.save()
 
     def setUp(self) -> None:
         super().setUp()
-        self.client.login(username=self.SUPERUSER_NAME, password=self.SUPERUSER_PASSWORD)
+        settings.AUTHENTICATION_BACKENDS = ["django.contrib.auth.backends.ModelBackend"]
+        self.client.login(username=self.SUPERUSER_NAME, password=self.SUPERUSER_NAME)
 
     def tearDown(self) -> None:
         super().tearDown()