When using Django models from a script, make the current user default…

… to an actual database user named "autotest_system". This allows for simpler, more consistent code.

Signed-off-by: Steve Howard <[email protected]>

frontend/afe/doctests/001_rpc_test.txt

@@ -909,8 +909,8 @@ ValidationError: {'hosts': u'Host(s) "ahost1, ahost2" are atomic group hosts but
 >>> peon_user.access_level = 0
 >>> from autotest_lib.client.common_lib.test_utils import mock
 >>> god = mock.mock_god()
->>> god.stub_function(rpc_interface.thread_local, "get_user")
->>> rpc_interface.thread_local.get_user.expect_call().and_return(peon_user)
+>>> god.stub_function(models.User, "current_user")
+>>> models.User.current_user.expect_call().and_return(peon_user)
 >>> rpc_interface.abort_host_queue_entries(job__id=job_id)
 Traceback (most recent call last):
 AclAccessViolation: You cannot abort the following job entries: 8-debug_user/two-label

frontend/afe/models.py

@@ -129,6 +129,8 @@ class User(dbmodels.Model, model_logic.ModelExtensions):
     ACCESS_ADMIN = 1
     ACCESS_USER = 0
 
+    AUTOTEST_SYSTEM = 'autotest_system'
+
     login = dbmodels.CharField(max_length=255, unique=True)
     access_level = dbmodels.IntegerField(default=ACCESS_USER, blank=True)
 
@@ -161,6 +163,16 @@ def is_superuser(self):
         return self.access_level >= self.ACCESS_ROOT
 
 
+    @classmethod
+    def current_user(cls):
+        user = thread_local.get_user()
+        if user is None:
+            user = cls.objects.get_or_create(login=cls.AUTOTEST_SYSTEM)
+            user.access_level = cls.ACCESS_ROOT
+            user.save()
+        return user
+
+
     class Meta:
         db_table = 'afe_users'
 
@@ -259,7 +271,7 @@ def save(self, *args, **kwargs):
         if not first_time:
             AclGroup.check_for_acl_violation_hosts([self])
         if self.locked and not self.locked_by:
-            self.locked_by = thread_local.get_user()
+            self.locked_by = User.current_user()
             self.lock_time = datetime.now()
             self.dirty = True
         elif not self.locked and self.locked_by:
@@ -276,7 +288,7 @@ def delete(self):
         AclGroup.check_for_acl_violation_hosts([self])
         for queue_entry in self.hostqueueentry_set.all():
             queue_entry.deleted = True
-            queue_entry.abort(thread_local.get_user())
+            queue_entry.abort()
         super(Host, self).delete()
 
 
@@ -459,7 +471,7 @@ class AclGroup(dbmodels.Model, model_logic.ModelExtensions):
 
     @staticmethod
     def check_for_acl_violation_hosts(hosts):
-        user = thread_local.get_user()
+        user = User.current_user()
         if user.is_superuser():
             return
         accessible_host_ids = set(
@@ -483,7 +495,7 @@ def check_abort_permissions(queue_entries):
            * the machine isn't ACL-accessible, or
            * the machine is in the "Everyone" ACL
         """
-        user = thread_local.get_user()
+        user = User.current_user()
         if user.is_superuser():
             return
         not_owned = queue_entries.exclude(job__owner=user.login)
@@ -509,7 +521,7 @@ def check_abort_permissions(queue_entries):
 
 
     def check_for_acl_violation_acl_group(self):
-        user = thread_local.get_user()
+        user = User.current_user()
         if user.is_superuser():
             return
         if self.name == 'Everyone':
@@ -551,12 +563,12 @@ def delete(self):
 
     def add_current_user_if_empty(self):
         if not self.users.count():
-            self.users.add(thread_local.get_user())
+            self.users.add(User.current_user())
 
 
     def perform_after_save(self, change):
         if not change:
-            self.users.add(thread_local.get_user())
+            self.users.add(User.current_user())
         self.add_current_user_if_empty()
         self.on_host_membership_change()
 
@@ -734,9 +746,9 @@ def user(self):
             return None
 
 
-    def abort(self, aborted_by):
+    def abort(self):
         for queue_entry in self.hostqueueentry_set.all():
-            queue_entry.abort(aborted_by)
+            queue_entry.abort()
 
 
     def tag(self):
@@ -845,18 +857,15 @@ def is_meta_host_entry(self):
 
 
     def log_abort(self, user):
-        if user is None:
-            # automatic system abort (i.e. job timeout)
-            return
         abort_log = AbortedHostQueueEntry(queue_entry=self, aborted_by=user)
         abort_log.save()
 
 
-    def abort(self, user):
+    def abort(self):
         # this isn't completely immune to race conditions since it's not atomic,
         # but it should be safe given the scheduler's behavior.
         if not self.complete and not self.aborted:
-            self.log_abort(user)
+            self.log_abort(User.current_user())
             self.aborted = True
             self.save()
 

frontend/afe/rpc_handler.py

@@ -7,8 +7,7 @@
 
 import traceback, pydoc, re, urllib, logging, logging.handlers
 from autotest_lib.frontend.afe.json_rpc import serviceHandler
-from autotest_lib.frontend.afe import rpc_utils
-from autotest_lib.frontend import thread_local
+from autotest_lib.frontend.afe import models, rpc_utils
 from autotest_lib.client.common_lib import global_config
 from autotest_lib.frontend.afe import rpcserver_logging
 
@@ -84,7 +83,7 @@ def encode_result(self, results):
 
 
     def handle_rpc_request(self, request):
-        user = thread_local.get_user()
+        user = models.User.current_user()
         json_request = self.raw_request_data(request)
         decoded_request = self.decode_request(json_request)
         decoded_result = self.dispatch_request(decoded_request)

frontend/afe/rpc_interface.py

@@ -31,7 +31,6 @@
 
 import datetime
 import common
-from autotest_lib.frontend import thread_local
 from autotest_lib.frontend.afe import models, model_logic
 from autotest_lib.frontend.afe import control_file, rpc_utils
 from autotest_lib.client.common_lib import global_config
@@ -289,7 +288,7 @@ def get_users(**filter_data):
 
 def add_acl_group(name, description=None):
     group = models.AclGroup.add_object(name=name, description=description)
-    group.users.add(thread_local.get_user())
+    group.users.add(models.User.current_user())
     return group.id
 
 
@@ -435,7 +434,7 @@ def create_job(name, priority, control_file, control_type,
 
     @returns The created Job id number.
     """
-    user = thread_local.get_user()
+    user = models.User.current_user()
     owner = user.login
     # input validation
     if not (hosts or meta_hosts or one_time_hosts or atomic_group_name
@@ -550,9 +549,8 @@ def abort_host_queue_entries(**filter_data):
     host_queue_entries = list(query.select_related())
     rpc_utils.check_abort_synchronous_jobs(host_queue_entries)
 
-    user = thread_local.get_user()
     for queue_entry in host_queue_entries:
-        queue_entry.abort(user)
+        queue_entry.abort()
 
 
 def reverify_hosts(**filter_data):
@@ -750,7 +748,7 @@ def delete_recurring_runs(**filter_data):
 
 
 def create_recurring_run(job_id, start_date, loop_period, loop_count):
-    owner = thread_local.get_user().login
+    owner = models.User.current_user().login
     job = models.Job.objects.get(id=job_id)
     return job.create_recurring_job(start_date=start_date,
                                     loop_period=loop_period,
@@ -813,7 +811,7 @@ def get_static_data():
     result['tests'] = get_tests(sort_by=['name'])
     result['profilers'] = get_profilers(sort_by=['name'])
     result['current_user'] = rpc_utils.prepare_for_serialization(
-        thread_local.get_user().get_object_dict())
+        models.User.current_user().get_object_dict())
     result['host_statuses'] = sorted(models.Host.Status.names)
     result['job_statuses'] = sorted(models.HostQueueEntry.Status.names)
     result['job_timeout_default'] = models.Job.DEFAULT_TIMEOUT

frontend/thread_local.py

@@ -1,14 +1,7 @@
 import threading
 
-# when using the models from a script, use this object to avoid null checks all
-# over the place
-class NullUser(object):
-    def is_superuser(self):
-        return True
-
-
 _store = threading.local()
-_store.user = NullUser()
+_store.user = None
 
 def set_user(user):
     """\

frontend/tko/rpc_interface.py

@@ -1,8 +1,7 @@
 import os, pickle, datetime, itertools, operator
 from django.db import models as dbmodels
-from autotest_lib.frontend import thread_local
 from autotest_lib.frontend.afe import rpc_utils, model_logic
-from autotest_lib.frontend.afe import readonly_connection
+from autotest_lib.frontend.afe import models as afe_models, readonly_connection
 from autotest_lib.frontend.tko import models, tko_rpc_utils, graphing_utils
 from autotest_lib.frontend.tko import preconfigs
 
@@ -334,7 +333,7 @@ def get_saved_queries(**filter_data):
 
 def add_saved_query(name, url_token):
     name = name.strip()
-    owner = thread_local.get_user()
+    owner = afe_models.User.current_user().login
     existing_list = list(models.SavedQuery.objects.filter(owner=owner,
                                                           name=name))
     if existing_list:
@@ -348,7 +347,7 @@ def add_saved_query(name, url_token):
 
 
 def delete_saved_queries(id_list):
-    user = thread_local.get_user()
+    user = afe_models.User.current_user().login
     query = models.SavedQuery.objects.filter(id__in=id_list, owner=user)
     if query.count() == 0:
         raise model_logic.ValidationError('No such queries found for this user')
@@ -416,7 +415,7 @@ def get_static_data():
     result['all_fields'] = sorted(model_fields + extra_fields)
     result['test_labels'] = get_test_labels(sort_by=['name'])
     result['current_user'] = rpc_utils.prepare_for_serialization(
-            thread_local.get_user().get_object_dict())
+            afe_models.User.current_user().get_object_dict())
     result['benchmark_key'] = benchmark_key
     result['tko_perf_view'] = tko_perf_view
     result['tko_test_view'] = model_fields

scheduler/monitor_db.py

@@ -2143,6 +2143,9 @@ def _log_abort(self):
                 aborted_on.add(t)
 
         # extract some actual, unique aborted by value and write it out
+        # TODO(showard): this conditional is now obsolete, we just need to leave
+        # it in temporarily for backwards compatibility over upgrades.  delete
+        # soon.
         assert len(aborted_by) <= 1
         if len(aborted_by) == 1:
             aborted_by_value = aborted_by.pop()
@@ -3565,9 +3568,8 @@ def run_with_ready_delay(self, queue_entry):
 
     def request_abort(self):
         """Request that this Job be aborted on the next scheduler cycle."""
-        queue_entries = HostQueueEntry.fetch(where="job_id=%s" % self.id)
-        for hqe in queue_entries:
-            hqe.update_field('aborted', True)
+        self_model = models.Job.objects.get(id=self.id)
+        self_model.abort()
 
 
     def schedule_delayed_callback_task(self, queue_entry):

scheduler/monitor_db_cleanup.py

@@ -65,7 +65,7 @@ def _abort_timed_out_jobs(self):
             where=['created_on + INTERVAL timeout HOUR < NOW()'])
         for job in query.distinct():
             logging.warning('Aborting job %d due to job timeout', job.id)
-            job.abort(None)
+            job.abort()
 
 
     def _abort_jobs_past_max_runtime(self):
@@ -83,7 +83,7 @@ def _abort_jobs_past_max_runtime(self):
             id__in=[row[0] for row in rows])
         for queue_entry in query.distinct():
             logging.warning('Aborting entry %s due to max runtime', queue_entry)
-            queue_entry.abort(None)
+            queue_entry.abort()
 
 
     def _check_for_db_inconsistencies(self):

scheduler/monitor_db_unittest.py

@@ -10,7 +10,6 @@
 from autotest_lib.client.common_lib.test_utils import mock
 from autotest_lib.client.common_lib.test_utils import unittest
 from autotest_lib.database import database_connection, migrate
-from autotest_lib.frontend import thread_local
 from autotest_lib.frontend.afe import models
 from autotest_lib.scheduler import monitor_db, drone_manager, email_manager
 from autotest_lib.scheduler import scheduler_config, gc_stats