From 345c36e9a2b3f49976eaa6c7f40f6e726be89e70 Mon Sep 17 00:00:00 2001 From: AlanCoding Date: Tue, 24 Sep 2024 14:26:29 -0400 Subject: [PATCH] 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 Add test and change to ValidationError Disconnect signals to fix failure --- .../utils/sync_to_resource_server.py | 10 +++ .../tests/resource_registry/test_utils.py | 71 ++++++++++++++++++- 2 files changed, 79 insertions(+), 2 deletions(-) diff --git a/ansible_base/resource_registry/utils/sync_to_resource_server.py b/ansible_base/resource_registry/utils/sync_to_resource_server.py index c8c837aab..9181a461e 100644 --- a/ansible_base/resource_registry/utils/sync_to_resource_server.py +++ b/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: diff --git a/test_app/tests/resource_registry/test_utils.py b/test_app/tests/resource_registry/test_utils.py index 9e94d4b90..4098d1354 100644 --- a/test_app/tests/resource_registry/test_utils.py +++ b/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')