Merge pull request #12884 from AlanCoding/is_testing

[tech debt] Move the IS_TESTING method out of settings

awx/main/analytics/subsystem_metrics.py

@@ -5,7 +5,9 @@ import logging
 
 from django.conf import settings
 from django.apps import apps
+
 from awx.main.consumers import emit_channel_notification
+from awx.main.utils import is_testing
 
 root_key = 'awx_metrics'
 logger = logging.getLogger('awx.main.analytics')
@@ -163,7 +165,7 @@ class Metrics:
         Instance = apps.get_model('main', 'Instance')
         if instance_name:
             self.instance_name = instance_name
-        elif settings.IS_TESTING():
+        elif is_testing():
             self.instance_name = "awx_testing"
         else:
             self.instance_name = Instance.objects.my_hostname()

awx/main/dispatch/publish.py

@@ -1,14 +1,13 @@
 import inspect
 import logging
-import sys
 import json
 import time
 from uuid import uuid4
 
-from django.conf import settings
 from django_guid import get_guid
 
 from . import pg_bus_conn
+from awx.main.utils import is_testing
 
 logger = logging.getLogger('awx.main.dispatch')
 
@@ -93,7 +92,7 @@ class task:
                 obj.update(**kw)
                 if callable(queue):
                     queue = queue()
-                if not settings.IS_TESTING(sys.argv):
+                if not is_testing():
                     with pg_bus_conn() as conn:
                         conn.notify(queue, json.dumps(obj))
                 return (obj, queue)

awx/main/scheduler/task_manager.py

@@ -39,7 +39,7 @@ from awx.main.utils import (
     ScheduleTaskManager,
     ScheduleWorkflowManager,
 )
-from awx.main.utils.common import task_manager_bulk_reschedule
+from awx.main.utils.common import task_manager_bulk_reschedule, is_testing
 from awx.main.signals import disable_activity_stream
 from awx.main.constants import ACTIVE_STATES
 from awx.main.scheduler.dependency_graph import DependencyGraph
@@ -97,7 +97,7 @@ class TaskBase:
         self.all_tasks = [t for t in qs]
 
     def record_aggregate_metrics(self, *args):
-        if not settings.IS_TESTING():
+        if not is_testing():
             # increment task_manager_schedule_calls regardless if the other
             # metrics are recorded
             s_metrics.Metrics(auto_pipe_execute=True).inc(f"{self.prefix}__schedule_calls", 1)

awx/main/utils/common.py

@@ -11,11 +11,12 @@ import os
 import subprocess
 import re
 import stat
+import sys
 import urllib.parse
 import threading
 import contextlib
 import tempfile
-from functools import reduce, wraps
+import functools
 
 # Django
 from django.core.exceptions import ObjectDoesNotExist, FieldDoesNotExist
@@ -73,6 +74,7 @@ __all__ = [
     'NullablePromptPseudoField',
     'model_instance_diff',
     'parse_yaml_or_json',
+    'is_testing',
     'RequireDebugTrueOrTest',
     'has_model_field_prefetched',
     'set_environ',
@@ -144,6 +146,19 @@ def underscore_to_camelcase(s):
     return ''.join(x.capitalize() or '_' for x in s.split('_'))
 
 
+@functools.cache
+def is_testing(argv=None):
+    '''Return True if running django or py.test unit tests.'''
+    if 'PYTEST_CURRENT_TEST' in os.environ.keys():
+        return True
+    argv = sys.argv if argv is None else argv
+    if len(argv) >= 1 and ('py.test' in argv[0] or 'py/test.py' in argv[0]):
+        return True
+    elif len(argv) >= 2 and argv[1] == 'test':
+        return True
+    return False
+
+
 class RequireDebugTrueOrTest(logging.Filter):
     """
     Logging filter to output when in DEBUG mode or running tests.
@@ -152,7 +167,7 @@ class RequireDebugTrueOrTest(logging.Filter):
     def filter(self, record):
         from django.conf import settings
 
-        return settings.DEBUG or settings.IS_TESTING()
+        return settings.DEBUG or is_testing()
 
 
 class IllegalArgumentError(ValueError):
@@ -174,7 +189,7 @@ def memoize(ttl=60, cache_key=None, track_function=False, cache=None):
     cache = cache or get_memoize_cache()
 
     def memoize_decorator(f):
-        @wraps(f)
+        @functools.wraps(f)
         def _memoizer(*args, **kwargs):
             if track_function:
                 cache_dict_key = slugify('%r %r' % (args, kwargs))
@@ -992,7 +1007,7 @@ def getattrd(obj, name, default=NoDefaultProvided):
     """
 
     try:
-        return reduce(getattr, name.split("."), obj)
+        return functools.reduce(getattr, name.split("."), obj)
     except AttributeError:
         if default != NoDefaultProvided:
             return default
@@ -1188,7 +1203,7 @@ def cleanup_new_process(func):
     Cleanup django connection, cache connection, before executing new thread or processes entry point, func.
     """
 
-    @wraps(func)
+    @functools.wraps(func)
     def wrapper_cleanup_new_process(*args, **kwargs):
         from awx.conf.settings import SettingsWrapper  # noqa
 
@@ -1202,7 +1217,7 @@ def cleanup_new_process(func):
 
 def log_excess_runtime(func_logger, cutoff=5.0):
     def log_excess_runtime_decorator(func):
-        @wraps(func)
+        @functools.wraps(func)
         def _new_func(*args, **kwargs):
             start_time = time.time()
             return_value = func(*args, **kwargs)

awx/settings/defaults.py

@@ -10,28 +10,6 @@ import socket
 from datetime import timedelta
 
 
-# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
-BASE_DIR = os.path.dirname(os.path.dirname(__file__))
-
-
-def is_testing(argv=None):
-    import sys
-
-    '''Return True if running django or py.test unit tests.'''
-    if 'PYTEST_CURRENT_TEST' in os.environ.keys():
-        return True
-    argv = sys.argv if argv is None else argv
-    if len(argv) >= 1 and ('py.test' in argv[0] or 'py/test.py' in argv[0]):
-        return True
-    elif len(argv) >= 2 and argv[1] == 'test':
-        return True
-    return False
-
-
-def IS_TESTING(argv=None):
-    return is_testing(argv)
-
-
 if "pytest" in sys.modules:
     from unittest import mock
 
@@ -40,9 +18,13 @@ if "pytest" in sys.modules:
 else:
     import ldap
 
+
 DEBUG = True
 SQL_DEBUG = DEBUG
 
+# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
+BASE_DIR = os.path.dirname(os.path.dirname(__file__))
+
 # FIXME: it would be nice to cycle back around and allow this to be
 # BigAutoField going forward, but we'd have to be explicit about our
 # existing models.