chore: update test for running on empty db

core/gql/gql_mutations/base_mutation.py

@@ -237,7 +237,7 @@ def _mutate(cls, user, **data):
 
     @classmethod
     def update_object(cls, user, object_to_update):
-        object_to_update.save(username=user.username)
+        object_to_update.save(user=user)
         return object_to_update
 
 

core/jwt.py

@@ -20,7 +20,6 @@ def on_token_issued(sender, request, user, **kwargs):
     if user.i_user:
         user.i_user.last_login = timezone.now()
         user.i_user.save()
-    pass
 
 
 def jwt_encode_user_key(payload, context=None):

core/jwt_authentication.py

@@ -5,6 +5,7 @@
 from graphql_jwt.exceptions import JSONWebTokenError
 from graphql_jwt.shortcuts import get_user_by_token
 from core.apps import CoreConfig
+from core.utils import set_current_user
 from django.conf import settings
 from django_ratelimit.core import is_ratelimited
 
@@ -40,7 +41,7 @@ def authenticate(self, request):
                         and user.health_facility.contract_end_date > date.today()
                     ):
                         raise exceptions.AuthenticationFailed("HF_CONTRACT_INVALID")
-
+            set_current_user(user)
             return user, None
 
     def enforce_csrf(self, request):

core/models/history_model.py

@@ -9,7 +9,7 @@
 from django.db import models
 from django.db.models import F
 from simple_history.models import HistoricalRecords
-from core.utils import CachedManager, CachedModelMixin
+from core.utils import CachedManager, CachedModelMixin, get_current_user
 
 # from core.datetimes.ad_datetime import datetime as py_datetime
 
@@ -104,15 +104,24 @@ def update(self, *args, user=None, username=None, save=True, **kwargs):
             self.save(*args, user=user, username=user, **kwargs)
         return self
 
+    def _get_user(self, user=None, username=None):
+
+        audit_user = get_current_user()
+        if audit_user:
+            return audit_user
+        elif user:
+            return user
+        elif username:
+            audit_user = User.objects.get(username=username, *User.filter_validity())
+            return audit_user
+        else:
+            raise ValidationError(
+                "Save error! Provide user or the username of the current user in `username` argument"
+            )
+
     def save(self, *args, user=None, username=None, **kwargs):
         # get the user data so as to assign later his uuid id in fields user_updated etc
-        if not user:
-            if username:
-                user = User.objects.get(username=username)
-            else:
-                raise ValidationError(
-                    "Save error! Provide user or the username of the current user in `username` argument"
-                )
+        user = self._get_user(user, username)
         now = py_datetime.now()
         # check if object has been newly created
         if self.id is None:
@@ -150,9 +159,10 @@ def save(self, *args, user=None, username=None, **kwargs):
                     raise ValidationError(
                         "Update error! You cannot update replaced entity"
                     )
-            result = super(HistoryModel, self).save(*args, **kwargs)
-            self.update_cache()
-            return result
+            errors = super(HistoryModel, self).save(*args, **kwargs)
+            if not errors:
+                self.update_cache()
+            return errors
         else:
             raise ValidationError(
                 "Record has not be updated - there are no changes in fields"
@@ -162,13 +172,7 @@ def delete_history(self):
         pass
 
     def delete(self, *args, user=None, username=None, **kwargs):
-        if not user:
-            if username:
-                user = User.objects.get(username=username)
-            else:
-                raise ValidationError(
-                    "Save error! Provide user or the username of the current user in `username` argument"
-                )
+        user = self._get_user(user, username)
         if not self.is_dirty(check_relationship=True) and not self.is_deleted:
 
             now = py_datetime.now()

core/models/openimis_graphql_test_case.py

@@ -12,7 +12,8 @@
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.sessions.backends.db import SessionStore
 from django.core.cache import cache
-
+from core.utils import clear_current_user
+from django.db import transaction
 
 logger = logging.getLogger(__name__)
 
@@ -106,13 +107,34 @@ def get_jwt(self):
 class openIMISGraphQLTestCase(GraphQLTestCase):
     GRAPHQL_URL = f"/{settings.SITE_ROOT()}graphql"
     GRAPHQL_SCHEMA = True
+    """
+    Enhanced version that wraps every test in an atomic transaction.
+    Prevents GraphQL validation errors (400) from closing the DB connection.
+    """
+    def _execute_test(self):
+        with transaction.atomic():
+            super()._execute_test()
+
+    def run(self, result=None):
+        """
+        Override run() to ensure atomic block even when pytest-django calls it.
+        """
+        with transaction.atomic():
+            super().run(result)
 
-    # client = None
     @classmethod
     def setUpClass(cls):
         # cls.client=Client(cls.schema)
+        clear_current_user()
+        cache.clear()
         super(openIMISGraphQLTestCase, cls).setUpClass()
 
+    def setUp(self):
+        # cls.client=Client(cls.schema)
+        clear_current_user()
+        cache.clear()
+        super().setUp(self)
+
     def get_mutation_result(
         self, mutation_uuid, token, internal=False, allow_exceptions=True
     ):
@@ -255,8 +277,10 @@ def send_mutation(
     # This validates the status code and if you get errors
     def build_params(self, params):
         def wrap_arg(v):
+            if isinstance(v, uuid.UUID):
+                return f'"{str(v).lower()}"'
             if isinstance(v, str):
-                return f'"{v}"'
+                return f'"{str(v)}"'
             if isinstance(v, list):
                 return f"[{','.join([str(wrap_arg(vv)) for vv in v])}]"
             if isinstance(v, dict):
@@ -274,6 +298,7 @@ def wrap_arg(v):
         ]
         return ", ".join(params_as_args)
 
-    def tearDwon(self):
+    def tearDown(self):
         cache.clear()
-        super().tearDwon()
+        clear_current_user()
+        super().tearDown()

core/services/utils/serviceUtils.py

@@ -5,11 +5,14 @@
 from django.contrib.contenttypes.models import ContentType
 from django.core.serializers.json import DjangoJSONEncoder
 from django.forms.models import model_to_dict
+from core.utils import get_current_user
 
 
 def check_authentication(function):
     def wrapper(self, *args, **kwargs):
-        if type(self.user) is AnonymousUser or not self.user.id:
+        if not self.user:
+            self.user = get_current_user()
+        if type(self.user) is AnonymousUser or (not self.user and not self.user.id):
             return {
                 "success": False,
                 "message": "Authentication required",

core/tasks.py

@@ -4,6 +4,7 @@
 
 from celery import shared_task
 from core.models import MutationLog, Language
+from core.utils import set_current_user
 from django.utils import translation
 
 logger = logging.getLogger(__name__)
@@ -22,6 +23,9 @@ def openimis_mutation_async(mutation_id, module, class_name):
     mutation = None
     try:
         mutation = MutationLog.objects.get(id=mutation_id)
+        # Set the current user for audit logging
+        if mutation.user:
+            set_current_user(mutation.user)
         # __import__ needs to import the module with .schema to force .schema to load, then .schema.TheRealMutation
         mutation_class = getattr(__import__(f"{module}.schema").schema, class_name)