Flip default of reverse sync to True

Thow an error if shared fields are edited with local management turned off

Fix grammar error

Co-authored-by: Rick Elrod <[email protected]>

Add test and change to ValidationError

Disconnect signals to fix failure

Author: AlanCoding

ansible_base/resource_registry/utils/sync_to_resource_server.py

@@ -55,6 +55,16 @@ def sync_to_resource_server(instance, action, ansible_id=None):
         logger.info(f"Skipping sync of resource {instance} because its service_id is local")
         return
 
+    # By this point we have passed the exit conditions
+    # If an edit requires a resource sync, but we intended to not allow local edits
+    # that has led to a contradiction, normally this is enforced by the service API
+    # but for direct ORM actions we throw an error as a last-resort safety measure
+    if not getattr(settings, "ALLOW_LOCAL_RESOURCE_MANAGEMENT"):
+        raise ValidationError(
+            "Configured to resource sync, which conflicts with ALLOW_LOCAL_RESOURCE_MANAGEMENT. "
+            "Try setting environment variable ANSIBLE_REVERSE_RESOURCE_SYNC to false."
+        )
+
     user_ansible_id = None
     user = get_current_user()
     if user:

test_app/tests/resource_registry/test_utils.py

@@ -1,16 +1,17 @@
 import os
 import uuid
-from contextlib import nullcontext
+from contextlib import contextmanager, nullcontext
 from unittest import mock
 
 import pytest
 from crum import impersonate
 from django.contrib.auth.models import AnonymousUser
 from django.db import IntegrityError, connection
-from django.test.utils import CaptureQueriesContext
+from django.test.utils import CaptureQueriesContext, override_settings
 from rest_framework.exceptions import ValidationError
 
 from ansible_base.resource_registry import apps
+from ansible_base.resource_registry.models import Resource
 from ansible_base.resource_registry.signals.handlers import no_reverse_sync
 from ansible_base.resource_registry.utils.sync_to_resource_server import sync_to_resource_server
 from test_app.models import Organization
@@ -246,3 +247,69 @@ def test_should_reverse_sync(self, settings, new_settings, should_sync):
             setattr(settings, key, value)
 
         assert apps._should_reverse_sync() == should_sync
+
+
+@pytest.mark.django_db
+class TestConflitingSyncSettings:
+    CONFLICT_SETTINGS = dict(
+        ALLOW_LOCAL_RESOURCE_MANAGEMENT=True,
+        RESOURCE_SERVER_SYNC_ENABLED=True,
+        RESOURCE_SERVER={'URL': 'https://foo.invalid', 'SECRET_KEY': 'abcdefghijklmnopqrstuvwxyz'},
+    )
+
+    @contextmanager
+    def apply_settings(self):
+        with override_settings(**self.CONFLICT_SETTINGS):
+            apps.connect_resource_signals(None)
+            yield
+            apps.disconnect_resource_signals(None)
+        apps.connect_resource_signals(None)
+
+    def mark_external(self, obj):
+        "Make object appears as if managed by another system"
+        if hasattr(obj, '_skip_reverse_resource_sync'):
+            delattr(obj, '_skip_reverse_resource_sync')
+        res = Resource.get_resource_for_object(obj)
+        res.service_id = uuid.uuid4()
+        res.save()
+
+    def test_block_org_save(self, organization):
+        with self.apply_settings():
+            organization.save()  # does not error
+            self.mark_external(organization)
+            with pytest.raises(ValidationError):
+                organization.name += 'foo'
+                organization.save()
+
+    def test_block_org_deletion(self, organization):
+        with self.apply_settings():
+            organization.save()  # create resource entry
+            self.mark_external(organization)
+            with pytest.raises(ValidationError):
+                organization.delete()
+
+    def test_change_non_synced_field(self, admin_user):
+        with self.apply_settings():
+            admin_user.save()  # create resource entry
+            self.mark_external(admin_user)
+            assert admin_user.is_superuser is True
+            admin_user.is_superuser = False
+            admin_user.save()  # does not error
+
+    def test_change_user_synced_field(self, admin_user):
+        with self.apply_settings():
+            admin_user.save()  # create resource entry
+            self.mark_external(admin_user)
+            with pytest.raises(ValidationError):
+                admin_user.username = 'some_other_username'
+                admin_user.save()
+
+    def test_create_shared_resource(self):
+        with self.apply_settings():
+            with pytest.raises(ValidationError):
+                Organization.objects.create(name='new org')
+
+    def test_create_allowed_no_sync(self):
+        with self.apply_settings():
+            with no_reverse_sync():
+                Organization.objects.create(name='new org')