[Stash] Test case 通不过，import error

apps/ops/models.py

@@ -18,7 +18,8 @@ class AdHoc(models.Model):
     id = models.UUIDField(default=uuid.uuid4, primary_key=True)
     name = models.CharField(max_length=128, blank=True, verbose_name=_('Name'))
     is_deleted = models.BooleanField(default=False)
-    date_create = models.DateTimeField(auto_created=True)
+    created_by = models.CharField(max_length=128, blank=True, default='')
+    date_create = models.DateTimeField(auto_now_add=True)
 
     @property
     def short_id(self):
@@ -45,7 +46,7 @@ class AdHocData(models.Model):
     _become_pass = models.CharField(default='', max_length=128)
     pattern = models.CharField(max_length=64, default='', verbose_name=_('Pattern'))
     created_by = models.CharField(max_length=64, verbose_name=_('Create by'))
-    date_created = models.DateTimeField(auto_created=True)
+    date_created = models.DateTimeField(auto_now_add=True)
 
     @property
     def tasks(self):

apps/ops/test_utils.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+#
+
+import sys
+import os
+from django.test import TestCase
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "jumpserver.settings")
+from ops.models import AdHoc, AdHocData
+from ops.utils import run_adhoc
+
+
+class TestRunAdHoc(TestCase):
+    def setUp(self):
+        adhoc = AdHoc(name="Test run adhoc")
+        adhoc.save()
+
+        self.data = AdHocData(subject=adhoc, run_as_admin=True, pattern='all')
+        self.data.tasks = [
+            {'name': 'run ls', 'action': {'module': 'shell', 'args': 'ls'}},
+            {'name': 'echo ', 'action': {'module': 'shell', 'args': 'echo 123'}},
+        ]
+        self.data.hosts = [
+            "testserver"
+        ]
+
+    def test_run(self):
+        pass
+
+
+
+
+
+

apps/ops/utils.py

@@ -1,16 +1,13 @@
 # ~*~ coding: utf-8 ~*~
-
-from __future__ import absolute_import, unicode_literals
-
-import json
 import re
-import time
-import uuid
 
+import time
 from django.utils import timezone
 
 from common.utils import get_logger, get_object_or_none
 from .ansible import AdHocRunner
+from .ansible.exceptions import AnsibleError
+from .models import AdHocRunHistory
 from assets.utils import get_assets_by_hostname_list
 
 logger = get_logger(__file__)
@@ -82,16 +79,43 @@ def hosts_add_become(hosts, adhoc_data):
     return hosts
 
 
-def run_adhoc(adhoc_data, forks=10):
-    tasks = adhoc_data.tasks
+def run_adhoc(adhoc_data, **options):
+    """
+    :param adhoc_data: Instance of AdHocData
+    :param options: ansible support option, like forks ...
+    :return:
+    """
+    name = adhoc_data.subject.name
     hostname_list = adhoc_data.hosts
-    adhoc_name = adhoc_data.subject.name
-
     if adhoc_data.run_as_admin:
-        hosts = get_hosts_with_admin(adhoc_data.hosts)
+        hosts = get_hosts_with_admin(hostname_list)
     else:
         hosts = get_hosts_with_run_user(hostname_list, adhoc_data.run_as)
         hosts_add_become(hosts, adhoc_data)  # admin user 自带become
 
     runner = AdHocRunner(hosts)
-    runner.set_option('forks', forks)
+    for k, v in options:
+        runner.set_option(k, v)
+
+    record = AdHocRunHistory(adhoc=adhoc_data)
+    time_start = time.time()
+    try:
+        result = runner.run(adhoc_data.tasks, adhoc_data.pattern, name)
+        record.is_finished = True
+        if result.results_summary.get('dark'):
+            record.is_success = False
+        else:
+            record.is_success = True
+        record.result = result.results_raw
+        record.summary = result.results_summary
+        return result
+    except AnsibleError as e:
+        logger.error("Failed run adhoc {}, {}".format(name, e))
+        raise
+    finally:
+        record.date_finished = timezone.now()
+        record.timedelta = time.time() - time_start
+        record.save()
+
+
+